From c8ed7f170dc6c2ec093e3ff80f338671e4d0ff0e Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Sun, 8 Nov 2026 08:19:15 +0100 Subject: [PATCH 1/2] feat(control): a journal of every change in the answer to "Are we in control?" Core samples each controlled device's control feedback, each source's freshness, whether the site may dispatch and the mode every 5 s. A changed verdict closes the subject's span and opens the next, with the power asked for and delivered and the verdict as served; a change to an info verdict must hold 15 s, a warning starts a span at once. Scored plan slots are kept with planned and delivered energy. GET /api/control/journal and the agent read events.get answer any window. Failed writes stay in memory for the next flush; notification_log rows the database was too busy for are retried in the background. Co-Authored-By: Claude Opus 5.5 --- .changeset/control-journal.md | 13 + docs/agents.md | 16 +- docs/roadmap.md | 2 +- go/cmd/ftw/app_link.go | 3 + go/cmd/ftw/control_journal.go | 109 +++++++ go/cmd/ftw/control_journal_test.go | 90 ++++++ go/cmd/ftw/main.go | 14 +- go/cmd/ftw/notification_log.go | 31 ++ go/internal/api/api.go | 1 + go/internal/api/api_control_journal.go | 216 ++++++++++++++ go/internal/api/api_control_journal_test.go | 126 ++++++++ go/internal/appproto/agents.go | 4 +- go/internal/appproto/agents_test.go | 6 +- go/internal/appproto/events.go | 171 +++++++++++ go/internal/appproto/events_test.go | 86 ++++++ go/internal/appproto/handler.go | 4 +- go/internal/control/dispatch.go | 100 ++++--- go/internal/control/slot_scored_test.go | 39 +++ go/internal/journal/journal.go | 309 ++++++++++++++++++++ go/internal/journal/journal_test.go | 199 +++++++++++++ go/internal/journal/slot.go | 32 ++ go/internal/state/control_journal.go | 284 ++++++++++++++++++ go/internal/state/control_journal_test.go | 115 ++++++++ go/internal/state/store.go | 28 ++ go/protocol/agents.go | 71 ++++- 25 files changed, 2014 insertions(+), 55 deletions(-) create mode 100644 .changeset/control-journal.md create mode 100644 go/cmd/ftw/control_journal.go create mode 100644 go/cmd/ftw/control_journal_test.go create mode 100644 go/cmd/ftw/notification_log.go create mode 100644 go/internal/api/api_control_journal.go create mode 100644 go/internal/api/api_control_journal_test.go create mode 100644 go/internal/appproto/events.go create mode 100644 go/internal/appproto/events_test.go create mode 100644 go/internal/control/slot_scored_test.go create mode 100644 go/internal/journal/journal.go create mode 100644 go/internal/journal/journal_test.go create mode 100644 go/internal/journal/slot.go create mode 100644 go/internal/state/control_journal.go create mode 100644 go/internal/state/control_journal_test.go diff --git a/.changeset/control-journal.md b/.changeset/control-journal.md new file mode 100644 index 000000000..d77fda361 --- /dev/null +++ b/.changeset/control-journal.md @@ -0,0 +1,13 @@ +--- +"ftw": minor +--- + +The box keeps a control journal. Every few seconds it records the answer to +"Are we in control?" for each controlled device, each source, whether the site +may dispatch and the operating mode; a change starts a new stretch, with the +power Core asked for and the power delivered while it held. Each scored plan +slot is kept too, with its planned and delivered energy. Read it over any +window with GET /api/control/journal, and as an agent with events.get, so the +owner, an assistant or support can learn afterwards why the app warned. A +notification the database was too busy to record is now written once it is +free. diff --git a/docs/agents.md b/docs/agents.md index dfeb0e005..e120d884f 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -384,7 +384,7 @@ The registry's `reads` table names each read and the scope it needs. | `hist.query` | As for a phone | `hist.chunk`, `hist.end` | | `health.get` | `{}` | `health {version, uptimeS, clock, drivers: {ok, degraded, faulted, offline}, meter: {srcState, ageMs}, dispatchBlockedBy}` | | `drivers.get` | `{}` | `drivers {drivers: [{src, kind, state, ageMs}]}` | -| `events.get` | `{n, beforeMs, afterMs}`, `n` at most 50 | `events {events, more}`, each FTW Cloud's `event` body with agent ids | +| `events.get` | `{n, beforeMs, afterMs, src, minSeverity}`, `n` at most 50 | `events {spans, notices, events, more}`: the control journal over the window, newest first (see below) | | `logs.get` | `{n, level, beforeMs, afterMs}`; `n` at most 200, `level` `WARN` by default | `logs {lines: [{ts, level, src, msg}], dropped, more}` | | `config.get` | `{section}` | `config {sections}` without a section; `config {section, doc}` with one | | `diag.run` | `{check, src}`; `check` is `clock`, `link`, `meter`, `planner` or `driver` | `diag {check, ok, facts}` | @@ -407,8 +407,18 @@ The registry's `reads` table names each read and the scope it needs. source that blocks dispatch goes by its agent id. Planner solve times, fallbacks, refusals and plan against actual join it when the box keeps them; a field the box cannot fill is absent. -- The box answers `events.get` and `diag.run` with `E_UNKNOWN_OP` until it - builds them. `cars.get` carries no name, maker, VIN, identifier or source, +- `events.get` reads the box's control journal: each stretch in which a + device function, a source, the site's dispatch or the mode held one + verdict, each scored plan slot, and the notifications the box sent, over + `afterMs`..`beforeMs` (the last 24 hours by default). A span gives `kind` + (battery, ev, pv, v2x_charger, device, source, dispatch, mode, slot), + `src` by agent id, Core's `status`, `reason` and `severity` as in + `control_feedback`, `startMs` and `endMs` (0 while it holds), `samples`, + `requestedW` and `actualW` as min, max and mean, and `last`: the numbers + of the latest verdict by field name. Text leaves only as Core's own codes, + never names. `minSeverity: "warning"` keeps warnings and alarms; `src` + keeps one source. `events` stays empty; it is kept for older readers. +- The box answers `diag.run` with `E_UNKNOWN_OP` until it builds it. `cars.get` carries no name, maker, VIN, identifier or source, and the redactor takes those out of `logs.get` and `config.get`; a box without the car store answers it `E_UNAVAILABLE`. `plan.get` names each charger and its car by agent id. diff --git a/docs/roadmap.md b/docs/roadmap.md index 2ce58eed6..84a831456 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -85,7 +85,7 @@ or reopen closed issues. |---|---|---| | Simple setup and first-day value | Discover mixed equipment. Confirm the main fuse and site meter. Read battery capacity where possible and ask for kWh when needed. Power settings and solar kWp are optional where safe device information and learning allow. Provide useful initial load and PV forecasts. | A fresh install reaches useful automatic operation without panel drawings or expert settings. Missing data and wrong start estimates have tested behaviour. Record device identity, known limits and uncertainty. Validate on named hardware combinations as well as simulators. | | A clear commissioning result | Check commands and measured response within known limits. Distinguish working telemetry from working control. Exclude failed control from both the plan and dispatch. | Show request, device response, measured effect and timing. Cover delayed response, refusal, disconnect, stale site data and recovery. A short commissioning test does not claim full hardware qualification. | -| Verified control: “Are we in control?” | Keep the fast local feel. Make request, acceptance, command, response, physical effect and freshness visible in the normal Flow experience. | Browser review on desktop and mobile, timing measurements on a target box, and traces with different sampling rates. A successful call, echoed setpoint or old reading never appears as a verified physical result. Test accepted, measured and confirmed evidence per device, and the status each one shows. Confirm one device while another is offline; keep unavailable background flows in the residual. Show overview status, alarm on lost measured proof after the response wait, and clear it on measured recovery. Include direction, tolerance, response delay, simultaneous load/solar changes, shared measurement sources, overridden setpoints, device limits and recovery. A reduced request remains visibly limited even when the device follows the reduced command. Examine existing Live and Flow views before deciding their final layout. | +| Verified control: “Are we in control?” | Keep the fast local feel. Make request, acceptance, command, response, physical effect and freshness visible in the normal Flow experience. Keep each change of the answer on the box, with the power asked for and delivered, so the owner, an agent or support can learn afterwards why the app warned. | Browser review on desktop and mobile, timing measurements on a target box, and traces with different sampling rates. A successful call, echoed setpoint or old reading never appears as a verified physical result. Test accepted, measured and confirmed evidence per device, and the status each one shows. Confirm one device while another is offline; keep unavailable background flows in the residual. Show overview status, alarm on lost measured proof after the response wait, and clear it on measured recovery. Include direction, tolerance, response delay, simultaneous load/solar changes, shared measurement sources, overridden setpoints, device limits and recovery. A reduced request remains visibly limited even when the device follows the reduced command. Examine existing Live and Flow views before deciding their final layout. | | Good automatic planning | Use site physics and charge/discharge efficiency. Default wear cost is zero; users may opt in. Keep hard SoC limits separate from forecast-based caution and explicit backup needs. | Cold-start and learned forecasts, stale inputs, multiple assets and unavailable optimizer paths have tests. Backtests use held-out periods and report uncertainty. Defaults and persisted settings agree across UI, Core and worker. Site runs establish practical benefit. | | Reliable daily charging | Persistent weekday target and deadline. Offline-car estimates. Direct SoC slider after connection, no extra save, prompt replanning, one-action Charge now and notifications when action is needed. | Test from app intent to charger/car response and delivered energy, including missed-goal risk, absent vehicle cloud, unknown SoC, reconnect and restart. Test notifications with the app closed. Confirm physical charging separately from simulation. | | Useful analysis and fair savings | Keep enough provenance to explain plans and outcomes. Main savings target compares with ordinary self-consumption on the same installation. | Actual cost reconciles with measured import/export and prices. Specify EV behaviour, initial and final stored-energy accounting, efficiency and coverage. Show missing and negative results. Label the current no-PV/no-battery comparison as total site value until replacement is verified. | diff --git a/go/cmd/ftw/app_link.go b/go/cmd/ftw/app_link.go index 61e3cf1d6..31e941ea6 100644 --- a/go/cmd/ftw/app_link.go +++ b/go/cmd/ftw/app_link.go @@ -797,8 +797,10 @@ func startAppLink( agents.access = &appuplink.AgentAccess{Enroll: enroll, Clock: agentClock, Push: agents.push, Holds: endAgentHolds(loadpoints)} agents.state = appproto.NewAgentState() + var journalSrc appproto.JournalSource if st != nil { agents.audit = newStateAudit(ctx, st) + journalSrc = stateJournal{st: st} } caps = append(caps, appproto.CapAccessAgents, appproto.CapAccessInvites) // One view for every agent session, so a charger's plug-in count is @@ -838,6 +840,7 @@ func startAppLink( Agents: agents.state, Audit: agents.audit, Logs: ringLogs{ring: agents.logs}, + Journal: journalSrc, Redactor: agents.redactor, Views: views, AgentKey: agents.key, diff --git a/go/cmd/ftw/control_journal.go b/go/cmd/ftw/control_journal.go new file mode 100644 index 000000000..c7057d37f --- /dev/null +++ b/go/cmd/ftw/control_journal.go @@ -0,0 +1,109 @@ +package main + +import ( + "context" + "log/slog" + "time" + + "github.com/srcfl/ftw/go/internal/appproto" + "github.com/srcfl/ftw/go/internal/journal" + "github.com/srcfl/ftw/go/internal/state" + "github.com/srcfl/ftw/go/protocol" +) + +const ( + // controlJournalEvery is how often the journal samples the box's + // answer to "Are we in control?". + controlJournalEvery = 5 * time.Second + // controlJournalKeep is how far back the journal answers. + controlJournalKeep = 366 * 24 * time.Hour +) + +// newControlJournal ends the spans a previous run left open, at the last +// moment that run saw them hold, and returns the recorder for this one. +func newControlJournal(st *state.Store) *journal.Recorder { + if n, err := st.CloseOpenControlSpans(context.Background()); err != nil { + slog.Warn("control journal: could not close spans the last run left open", "err", err) + } else if n > 0 { + slog.Info("control journal: closed spans the last run left open", "spans", n) + } + return journal.New(st) +} + +// runControlJournal samples, writes and prunes until ctx ends. +func runControlJournal(ctx context.Context, rec *journal.Recorder, sample func(time.Time) []journal.Verdict, st *state.Store) { + t := time.NewTicker(controlJournalEvery) + defer t.Stop() + var prunedAt time.Time + for { + select { + case <-ctx.Done(): + return + case now := <-t.C: + rec.Observe(now, sample(now)) + flushCtx, cancel := context.WithTimeout(ctx, controlJournalEvery) + rec.Flush(flushCtx, now) + cancel() + if now.Sub(prunedAt) < 24*time.Hour { + continue + } + prunedAt = now + if n, err := st.PruneControlSpans(ctx, now.Add(-controlJournalKeep).UnixMilli()); err != nil { + slog.Warn("control journal: prune failed", "err", err) + } else if n > 0 { + slog.Info("control journal: pruned old spans", "spans", n) + } + } + } +} + +// stopControlJournal ends every open span now and writes them, so a clean +// stop leaves no span running through the time the box was down. +func stopControlJournal(rec *journal.Recorder) { + now := time.Now() + rec.CloseAll(now) + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + rec.Flush(ctx, now) +} + +// stateJournal serves events.get from the control journal and the +// notification history. +type stateJournal struct{ st *state.Store } + +func (j stateJournal) Journal(ctx context.Context, q appproto.JournalQuery) ([]appproto.JournalSpan, []appproto.JournalNotice, bool, error) { + spans, more, err := j.st.ControlSpans(ctx, state.ControlSpanQuery{ + SinceMs: q.SinceMs, UntilMs: q.UntilMs, Driver: q.Driver, Severities: q.Severities, Limit: q.Limit, + }) + if err != nil { + return nil, nil, false, err + } + notices, noticesMore, err := j.st.NotificationsBetween(ctx, q.SinceMs, q.UntilMs, q.Limit) + if err != nil { + return nil, nil, false, err + } + out := make([]appproto.JournalSpan, 0, len(spans)) + for _, sp := range spans { + out = append(out, appproto.JournalSpan{ + Driver: sp.Driver, Kind: sp.Kind, Status: sp.Status, Reason: sp.Reason, Severity: sp.Severity, Mode: sp.Mode, + StartMs: sp.StartMs, EndMs: sp.EndMs, Samples: sp.Samples, + Requested: powerRange(sp.Requested), Actual: powerRange(sp.Actual), Last: sp.Last, + }) + } + var ns []appproto.JournalNotice + for _, n := range notices { + if q.Driver != "" && n.Driver != q.Driver { + continue + } + ns = append(ns, appproto.JournalNotice{TsMs: n.TsMs, Type: n.EventType, Driver: n.Driver, Status: n.Status, Priority: n.Priority}) + } + return out, ns, more || noticesMore, nil +} + +func powerRange(p state.PowerStats) *protocol.PowerRange { + mean, ok := p.Mean() + if !ok { + return nil + } + return &protocol.PowerRange{Min: p.Min, Max: p.Max, Mean: mean} +} diff --git a/go/cmd/ftw/control_journal_test.go b/go/cmd/ftw/control_journal_test.go new file mode 100644 index 000000000..d77f85f7d --- /dev/null +++ b/go/cmd/ftw/control_journal_test.go @@ -0,0 +1,90 @@ +package main + +import ( + "context" + "errors" + "path/filepath" + "sync/atomic" + "testing" + "time" + + "github.com/srcfl/ftw/go/internal/appproto" + "github.com/srcfl/ftw/go/internal/journal" + "github.com/srcfl/ftw/go/internal/state" +) + +func openJournalStore(t *testing.T) *state.Store { + t.Helper() + st, err := state.Open(filepath.Join(t.TempDir(), "state.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = st.Close() }) + return st +} + +// A stop ends the spans now and writes them; the next start finds none +// left open. +func TestStopControlJournalClosesAndWrites(t *testing.T) { + st := openJournalStore(t) + rec := newControlJournal(st) + now := time.Now() + rec.Observe(now, []journal.Verdict{{Subject: "site:dispatch", Kind: "dispatch", Status: "free", Severity: "info"}}) + stopControlJournal(rec) + spans, _, err := st.ControlSpans(context.Background(), state.ControlSpanQuery{SinceMs: 0, UntilMs: now.Add(time.Minute).UnixMilli()}) + if err != nil || len(spans) != 1 || spans[0].EndMs == 0 { + t.Fatalf("spans %+v err %v", spans, err) + } + if n, _ := st.CloseOpenControlSpans(context.Background()); n != 0 { + t.Fatalf("%d spans left open after a clean stop", n) + } +} + +// events.get for one source reads that source's spans and notices only. +func TestStateJournalKeepsOneSource(t *testing.T) { + st := openJournalStore(t) + ctx := context.Background() + spans := []*state.ControlSpan{ + {Subject: "battery:sungrow", Driver: "sungrow", Kind: "battery", Status: "limited", Severity: "warning", StartMs: 100, UpdatedMs: 200}, + {Subject: "ev:easee", Driver: "easee", Kind: "ev", Status: "following", Severity: "info", StartMs: 100, UpdatedMs: 200}, + } + spans[0].Actual.Add(ptrW(1250)) + if err := st.SaveControlSpans(ctx, spans); err != nil { + t.Fatal(err) + } + for _, d := range []string{"sungrow", "easee"} { + if err := st.RecordNotification(state.NotificationEntry{TsMs: 150, EventType: "control.not_following", Driver: d, Status: "sent"}); err != nil { + t.Fatal(err) + } + } + got, notices, more, err := stateJournal{st: st}.Journal(ctx, appproto.JournalQuery{SinceMs: 0, UntilMs: 1_000, Driver: "sungrow", Limit: 50}) + if err != nil || more || len(got) != 1 || got[0].Kind != "battery" || got[0].Actual == nil || got[0].Actual.Mean != 1250 || got[0].Requested != nil { + t.Fatalf("spans %+v more %v err %v", got, more, err) + } + if len(notices) != 1 || notices[0].Driver != "sungrow" { + t.Fatalf("notices %+v", notices) + } +} + +// A locked database delays a notification row; it does not lose it. +func TestRecordWithRetryWritesAfterAFailure(t *testing.T) { + saved := notificationLogRetries + notificationLogRetries = []time.Duration{time.Millisecond, time.Millisecond, time.Millisecond} + t.Cleanup(func() { notificationLogRetries = saved }) + var tries atomic.Int32 + done := make(chan struct{}) + recordWithRetry("notification_log", func() error { + if tries.Add(1) < 3 { + return errors.New("database is locked (5) (SQLITE_BUSY)") + } + close(done) + return nil + }) + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatalf("not written after %d tries", tries.Load()) + } +} + +func ptrW(v float64) *float64 { return &v } diff --git a/go/cmd/ftw/main.go b/go/cmd/ftw/main.go index 27bc7a544..77f505ebb 100644 --- a/go/cmd/ftw/main.go +++ b/go/cmd/ftw/main.go @@ -57,6 +57,7 @@ import ( "github.com/srcfl/ftw/go/internal/ftwcli" "github.com/srcfl/ftw/go/internal/gatewayidentity" "github.com/srcfl/ftw/go/internal/ha" + "github.com/srcfl/ftw/go/internal/journal" "github.com/srcfl/ftw/go/internal/loadmodel" "github.com/srcfl/ftw/go/internal/loadpoint" "github.com/srcfl/ftw/go/internal/localzone" @@ -618,6 +619,11 @@ func main() { // Restore durable modes and targets. Forced Charge ends at restart; // the planner makes a fresh plan instead of resuming that override. restoreStoredMode(ctrl, st, cfg) + // The control journal keeps each change in the box's answer to "Are we + // in control?", and each scored plan slot, so a warning can be explained + // after it has passed (GET /api/control/journal, events.get). + ctrlJournal := newControlJournal(st) + ctrl.SlotScored = func(d control.SlotDelivery) { ctrlJournal.Record(journal.SlotSpan(d)) } storedTrust, _ := st.LoadConfig(config.StateKeyForecastTrust) storedExport, _ := st.LoadConfig(config.StateKeyBatteryExport) storedSafetyK, _ := st.LoadConfig(config.StateKeySafetyK) @@ -2725,6 +2731,7 @@ func main() { // The control.not_following rule reads the box's own answer to "Are we // in control?", the one GET /api/status serves. notifSvc.SetControlReader(srv.ControlStates) + go runControlJournal(ctx, ctrlJournal, srv.JournalVerdicts, st) // FuseReader: on each HealthTick the fuse_over_limit rule reads // the site meter's live per-phase currents from telemetry and // compares against cfg.Fuse.MaxAmps. Closes over cfg + cfgMu so @@ -2759,7 +2766,7 @@ func main() { if !ok { return } - if err := st.RecordNotification(state.NotificationEntry{ + entry := state.NotificationEntry{ TsMs: ev.Time.UnixMilli(), EventType: ev.EventType, Driver: ev.Driver, @@ -2768,9 +2775,8 @@ func main() { Priority: ev.Priority, Status: ev.Status, Error: ev.Error, - }); err != nil { - slog.Warn("notification_log: record failed", "err", err) } + recordWithRetry("notification_log", func() error { return st.RecordNotification(entry) }) }) // Web push runs beside whatever provider the config names. The provider // itself was built before the uplink (the dead man's rows carry its @@ -2933,6 +2939,7 @@ func main() { slog.Info("shutting down") cloudLink.Bye(cloudwire.ByeShutdown, cloudByeWait) markNativeCleanExit(nativeRoot) + stopControlJournal(ctrlJournal) flushHistoryOnStop(st) if err := st.RecordEvent("shutdown"); err != nil { slog.Warn("failed to persist shutdown event", "err", err) @@ -2948,6 +2955,7 @@ func main() { // back "box out of reach" until the box is back. cloudLink.Bye(cloudwire.ByeUpdate, cloudByeWait) markNativeCleanExit(nativeRoot) + stopControlJournal(ctrlJournal) flushHistoryOnStop(st) if err := st.RecordEvent("restart"); err != nil { slog.Warn("failed to persist restart event", "err", err) diff --git a/go/cmd/ftw/notification_log.go b/go/cmd/ftw/notification_log.go new file mode 100644 index 000000000..47578b4fd --- /dev/null +++ b/go/cmd/ftw/notification_log.go @@ -0,0 +1,31 @@ +package main + +import ( + "log/slog" + "time" +) + +// notificationLogRetries spaces the tries after a failed notification_log +// write: a long writer, such as the daily snapshot, can hold the database +// for longer than the busy timeout, and the row is the only record of what +// the owner was told. +var notificationLogRetries = []time.Duration{time.Second, 4 * time.Second, 15 * time.Second, time.Minute, 5 * time.Minute} + +// recordWithRetry writes now and, if that fails, tries again in the +// background so the event bus never waits on the database. +func recordWithRetry(what string, write func() error) { + err := write() + if err == nil { + return + } + go func() { + for i, wait := range notificationLogRetries { + time.Sleep(wait) + if err = write(); err == nil { + slog.Info(what+": recorded after a retry", "tries", i+2) + return + } + } + slog.Warn(what+": record failed after retries", "tries", len(notificationLogRetries)+1, "err", err) + }() +} diff --git a/go/internal/api/api.go b/go/internal/api/api.go index 429a795c8..a267b2673 100644 --- a/go/internal/api/api.go +++ b/go/internal/api/api.go @@ -506,6 +506,7 @@ func (s *Server) routes() { s.handle("GET /api/notifications/status", Read, s.handleNotificationsStatus) s.handle("GET /api/notifications/defaults", Read, s.handleNotificationsDefaults) s.handle("GET /api/notifications/history", Read, s.handleNotificationsHistory) + s.handle("GET /api/control/journal", Read, s.handleControlJournal) s.handle("POST /api/notifications/test", Configure, s.handleNotificationsTest) s.handle("GET /api/notifications/vapid", Read, s.handleNotificationsVAPID) s.handle("POST /api/notifications/subscriptions", Configure, s.handlePushSubscribe) diff --git a/go/internal/api/api_control_journal.go b/go/internal/api/api_control_journal.go new file mode 100644 index 000000000..4eaae6ecd --- /dev/null +++ b/go/internal/api/api_control_journal.go @@ -0,0 +1,216 @@ +package api + +import ( + "encoding/json" + "net/http" + "strconv" + "strings" + "time" + + "github.com/srcfl/ftw/go/internal/journal" + "github.com/srcfl/ftw/go/internal/state" + "github.com/srcfl/ftw/go/internal/telemetry" +) + +// JournalVerdicts is what the control journal samples: each device +// function's control feedback as GET /api/status serves it, each source's +// freshness, whether the site may dispatch, and the operating mode. +func (s *Server) JournalVerdicts(now time.Time) []journal.Verdict { + var out []journal.Verdict + for _, f := range s.controlFeedback(now) { + detail, _ := json.Marshal(f) + out = append(out, journal.Verdict{ + Subject: f.Kind + ":" + f.Driver, Driver: f.Driver, Kind: f.Kind, + Status: f.Status, Reason: f.Reason, Severity: f.Severity, Mode: f.Mode, + RequestedW: f.RequestedW, ActualW: f.ActualW, Detail: detail, + }) + } + if s.deps.SiteDispatchBlocked != nil { + v := journal.Verdict{Subject: "site:dispatch", Kind: "dispatch", Status: "free", Severity: "info"} + if reason := s.deps.SiteDispatchBlocked(); reason != "" { + v.Status, v.Reason, v.Severity = "blocked", reason, "warning" + } + out = append(out, v) + } + if s.deps.Ctrl != nil && s.deps.CtrlMu != nil { + s.deps.CtrlMu.Lock() + mode := string(s.deps.Ctrl.Mode) + s.deps.CtrlMu.Unlock() + out = append(out, journal.Verdict{Subject: "site:mode", Kind: "mode", Status: mode, Severity: "info", Mode: mode}) + } + if s.deps.Tel != nil { + disabled := map[string]bool{} + if s.deps.Cfg != nil && s.deps.CfgMu != nil { + s.deps.CfgMu.RLock() + for _, d := range s.deps.Cfg.Drivers { + disabled[d.Name] = d.Disabled + } + s.deps.CfgMu.RUnlock() + } + for name, h := range s.deps.Tel.AllHealth() { + if disabled[name] { + continue + } + out = append(out, sourceVerdict(name, h)) + } + } + return out +} + +// sourceVerdict is a driver's freshness as the journal keeps it. The error +// text changes from poll to poll, so it is detail, not the reason. +func sourceVerdict(name string, h telemetry.DriverHealth) journal.Verdict { + v := journal.Verdict{Subject: "source:" + name, Driver: name, Kind: "source", Status: "live", Severity: "info"} + switch { + case h.DeviceFault: + v.Status, v.Reason, v.Severity = "fault", "device_fault", "warning" + case h.Status == telemetry.StatusOffline: + v.Status, v.Reason, v.Severity = "offline", "telemetry_stale", "warning" + case h.Status == telemetry.StatusDegraded: + v.Status, v.Reason = "lagging", "read_failed" + } + detail := map[string]any{"status": h.Status.String(), "consecutive_errors": h.ConsecutiveErrors} + if h.LastSuccess != nil { + detail["last_success_ms"] = h.LastSuccess.UnixMilli() + } + if h.LastError != "" { + detail["last_error"] = h.LastError + } + if h.DeviceFaultReason != "" { + detail["device_fault_reason"] = h.DeviceFaultReason + } + v.Detail, _ = json.Marshal(detail) + return v +} + +// journalSpan is a span as GET /api/control/journal serves it. +type journalSpan struct { + ID int64 `json:"id"` + Subject string `json:"subject"` + Driver string `json:"driver,omitempty"` + Kind string `json:"kind"` + Status string `json:"status"` + Reason string `json:"reason,omitempty"` + Severity string `json:"severity,omitempty"` + Mode string `json:"mode,omitempty"` + StartMs int64 `json:"start_ms"` + EndMs *int64 `json:"end_ms"` + UpdatedMs int64 `json:"updated_ms"` + DurationS float64 `json:"duration_s"` + Samples int64 `json:"samples"` + RequestedW *powerSummary `json:"requested_w"` + ActualW *powerSummary `json:"actual_w"` + First json.RawMessage `json:"first,omitempty"` + Last json.RawMessage `json:"last,omitempty"` +} + +type powerSummary struct { + Min float64 `json:"min"` + Max float64 `json:"max"` + Mean float64 `json:"mean"` + N int64 `json:"n"` +} + +func summarise(p state.PowerStats) *powerSummary { + mean, ok := p.Mean() + if !ok { + return nil + } + return &powerSummary{Min: p.Min, Max: p.Max, Mean: mean, N: p.N} +} + +func toJournalSpan(sp state.ControlSpan, nowMs int64) journalSpan { + out := journalSpan{ + ID: sp.ID, Subject: sp.Subject, Driver: sp.Driver, Kind: sp.Kind, Status: sp.Status, + Reason: sp.Reason, Severity: sp.Severity, Mode: sp.Mode, StartMs: sp.StartMs, UpdatedMs: sp.UpdatedMs, + Samples: sp.Samples, RequestedW: summarise(sp.Requested), ActualW: summarise(sp.Actual), + First: sp.First, Last: sp.Last, + } + end := nowMs + if sp.EndMs != 0 { + e := sp.EndMs + out.EndMs, end = &e, e + } + out.DurationS = float64(end-sp.StartMs) / 1000 + return out +} + +// journalSeverities turns ?severity= into the severities to keep: a single +// level keeps it and those above it; a comma list keeps exactly those. +func journalSeverities(q string) []string { + if q == "" { + return nil + } + if strings.Contains(q, ",") { + return strings.Split(q, ",") + } + switch q { + case "info": + return nil + case "warning": + return []string{"warning", "alarm"} + default: + return []string{q} + } +} + +// GET /api/control/journal answers "what happened" over a window: each +// stretch a device function, a source, the site's dispatch or the mode held +// one verdict, with the power asked for and delivered, each scored plan +// slot, and the notifications the box sent. since and until are Unix ms +// (the last 24 hours by default); subject keeps subjects with that prefix, +// e.g. battery: or source:easee; severity=warning keeps warnings and alarms. +func (s *Server) handleControlJournal(w http.ResponseWriter, r *http.Request) { + if s.deps.State == nil { + writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "the control journal needs the state store"}) + return + } + now := time.Now() + q := r.URL.Query() + untilMs, sinceMs := now.UnixMilli(), now.Add(-24*time.Hour).UnixMilli() + for key, dst := range map[string]*int64{"since": &sinceMs, "until": &untilMs} { + if v := q.Get(key); v != "" { + n, err := strconv.ParseInt(v, 10, 64) + if err != nil || n < 0 { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": key + " must be Unix ms"}) + return + } + *dst = n + } + } + if untilMs < sinceMs { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "until must be at or after since"}) + return + } + limit := 200 + if v := q.Get("limit"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + limit = min(n, 1000) + } + } + spans, more, err := s.deps.State.ControlSpans(r.Context(), state.ControlSpanQuery{ + SinceMs: sinceMs, UntilMs: untilMs, SubjectPrefix: q.Get("subject"), Driver: q.Get("driver"), + Severities: journalSeverities(q.Get("severity")), Limit: limit, + }) + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()}) + return + } + notices, noticesMore, err := s.deps.State.NotificationsBetween(r.Context(), sinceMs, untilMs, limit) + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()}) + return + } + out := struct { + SinceMs int64 `json:"since_ms"` + UntilMs int64 `json:"until_ms"` + Spans []journalSpan `json:"spans"` + More bool `json:"more"` + Notifications []state.NotificationEntry `json:"notifications"` + NoticesMore bool `json:"notifications_more"` + }{SinceMs: sinceMs, UntilMs: untilMs, Spans: []journalSpan{}, More: more, Notifications: notices, NoticesMore: noticesMore} + for _, sp := range spans { + out.Spans = append(out.Spans, toJournalSpan(sp, now.UnixMilli())) + } + writeJSON(w, http.StatusOK, out) +} diff --git a/go/internal/api/api_control_journal_test.go b/go/internal/api/api_control_journal_test.go new file mode 100644 index 000000000..d0d7d5e91 --- /dev/null +++ b/go/internal/api/api_control_journal_test.go @@ -0,0 +1,126 @@ +package api + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "path/filepath" + "strconv" + "sync" + "testing" + "time" + + "github.com/srcfl/ftw/go/internal/control" + "github.com/srcfl/ftw/go/internal/journal" + "github.com/srcfl/ftw/go/internal/state" + "github.com/srcfl/ftw/go/internal/telemetry" +) + +// The journal samples what the app shows: each controlled device's verdict, +// whether the site may dispatch (the app's "FTW is not in control"), the +// mode, and each source that has gone quiet (the app's "is not reporting"). +func TestJournalVerdictsCoverDevicesSiteAndSources(t *testing.T) { + tel := telemetry.NewStore() + soc := 0.2 + tel.Update("sungrow", telemetry.DerMeter, 4240, nil, nil) + tel.Update("sungrow", telemetry.DerBattery, 1250, &soc, nil) + tel.RecordDriverSuccess("sungrow") + tel.RecordDriverSuccess("myuplink") + tel.DriverHealthMut("myuplink").Status = telemetry.StatusOffline + ctrl := control.NewState(0, 42, "sungrow") + ctrl.Mode = control.ModePlannerArbitrage + ctrl.LastTargets = []control.DispatchTarget{{Driver: "sungrow", TargetW: 4250}} + srv := New(&Deps{Tel: tel, Ctrl: ctrl, CtrlMu: &sync.Mutex{}, SiteDispatchBlocked: func() string { return "site_meter_stale" }}) + + got := map[string]journal.Verdict{} + for _, v := range srv.JournalVerdicts(time.Now()) { + got[v.Subject] = v + } + if v, ok := got["battery:sungrow"]; !ok || v.Status == "" || len(v.Detail) == 0 { + t.Fatalf("battery verdict: %+v", v) + } + if v := got["site:dispatch"]; v.Status != "blocked" || v.Reason != "site_meter_stale" || v.Severity != "warning" { + t.Fatalf("dispatch verdict: %+v", v) + } + if v := got["site:mode"]; v.Status != "planner_arbitrage" { + t.Fatalf("mode verdict: %+v", v) + } + if v := got["source:myuplink"]; v.Status != "offline" || v.Severity != "warning" { + t.Fatalf("quiet source: %+v", v) + } + if v := got["source:sungrow"]; v.Status != "live" || v.Severity != "info" { + t.Fatalf("live source: %+v", v) + } +} + +func TestControlJournalAnswersAWindow(t *testing.T) { + st, err := state.Open(filepath.Join(t.TempDir(), "state.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = st.Close() }) + t0 := time.Date(2026, 11, 8, 6, 0, 0, 0, time.Local).UnixMilli() + minute := int64(60_000) + limited := &state.ControlSpan{Subject: "battery:sungrow", Driver: "sungrow", Kind: "battery", Status: "limited", + Reason: "power_below_target", Severity: "warning", StartMs: t0 + 14*minute, EndMs: t0 + 30*minute, UpdatedMs: t0 + 30*minute, Samples: 192, + First: json.RawMessage(`{"requested_w":4298,"actual_w":1298}`)} + for _, w := range []float64{4298, 4250, 4162} { + limited.Requested.Add(&w) + } + actual := 1250.0 + limited.Actual.Add(&actual) + following := &state.ControlSpan{Subject: "battery:sungrow", Driver: "sungrow", Kind: "battery", Status: "following", + Reason: "idle", Severity: "info", StartMs: t0 + 30*minute, UpdatedMs: t0 + 40*minute, Samples: 120} + if err := st.SaveControlSpans(context.Background(), []*state.ControlSpan{limited, following}); err != nil { + t.Fatal(err) + } + if err := st.RecordNotification(state.NotificationEntry{TsMs: t0 + 20*minute, EventType: "control.not_following", Status: "sent"}); err != nil { + t.Fatal(err) + } + srv := New(&Deps{State: st}) + + get := func(query string) (map[string]json.RawMessage, int) { + rec := httptest.NewRecorder() + srv.handleControlJournal(rec, httptest.NewRequest(http.MethodGet, "/api/control/journal?"+query, nil)) + var out map[string]json.RawMessage + _ = json.Unmarshal(rec.Body.Bytes(), &out) + return out, rec.Code + } + out, code := get("since=" + msText(t0+15*minute) + "&until=" + msText(t0+20*minute) + "&severity=warning") + if code != http.StatusOK { + t.Fatalf("status %d", code) + } + var spans []struct { + Status string `json:"status"` + EndMs *int64 `json:"end_ms"` + DurationS float64 `json:"duration_s"` + RequestedW *struct{ Mean float64 } `json:"requested_w"` + ActualW *struct{ Max float64 } `json:"actual_w"` + First map[string]float64 `json:"first"` + } + _ = json.Unmarshal(out["spans"], &spans) + if len(spans) != 1 || spans[0].Status != "limited" || spans[0].EndMs == nil || spans[0].DurationS != 960 { + t.Fatalf("spans: %s", out["spans"]) + } + if spans[0].RequestedW == nil || spans[0].RequestedW.Mean < 4236 || spans[0].RequestedW.Mean > 4237 || spans[0].ActualW.Max != 1250 || spans[0].First["requested_w"] != 4298 { + t.Fatalf("numbers: %s", out["spans"]) + } + var notices []state.NotificationEntry + _ = json.Unmarshal(out["notifications"], ¬ices) + if len(notices) != 1 || notices[0].EventType != "control.not_following" { + t.Fatalf("notifications: %s", out["notifications"]) + } + + // The open span runs to now and says so with a null end. + out, _ = get("since=" + msText(t0+35*minute) + "&until=" + msText(t0+36*minute)) + _ = json.Unmarshal(out["spans"], &spans) + if len(spans) != 1 || spans[0].Status != "following" || spans[0].EndMs != nil { + t.Fatalf("open span: %s", out["spans"]) + } + if _, code := get("since=10&until=5"); code != http.StatusBadRequest { + t.Fatalf("reversed window: status %d", code) + } +} + +func msText(v int64) string { return strconv.FormatInt(v, 10) } diff --git a/go/internal/appproto/agents.go b/go/internal/appproto/agents.go index 1ac321b40..eeb885975 100644 --- a/go/internal/appproto/agents.go +++ b/go/internal/appproto/agents.go @@ -811,8 +811,10 @@ func (h *Handler) onRead(env Envelope) error { return h.onConfigGet(env) case protocol.MsgCarsGet: return h.onCarsGet(env) + case protocol.MsgEventsGet: + return h.onEventsGet(env) default: - // events.get and diag.run are in the contract and not built yet. + // diag.run is in the contract and not built yet. return h.refuse(env.ID, ErrUnknownOp, map[string]any{"t": env.T}) } } diff --git a/go/internal/appproto/agents_test.go b/go/internal/appproto/agents_test.go index 7d0b91ae0..315c1acce 100644 --- a/go/internal/appproto/agents_test.go +++ b/go/internal/appproto/agents_test.go @@ -60,7 +60,7 @@ func (a *testAccess) EndHolds(agentID string) { func (a *testAccess) SetApprovalKey(d string, m protocol.ApprovalKeySet) (int64, error) { return a.id.SetApprovalKey(d, m) } -func (a *testAccess) Wall() (int64, bool) { return a.wall() } +func (a *testAccess) Wall() (int64, bool) { return a.wall() } func (a *testAccess) Notify(kind, _ string) { a.notified = append(a.notified, kind) } // enrolGrants reads one row the way appuplink's liveGrant does. @@ -133,6 +133,8 @@ type agentRig struct { owner appenroll.Grant signer *approvaltest.Signer logs []redact.Line + // journal, when a test sets it, serves events.get. + journal JournalSource // cars and chargers, when a test sets them, are the box's car store // and the plan's chargers. cars *fakeCars @@ -214,7 +216,7 @@ func (r *agentRig) handler(kind, dev, role string) (*Handler, *recorder) { Clock: r.clock, Site: r.box, Info: r.box, Modes: r.box, Plans: r.box, Codec: testCodec{}, Sender: rec, Caller: caller, Grants: grants, Commands: NewCmdLog(), Loadpoints: r.lp, Access: r.access, Agents: r.agents, Audit: r.audit, - Logs: r, Redactor: func() *redact.Redactor { return redact.New(testAgentKey, []string{"Easee carport"}) }, + Logs: r, Journal: r.journal, Redactor: func() *redact.Redactor { return redact.New(testAgentKey, []string{"Easee carport"}) }, Views: testViews{}, AgentKey: testAgentKey, SrcGrid: "meter.p1", SrcPV: "inverter.sungrow", SrcBattery: "battery.sungrow", NewLeaseID: func() string { return "lease-test" }, diff --git a/go/internal/appproto/events.go b/go/internal/appproto/events.go new file mode 100644 index 000000000..433a5c66b --- /dev/null +++ b/go/internal/appproto/events.go @@ -0,0 +1,171 @@ +package appproto + +import ( + "context" + "encoding/json" + "time" + + "github.com/fxamacker/cbor/v2" + + "github.com/srcfl/ftw/go/protocol" +) + +// JournalSpan is a control journal span as events.get reads it, before +// its driver name becomes an agent id. +type JournalSpan struct { + Driver string + Kind string + Status string + Reason string + Severity string + Mode string + StartMs int64 + EndMs int64 + Samples int64 + Requested *protocol.PowerRange + Actual *protocol.PowerRange + // Last is the verdict as Core served it at the latest sample. + Last json.RawMessage +} + +// JournalNotice is one notification attempt as events.get reads it. +type JournalNotice struct { + TsMs int64 + Type string + Driver string + Status string + Priority int +} + +// JournalQuery picks what events.get reads. +type JournalQuery struct { + SinceMs, UntilMs int64 + // Driver keeps one source; empty keeps all. + Driver string + // Severities keeps these; empty keeps all. + Severities []string + Limit int +} + +// JournalSource is the box's control journal and notification history. +type JournalSource interface { + Journal(ctx context.Context, q JournalQuery) (spans []JournalSpan, notices []JournalNotice, more bool, err error) +} + +// eventsWindow is how far back events.get looks when asked for no window. +const eventsWindow = 24 * time.Hour + +// onEventsGet answers what the box recorded about control over a window. +// Names stay home: a source goes by its agent id, and only the numbers of a +// verdict's text leave the box. +func (h *Handler) onEventsGet(env Envelope) error { + if h.cfg.Journal == nil { + return h.refuse(env.ID, ErrUnavailable, map[string]any{"subsystem": "events"}) + } + var m protocol.EventsGet + if len(env.B) > 0 { + if err := Unmarshal(env.B, &m); err != nil { + return h.refuse(env.ID, ErrUnknownOp, map[string]any{"t": env.T}) + } + } + if m.BeforeMs < 0 || m.AfterMs < 0 || (m.BeforeMs > 0 && m.AfterMs >= m.BeforeMs) { + return h.refuse(env.ID, ErrUnknownOp, map[string]any{"t": env.T, "field": "beforeMs"}) + } + q := JournalQuery{SinceMs: m.AfterMs, UntilMs: m.BeforeMs, Limit: int(m.N)} + nowMs, _ := h.wall() + if q.UntilMs == 0 { + q.UntilMs = nowMs + } + if q.SinceMs == 0 { + q.SinceMs = q.UntilMs - eventsWindow.Milliseconds() + } + if q.Limit <= 0 || q.Limit > protocol.EventsMax { + q.Limit = protocol.EventsMax + } + switch m.MinSeverity { + case "", "info": + case "warning": + q.Severities = []string{"warning", "alarm"} + case "alarm": + q.Severities = []string{"alarm"} + default: + return h.refuse(env.ID, ErrUnknownOp, map[string]any{"t": env.T, "field": "minSeverity"}) + } + if m.Src != "" { + driver, ok := h.sourceForID(m.Src) + if !ok { + return h.refuse(env.ID, ErrUnknownOp, map[string]any{"t": env.T, "field": "src"}) + } + q.Driver = driver + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + spans, notices, more, err := h.cfg.Journal.Journal(ctx, q) + if err != nil { + return h.refuse(env.ID, ErrUnavailable, map[string]any{"subsystem": "events"}) + } + out := protocol.Events{Events: []cbor.RawMessage{}, Spans: []protocol.EventSpan{}, Notices: []protocol.EventNotice{}, More: more} + for _, sp := range spans { + es := protocol.EventSpan{ + Kind: sp.Kind, Status: sp.Status, Reason: sp.Reason, Severity: sp.Severity, Mode: sp.Mode, + StartMs: sp.StartMs, EndMs: sp.EndMs, Samples: sp.Samples, + RequestedW: sp.Requested, ActualW: sp.Actual, Last: verdictNumbers(sp.Last), + } + if sp.Driver != "" { + es.Src = h.srcID(sp.Driver) + } + out.Spans = append(out.Spans, es) + } + for _, n := range notices { + en := protocol.EventNotice{Ts: n.TsMs, Type: n.Type, Status: n.Status, Priority: n.Priority} + if n.Driver != "" { + en.Src = h.srcID(n.Driver) + } + out.Notices = append(out.Notices, en) + } + // One bulk frame holds the answer; the oldest give way and More says so. + for !h.fits(protocol.MsgEvents, env.ID, out) && (len(out.Spans) > 0 || len(out.Notices) > 0) { + out.More = true + if len(out.Notices) > len(out.Spans) { + out.Notices = out.Notices[:len(out.Notices)*3/4] + } else { + out.Spans = out.Spans[:len(out.Spans)*3/4] + } + } + return h.sendBulk(protocol.MsgEvents, env.ID, out) +} + +// sourceForID finds the source an agent id names. +func (h *Handler) sourceForID(id string) (string, bool) { + if h.cfg.Site == nil { + return "", false + } + for _, s := range h.cfg.Site.Snapshot().Sources { + if h.srcID(s.ID) == id { + return s.ID, true + } + } + return "", false +} + +// verdictNumbers keeps the numeric fields of a verdict's text. Names are +// text and stay home. +func verdictNumbers(raw json.RawMessage) map[string]float64 { + if len(raw) == 0 { + return nil + } + var fields map[string]any + if json.Unmarshal(raw, &fields) != nil { + return nil + } + out := map[string]float64{} + for k, v := range fields { + if f, ok := v.(float64); ok { + out[k] = f + } + } + if len(out) == 0 { + return nil + } + return out +} diff --git a/go/internal/appproto/events_test.go b/go/internal/appproto/events_test.go new file mode 100644 index 000000000..474d768f7 --- /dev/null +++ b/go/internal/appproto/events_test.go @@ -0,0 +1,86 @@ +package appproto + +import ( + "context" + "encoding/json" + "strings" + "testing" + "time" + + "github.com/srcfl/ftw/go/internal/apiauth" + "github.com/srcfl/ftw/go/protocol" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +type fakeJournal struct { + spans []JournalSpan + notices []JournalNotice + asked JournalQuery +} + +func (f *fakeJournal) Journal(_ context.Context, q JournalQuery) ([]JournalSpan, []JournalNotice, bool, error) { + f.asked = q + return f.spans, f.notices, false, nil +} + +// An agent asks why the app warned this morning. It gets the box's own +// verdicts and numbers over the window, and no names. +func TestEventsReachAnAgentWithoutNames(t *testing.T) { + r := newAgentRig(t) + j := &fakeJournal{ + spans: []JournalSpan{{ + Driver: "battery.sungrow", Kind: "battery", Status: "limited", Reason: "power_below_target", Severity: "warning", + Mode: "planner_arbitrage", StartMs: 1_000, EndMs: 961_000, Samples: 192, + Requested: &protocol.PowerRange{Min: 4162, Max: 4361, Mean: 4255}, + Actual: &protocol.PowerRange{Min: 1162, Max: 1362, Mean: 1255}, + Last: json.RawMessage(`{"driver":"battery.sungrow","site_meter":"meter.p1","sent_w":4250,"readback_w":4239,"battery_soc":0.2}`), + }, {Kind: "dispatch", Status: "free", Severity: "info", StartMs: 500}}, + notices: []JournalNotice{{TsMs: 2_000, Type: "control.not_following", Driver: "battery.sungrow", Status: "sent", Priority: 4}}, + } + r.journal = j + row, _ := r.grant(apiauth.RoleAgent, []string{protocol.ScopeEventsRead}, nil, protocol.Caps{}, time.Hour, protocol.Principal{}) + h, rec := r.agentSession(row) + src := cloudwire.SourceID(testAgentKey, "battery.sungrow") + deliver(t, h, protocol.MsgEventsGet, ptrU32(1), protocol.EventsGet{AfterMs: 100, BeforeMs: 2_000_000, Src: src, MinSeverity: "warning"}) + ev := body[protocol.Events](t, rec.last(t, protocol.MsgEvents)) + + if j.asked.Driver != "battery.sungrow" || j.asked.SinceMs != 100 || j.asked.UntilMs != 2_000_000 || + len(j.asked.Severities) != 2 || j.asked.Limit != protocol.EventsMax { + t.Fatalf("query %+v", j.asked) + } + if len(ev.Spans) != 2 || len(ev.Notices) != 1 { + t.Fatalf("answer %+v", ev) + } + sp := ev.Spans[0] + if sp.Src != src || sp.Status != "limited" || sp.ActualW == nil || sp.ActualW.Mean != 1255 || sp.Last["sent_w"] != 4250 || sp.Last["battery_soc"] != 0.2 { + t.Fatalf("span %+v", sp) + } + if ev.Spans[1].Src != "" || ev.Notices[0].Src != src { + t.Fatalf("ids: %+v / %+v", ev.Spans[1], ev.Notices[0]) + } + raw, _ := json.Marshal(ev) + for _, name := range []string{"battery.sungrow", "meter.p1"} { + if strings.Contains(string(raw), name) { + t.Errorf("the answer names %q: %s", name, raw) + } + } +} + +func TestEventsRefusesWhatItCannotAnswer(t *testing.T) { + r := newAgentRig(t) + row, _ := r.grant(apiauth.RoleAgent, []string{protocol.ScopeEventsRead}, nil, protocol.Caps{}, time.Hour, protocol.Principal{}) + h, rec := r.agentSession(row) + deliver(t, h, protocol.MsgEventsGet, ptrU32(1), protocol.EventsGet{}) + if got := errCode(t, rec); got != ErrUnavailable { + t.Errorf("a box without a journal: %s", got) + } + r.journal = &fakeJournal{} + h, rec = r.agentSession(row) + for _, m := range []protocol.EventsGet{{AfterMs: 7, BeforeMs: 7}, {Src: "not-a-source"}, {MinSeverity: "loud"}} { + rec.reset() + deliver(t, h, protocol.MsgEventsGet, ptrU32(2), m) + if got := errCode(t, rec); got != ErrUnknownOp { + t.Errorf("%+v: %s", m, got) + } + } +} diff --git a/go/internal/appproto/handler.go b/go/internal/appproto/handler.go index 5984f8c37..45963cf00 100644 --- a/go/internal/appproto/handler.go +++ b/go/internal/appproto/handler.go @@ -107,7 +107,9 @@ type Config struct { // Logs and Redactor serve logs.get, ConfigDoc and Redactor config.get, // Views live.get and loadpoints.get. Each read answers E_UNAVAILABLE // without its ports. - Logs LogSource + Logs LogSource + // Journal serves events.get: the control journal and notifications. + Journal JournalSource Redactor func() *redact.Redactor ConfigDoc func() (any, error) Views AgentViews diff --git a/go/internal/control/dispatch.go b/go/internal/control/dispatch.go index 37a3899e5..88843fda2 100644 --- a/go/internal/control/dispatch.go +++ b/go/internal/control/dispatch.go @@ -529,6 +529,9 @@ type State struct { // the API; mutated only inside ComputeDispatch under the caller's // outer ctrlMu. SlotDeliveryStats SlotDeliveryStats + // SlotScored, when set, receives each plan slot Core scored. It runs + // under the control lock and must not block. + SlotScored func(SlotDelivery) // PlanStale tracks whether the last cycle fell back to self_consumption // because the plan was missing. Surfaced via the API for the UI. @@ -802,6 +805,22 @@ func resetEnergyDispatchBookkeeping(state *State) { state.lastTickTs = time.Time{} } +// SlotDelivery is one scored plan slot: the battery energy the plan asked +// for over the slot, what the fleet delivered, and Core's verdict on it. +type SlotDelivery struct { + SlotStart time.Time + SlotEnd time.Time + Mode Mode + PlannedWh float64 + ActualWh float64 + // Ratio is |actual| / |planned|; 0 for a sign mismatch. + Ratio float64 + Replans uint64 + // Verdict is on_target, under_delivered, over_delivered or + // sign_mismatch. + Verdict string +} + // SlotDeliveryStats tracks observed gaps between the plan's per-slot // BatteryEnergyWh and the live fleet's actual delivered Wh. Updated at // slot rollover by updateSlotDeliveryMetrics. Pure observability — no @@ -885,48 +904,61 @@ func updateSlotDeliveryMetrics(state *State, currentTotalW float64, now time.Tim func scoreSlotDeliveryMetrics(state *State) { plannedWh := state.slotActualPlannedWh - if math.Abs(plannedWh) > slotDeliveryIdleWhCutoff { - actualWh := state.slotActualWh - // Sign mismatch is the categorical failure mode: we moved - // energy in the opposite direction from the plan. Caught - // before the magnitude-ratio check below, which would - // otherwise treat opposite-sign-equal-magnitude (planned - // −425, actual +425) as ratio 1.0 = on target. - if plannedWh*actualWh < 0 && math.Abs(actualWh) > slotDeliveryIdleWhCutoff { - slog.Info("dispatch slot sign mismatch", + if math.Abs(plannedWh) <= slotDeliveryIdleWhCutoff { + return + } + actualWh := state.slotActualWh + d := SlotDelivery{ + SlotStart: state.slotActualSlotStart, SlotEnd: state.slotActualDirective.SlotEnd, + Mode: state.Mode, PlannedWh: plannedWh, ActualWh: actualWh, + Replans: state.slotActualReplans, Verdict: "on_target", + } + // Sign mismatch is the categorical failure mode: we moved + // energy in the opposite direction from the plan. Caught + // before the magnitude-ratio check below, which would + // otherwise treat opposite-sign-equal-magnitude (planned + // −425, actual +425) as ratio 1.0 = on target. + if plannedWh*actualWh < 0 && math.Abs(actualWh) > slotDeliveryIdleWhCutoff { + slog.Info("dispatch slot sign mismatch", + "mode", state.Mode, + "replans", state.slotActualReplans, + "planned_basis", "accepted_intervals", + "planned_wh", plannedWh, + "actual_wh", actualWh, + "slot_start", state.slotActualSlotStart) + state.SlotDeliveryStats.SignMismatchCount++ + d.Verdict = "sign_mismatch" + } else { + ratio := math.Abs(actualWh) / math.Max(math.Abs(plannedWh), 1) + d.Ratio = ratio + switch { + case ratio > slotDeliveryOverThreshold: + slog.Info("dispatch slot over-delivery", "mode", state.Mode, "replans", state.slotActualReplans, "planned_basis", "accepted_intervals", "planned_wh", plannedWh, "actual_wh", actualWh, + "ratio", ratio, "slot_start", state.slotActualSlotStart) - state.SlotDeliveryStats.SignMismatchCount++ - } else { - ratio := math.Abs(actualWh) / math.Max(math.Abs(plannedWh), 1) - switch { - case ratio > slotDeliveryOverThreshold: - slog.Info("dispatch slot over-delivery", - "mode", state.Mode, - "replans", state.slotActualReplans, - "planned_basis", "accepted_intervals", - "planned_wh", plannedWh, - "actual_wh", actualWh, - "ratio", ratio, - "slot_start", state.slotActualSlotStart) - state.SlotDeliveryStats.OverDeliveryCount++ - case ratio < slotDeliveryUnderThreshold: - slog.Info("dispatch slot under-delivery", - "mode", state.Mode, - "replans", state.slotActualReplans, - "planned_basis", "accepted_intervals", - "planned_wh", plannedWh, - "actual_wh", actualWh, - "ratio", ratio, - "slot_start", state.slotActualSlotStart) - state.SlotDeliveryStats.UnderDeliveryCount++ - } + state.SlotDeliveryStats.OverDeliveryCount++ + d.Verdict = "over_delivered" + case ratio < slotDeliveryUnderThreshold: + slog.Info("dispatch slot under-delivery", + "mode", state.Mode, + "replans", state.slotActualReplans, + "planned_basis", "accepted_intervals", + "planned_wh", plannedWh, + "actual_wh", actualWh, + "ratio", ratio, + "slot_start", state.slotActualSlotStart) + state.SlotDeliveryStats.UnderDeliveryCount++ + d.Verdict = "under_delivered" } } + if state.SlotScored != nil { + state.SlotScored(d) + } } type plannerSelfDecision struct { diff --git a/go/internal/control/slot_scored_test.go b/go/internal/control/slot_scored_test.go new file mode 100644 index 000000000..babf2d0df --- /dev/null +++ b/go/internal/control/slot_scored_test.go @@ -0,0 +1,39 @@ +package control + +import ( + "math" + "testing" + "time" +) + +// A scored slot reaches SlotScored with its numbers and verdict, so the +// control journal can keep it after the log line has rolled away. +func TestSlotScoredReportsEachScoredSlot(t *testing.T) { + now := time.Now() + slot1Start := now.Add(-1 * time.Minute) + slot1 := SlotDirective{SlotStart: slot1Start, SlotEnd: slot1Start.Add(15 * time.Minute), BatteryEnergyWh: 1250} + slot2 := SlotDirective{SlotStart: slot1.SlotEnd, SlotEnd: slot1.SlotEnd.Add(15 * time.Minute), BatteryEnergyWh: 1250} + store := seedStore(0, []struct { + name string + currentW, soc float64 + }{{"ferroamp", 1200, 0.2}}) + active := slot1 + st := makeSlotMetricsState("ferroamp", func(time.Time) (SlotDirective, bool) { return active, true }) + var got []SlotDelivery + st.SlotScored = func(d SlotDelivery) { got = append(got, d) } + + _ = ComputeDispatch(store, st, caps(map[string]float64{"ferroamp": 9600}), 11040) + st.slotPlannedPastWh = 1250 + st.slotActualWh = 296 // 06:15 on 8 November 2026: 24 % of the plan + + active = slot2 + _ = ComputeDispatch(store, st, caps(map[string]float64{"ferroamp": 9600}), 11040) + if len(got) != 1 { + t.Fatalf("want one scored slot, got %+v", got) + } + d := got[0] + if d.Verdict != "under_delivered" || math.Abs(d.PlannedWh-1250) > 0.01 || math.Abs(d.ActualWh-296) > 0.01 || + math.Abs(d.Ratio-0.2368) > 1e-4 || !d.SlotEnd.Equal(slot1.SlotEnd) || d.SlotStart.IsZero() { + t.Fatalf("scored slot: %+v", d) + } +} diff --git a/go/internal/journal/journal.go b/go/internal/journal/journal.go new file mode 100644 index 000000000..dbe8fd1ba --- /dev/null +++ b/go/internal/journal/journal.go @@ -0,0 +1,309 @@ +// Package journal records how the box's answer to "Are we in control?" +// changes, so a warning can be explained after it has passed. Every few +// seconds it takes each subject's verdict: a device function's control +// feedback, a source's freshness, whether the site may dispatch, the mode. +// A changed verdict closes the subject's span and opens the next, so the +// spans tile its history. A span carries the power Core asked for and the +// power measured while it held, and the verdict as Core served it. +package journal + +import ( + "context" + "encoding/json" + "log/slog" + "sync" + "time" + + "github.com/srcfl/ftw/go/internal/state" +) + +// Verdict is one subject's state at one sample. +type Verdict struct { + // Subject names what holds the verdict, e.g. "battery:sungrow". + Subject string + Driver string + Kind string + Status string + Reason string + Severity string + Mode string + // RequestedW and ActualW are nil when unknown, never an invented 0. + RequestedW *float64 + ActualW *float64 + // Detail is the verdict as Core served it, kept as the span's first + // and latest record. + Detail json.RawMessage +} + +// Store saves spans. state.Store implements it. +type Store interface { + SaveControlSpans(ctx context.Context, spans []*state.ControlSpan) error +} + +const ( + // StatsEvery is how often a span that still holds writes its running + // numbers. A changed verdict is written at the next flush. + StatsEvery = time.Minute + // maxPending bounds what a failing database leaves in memory. + maxPending = 5000 + // failLogEvery spaces the warnings while writes keep failing. + failLogEvery = 10 * time.Minute + // Settle is how long a change to an info verdict must hold before it + // starts a span. Each new order makes a device "waiting" for a second + // or two; that is not a change worth a span. A warning or an alarm + // starts one at once, however short, and a settled change is dated + // from when it was first seen. + Settle = 15 * time.Second +) + +// Recorder turns samples into spans and writes them. Observe and Flush run +// on one goroutine; Record may be called from any. +type Recorder struct { + mu sync.Mutex + store Store + open map[string]*state.ControlSpan + // candidate holds a change to an info verdict until it settles. + candidate map[string]*state.ControlSpan + // pending holds spans with changes not yet written, each with the + // version it was marked at, so a write of an older copy does not clear + // a newer change. + pending map[*state.ControlSpan]uint64 + order []*state.ControlSpan + version uint64 + statsAt time.Time + failingAt time.Time + loggedAt time.Time + dropped int +} + +// New returns a recorder that writes to store. +func New(store Store) *Recorder { + return &Recorder{store: store, open: map[string]*state.ControlSpan{}, candidate: map[string]*state.ControlSpan{}, + pending: map[*state.ControlSpan]uint64{}} +} + +// Observe takes one sample of every subject. A subject missing from the +// sample has gone, and its span ends now. +func (r *Recorder) Observe(now time.Time, verdicts []Verdict) { + if r == nil { + return + } + nowMs := now.UnixMilli() + r.mu.Lock() + defer r.mu.Unlock() + seen := map[string]bool{} + for _, v := range verdicts { + if v.Subject == "" || seen[v.Subject] { + continue + } + seen[v.Subject] = true + sp := r.open[v.Subject] + if sp != nil && sameVerdict(sp, v) { + addSample(sp, v, nowMs) + delete(r.candidate, v.Subject) + continue + } + if sp == nil || v.Severity == "warning" || v.Severity == "alarm" { + delete(r.candidate, v.Subject) + r.startLocked(newSpan(v, nowMs), nowMs) + continue + } + c := r.candidate[v.Subject] + if c == nil || !sameVerdict(c, v) { + r.candidate[v.Subject] = newSpan(v, nowMs) + continue + } + addSample(c, v, nowMs) + if nowMs-c.StartMs >= Settle.Milliseconds() { + delete(r.candidate, v.Subject) + r.startLocked(c, c.StartMs) + } + } + for subject, sp := range r.open { + if !seen[subject] { + r.closeLocked(sp, nowMs) + delete(r.open, subject) + delete(r.candidate, subject) + } + } +} + +// startLocked ends the subject's span at atMs and opens next. +func (r *Recorder) startLocked(next *state.ControlSpan, atMs int64) { + if sp := r.open[next.Subject]; sp != nil { + r.closeLocked(sp, atMs) + } + r.open[next.Subject] = next + r.markLocked(next) +} + +func sameVerdict(sp *state.ControlSpan, v Verdict) bool { + return sp.Status == v.Status && sp.Reason == v.Reason && sp.Severity == v.Severity && sp.Mode == v.Mode +} + +func newSpan(v Verdict, nowMs int64) *state.ControlSpan { + sp := &state.ControlSpan{ + Subject: v.Subject, Driver: v.Driver, Kind: v.Kind, + Status: v.Status, Reason: v.Reason, Severity: v.Severity, Mode: v.Mode, + StartMs: nowMs, UpdatedMs: nowMs, Samples: 1, First: v.Detail, Last: v.Detail, + } + sp.Requested.Add(v.RequestedW) + sp.Actual.Add(v.ActualW) + return sp +} + +func addSample(sp *state.ControlSpan, v Verdict, nowMs int64) { + sp.Samples++ + sp.UpdatedMs = nowMs + sp.Requested.Add(v.RequestedW) + sp.Actual.Add(v.ActualW) + if len(v.Detail) > 0 { + sp.Last = v.Detail + } +} + +// Record adds a span that is already complete, such as a scored plan slot. +func (r *Recorder) Record(sp state.ControlSpan) { + if r == nil { + return + } + r.mu.Lock() + defer r.mu.Unlock() + r.markLocked(&sp) +} + +// CloseAll ends every open span now, for a clean shutdown. +func (r *Recorder) CloseAll(now time.Time) { + if r == nil { + return + } + r.mu.Lock() + defer r.mu.Unlock() + for subject, sp := range r.open { + r.closeLocked(sp, now.UnixMilli()) + delete(r.open, subject) + } + clear(r.candidate) +} + +// Flush writes changed spans, and once a StatsEvery the running numbers of +// spans that still hold. A failed write keeps everything for the next one. +func (r *Recorder) Flush(ctx context.Context, now time.Time) { + if r == nil || r.store == nil { + return + } + type entry struct { + orig *state.ControlSpan + version uint64 + } + r.mu.Lock() + var batch []entry + inBatch := map[*state.ControlSpan]bool{} + for _, sp := range r.order { + if v, ok := r.pending[sp]; ok && !inBatch[sp] { + batch = append(batch, entry{sp, v}) + inBatch[sp] = true + } + } + if now.Sub(r.statsAt) >= StatsEvery { + r.statsAt = now + for _, sp := range r.open { + if !inBatch[sp] { + batch = append(batch, entry{sp, 0}) + inBatch[sp] = true + } + } + } + copies := make([]*state.ControlSpan, len(batch)) + for i, e := range batch { + c := *e.orig + copies[i] = &c + } + r.mu.Unlock() + if len(copies) == 0 { + return + } + + err := r.store.SaveControlSpans(ctx, copies) + + r.mu.Lock() + defer r.mu.Unlock() + if err != nil { + if r.failingAt.IsZero() { + r.failingAt = now + } + if now.Sub(r.loggedAt) >= failLogEvery { + r.loggedAt = now + slog.Warn("control journal: write failed; keeping spans for the next try", + "pending", len(r.pending), "since", r.failingAt, "err", err) + } + return + } + if !r.failingAt.IsZero() { + slog.Info("control journal: writes recovered", "failing_since", r.failingAt, "dropped", r.dropped) + r.failingAt, r.loggedAt, r.dropped = time.Time{}, time.Time{}, 0 + } + for i, e := range batch { + if e.orig.ID == 0 { + e.orig.ID = copies[i].ID + } + if v, ok := r.pending[e.orig]; ok && v == e.version { + delete(r.pending, e.orig) + } + } + r.compactLocked() +} + +// Open reports the spans that hold now, for tests and diagnostics. +func (r *Recorder) Open() []state.ControlSpan { + r.mu.Lock() + defer r.mu.Unlock() + out := make([]state.ControlSpan, 0, len(r.open)) + for _, sp := range r.open { + out = append(out, *sp) + } + return out +} + +func (r *Recorder) closeLocked(sp *state.ControlSpan, endMs int64) { + sp.EndMs = endMs + if endMs > sp.UpdatedMs { + sp.UpdatedMs = endMs + } + r.markLocked(sp) +} + +func (r *Recorder) markLocked(sp *state.ControlSpan) { + r.version++ + if _, ok := r.pending[sp]; !ok { + r.order = append(r.order, sp) + } + r.pending[sp] = r.version + for len(r.pending) > maxPending { + // The oldest closed span gives way; an open one is still needed. + dropped := false + for i, old := range r.order { + if _, ok := r.pending[old]; ok && old.EndMs != 0 { + delete(r.pending, old) + r.order = append(r.order[:i], r.order[i+1:]...) + r.dropped++ + dropped = true + break + } + } + if !dropped { + break + } + } +} + +// compactLocked drops written spans from the order list. +func (r *Recorder) compactLocked() { + kept := r.order[:0] + for _, sp := range r.order { + if _, ok := r.pending[sp]; ok { + kept = append(kept, sp) + } + } + r.order = kept +} diff --git a/go/internal/journal/journal_test.go b/go/internal/journal/journal_test.go new file mode 100644 index 000000000..27cdbabc8 --- /dev/null +++ b/go/internal/journal/journal_test.go @@ -0,0 +1,199 @@ +package journal + +import ( + "context" + "encoding/json" + "errors" + "testing" + "time" + + "github.com/srcfl/ftw/go/internal/state" +) + +// fakeStore keeps the latest copy of each span by id, as the table would. +type fakeStore struct { + rows map[int64]state.ControlSpan + nextID int64 + fail bool + writes int +} + +func (f *fakeStore) SaveControlSpans(_ context.Context, spans []*state.ControlSpan) error { + f.writes++ + if f.fail { + return errors.New("database is locked") + } + if f.rows == nil { + f.rows = map[int64]state.ControlSpan{} + } + for _, sp := range spans { + if sp.ID == 0 { + f.nextID++ + sp.ID = f.nextID + } + f.rows[sp.ID] = *sp + } + return nil +} + +func (f *fakeStore) bySubject(subject string) []state.ControlSpan { + var out []state.ControlSpan + for id := int64(1); id <= f.nextID; id++ { + if sp, ok := f.rows[id]; ok && sp.Subject == subject { + out = append(out, sp) + } + } + return out +} + +func p(v float64) *float64 { return &v } + +func battery(status, reason, severity string, requested, actual float64) Verdict { + detail, _ := json.Marshal(map[string]any{"status": status, "requested_w": requested, "actual_w": actual}) + return Verdict{Subject: "battery:sungrow", Driver: "sungrow", Kind: "battery", Status: status, Reason: reason, + Severity: severity, Mode: "planner_arbitrage", RequestedW: p(requested), ActualW: p(actual), Detail: detail} +} + +// The morning of 8 November 2026: the battery was ordered 4.2 kW and took +// 1.2 kW for a quarter of an hour. The journal keeps the stretch, its +// numbers and the verdict before and after it. +func TestRecorderTilesAVerdictsHistory(t *testing.T) { + st := &fakeStore{} + r := New(st) + t0 := time.Date(2026, 11, 8, 6, 0, 0, 0, time.Local) + at := func(min int) time.Time { return t0.Add(time.Duration(min) * time.Minute) } + + r.Observe(at(0), []Verdict{battery("following", "plan", "info", 2440, 2440)}) + r.Flush(context.Background(), at(0)) + for m := 14; m <= 27; m++ { + r.Observe(at(m), []Verdict{battery("limited", "power_below_target", "warning", 4250, 1250)}) + r.Flush(context.Background(), at(m)) + } + // The return to an info verdict counts once it has held for Settle, + // and from when it was first seen. + r.Observe(at(30), []Verdict{battery("following", "idle", "info", 0, 0)}) + r.Flush(context.Background(), at(30)) + r.Observe(at(30).Add(Settle), []Verdict{battery("following", "idle", "info", 0, 0)}) + r.Flush(context.Background(), at(30).Add(Settle)) + + spans := st.bySubject("battery:sungrow") + if len(spans) != 3 { + t.Fatalf("want 3 spans, got %+v", spans) + } + first, warn, last := spans[0], spans[1], spans[2] + if first.EndMs != warn.StartMs || warn.EndMs != last.StartMs || last.EndMs != 0 { + t.Fatalf("spans do not tile: %d-%d, %d-%d, %d-", first.StartMs, first.EndMs, warn.StartMs, warn.EndMs, last.StartMs) + } + if warn.StartMs != at(14).UnixMilli() || warn.EndMs != at(30).UnixMilli() || warn.Severity != "warning" || warn.Samples != 14 { + t.Fatalf("warning span: %+v", warn) + } + if mean, _ := warn.Requested.Mean(); mean != 4250 { + t.Fatalf("requested mean %v", mean) + } + if warn.Actual.Max != 1250 || len(warn.First) == 0 || len(warn.Last) == 0 { + t.Fatalf("actual or verdict text lost: %+v", warn) + } +} + +// A subject that drops out of the sample, such as a driver removed from +// the config, has its span closed rather than left running. +func TestRecorderClosesASubjectThatWent(t *testing.T) { + st := &fakeStore{} + r := New(st) + t0 := time.Unix(1_000, 0) + r.Observe(t0, []Verdict{{Subject: "source:easee", Kind: "source", Status: "live", Severity: "info"}}) + r.Observe(t0.Add(5*time.Second), nil) + r.Flush(context.Background(), t0.Add(5*time.Second)) + got := st.bySubject("source:easee") + if len(got) != 1 || got[0].EndMs != t0.Add(5*time.Second).UnixMilli() { + t.Fatalf("span not closed: %+v", got) + } + if len(r.Open()) != 0 { + t.Fatalf("still open: %+v", r.Open()) + } +} + +// A locked database must not lose a verdict: the spans wait in memory and +// the next flush writes them, with the change that came in between. +func TestRecorderKeepsSpansThroughFailedWrites(t *testing.T) { + st := &fakeStore{fail: true} + r := New(st) + t0 := time.Unix(10_000, 0) + r.Observe(t0, []Verdict{battery("following", "plan", "info", 1000, 1000)}) + r.Flush(context.Background(), t0) + r.Observe(t0.Add(5*time.Second), []Verdict{battery("not_following", "no_power_response", "warning", 1000, 0)}) + r.Flush(context.Background(), t0.Add(5*time.Second)) + if len(st.rows) != 0 { + t.Fatalf("a failing store took rows: %+v", st.rows) + } + st.fail = false + r.Flush(context.Background(), t0.Add(10*time.Second)) + spans := st.bySubject("battery:sungrow") + if len(spans) != 2 || spans[0].EndMs != spans[1].StartMs || spans[1].Status != "not_following" { + t.Fatalf("spans after recovery: %+v", spans) + } + writes := st.writes + r.Flush(context.Background(), t0.Add(15*time.Second)) + if st.writes != writes { + t.Fatal("nothing changed, yet the recorder wrote again") + } +} + +// A span that holds writes its running numbers once a minute, so a crash +// loses at most a minute of them. +func TestRecorderWritesHoldingSpansOnceAMinute(t *testing.T) { + st := &fakeStore{} + r := New(st) + t0 := time.Unix(20_000, 0) + r.Observe(t0, []Verdict{battery("following", "plan", "info", 500, 500)}) + r.Flush(context.Background(), t0) + for s := 5; s <= 55; s += 5 { + r.Observe(t0.Add(time.Duration(s)*time.Second), []Verdict{battery("following", "plan", "info", 500, 500)}) + r.Flush(context.Background(), t0.Add(time.Duration(s)*time.Second)) + } + if got := st.bySubject("battery:sungrow")[0].Samples; got != 1 { + t.Fatalf("samples written before a minute: %d", got) + } + r.Observe(t0.Add(65*time.Second), []Verdict{battery("following", "plan", "info", 500, 500)}) + r.Flush(context.Background(), t0.Add(65*time.Second)) + if got := st.bySubject("battery:sungrow")[0]; got.Samples != 13 || got.UpdatedMs != t0.Add(65*time.Second).UnixMilli() { + t.Fatalf("after a minute: %+v", got) + } +} + +func TestRecorderRecordsAFinishedSpan(t *testing.T) { + st := &fakeStore{} + r := New(st) + r.Record(state.ControlSpan{Subject: "slot:battery", Kind: "slot", Status: "under_delivered", Severity: "warning", StartMs: 1, EndMs: 900_001, UpdatedMs: 900_001}) + r.Flush(context.Background(), time.Unix(1_000, 0)) + if got := st.bySubject("slot:battery"); len(got) != 1 || got[0].Status != "under_delivered" { + t.Fatalf("slot span: %+v", got) + } +} + +// Each new order leaves a device "waiting" for a moment. A blip to an info +// verdict shorter than Settle starts no span; a warning does, however short. +func TestRecorderSettlesInfoChangesButNotWarnings(t *testing.T) { + st := &fakeStore{} + r := New(st) + t0 := time.Unix(30_000, 0) + tick := func(s int, v Verdict) { + at := t0.Add(time.Duration(s) * time.Second) + r.Observe(at, []Verdict{v}) + r.Flush(context.Background(), at) + } + tick(0, battery("following", "plan", "info", 1000, 1000)) + tick(5, battery("waiting", "waiting_response", "info", 1200, 1000)) + tick(10, battery("following", "plan", "info", 1200, 1200)) + if got := st.bySubject("battery:sungrow"); len(got) != 1 { + t.Fatalf("a 5 s info blip started a span: %+v", got) + } + tick(15, battery("not_following", "no_power_response", "warning", 1200, 0)) + tick(18, battery("following", "plan", "info", 1200, 1200)) + tick(33, battery("following", "plan", "info", 1200, 1200)) + got := st.bySubject("battery:sungrow") + if len(got) != 3 || got[1].Status != "not_following" || + got[1].StartMs != t0.Add(15*time.Second).UnixMilli() || got[1].EndMs != t0.Add(18*time.Second).UnixMilli() { + t.Fatalf("a 3 s warning must be its own span, ended when the recovery was first seen: %+v", got) + } +} diff --git a/go/internal/journal/slot.go b/go/internal/journal/slot.go new file mode 100644 index 000000000..c4f23974b --- /dev/null +++ b/go/internal/journal/slot.go @@ -0,0 +1,32 @@ +package journal + +import ( + "encoding/json" + + "github.com/srcfl/ftw/go/internal/control" + "github.com/srcfl/ftw/go/internal/state" +) + +// SlotSpan is a scored plan slot as the journal keeps it: the plan's +// average battery power over the slot as asked for, the delivered average +// as actual, and the energy behind both. +func SlotSpan(d control.SlotDelivery) state.ControlSpan { + severity := "info" + if d.Verdict == "under_delivered" || d.Verdict == "sign_mismatch" { + severity = "warning" + } + sp := state.ControlSpan{ + Subject: "slot:battery", Kind: "slot", Status: d.Verdict, Severity: severity, Mode: string(d.Mode), + StartMs: d.SlotStart.UnixMilli(), EndMs: d.SlotEnd.UnixMilli(), UpdatedMs: d.SlotEnd.UnixMilli(), Samples: 1, + } + if hours := d.SlotEnd.Sub(d.SlotStart).Hours(); hours > 0 { + planned, actual := d.PlannedWh/hours, d.ActualWh/hours + sp.Requested.Add(&planned) + sp.Actual.Add(&actual) + } + sp.First, _ = json.Marshal(map[string]any{ + "planned_wh": d.PlannedWh, "actual_wh": d.ActualWh, "ratio": d.Ratio, "replans": d.Replans, + }) + sp.Last = sp.First + return sp +} diff --git a/go/internal/state/control_journal.go b/go/internal/state/control_journal.go new file mode 100644 index 000000000..23cb1db96 --- /dev/null +++ b/go/internal/state/control_journal.go @@ -0,0 +1,284 @@ +package state + +import ( + "context" + "encoding/json" + "fmt" + "strings" +) + +// ControlSpan is one stretch of time in which a subject held one verdict: +// a device function's answer to "Are we in control?", a source's freshness, +// whether the site may dispatch, the operating mode, or one plan slot's +// delivery. The journal opens a span when the verdict changes and closes the +// one before it, so the spans of a subject tile its history. +type ControlSpan struct { + ID int64 `json:"id"` + // Subject names what held the verdict, e.g. "battery:sungrow", + // "source:easee", "site:dispatch", "site:mode" or "slot:battery". + Subject string `json:"subject"` + Driver string `json:"driver,omitempty"` + Kind string `json:"kind"` + Status string `json:"status"` + Reason string `json:"reason,omitempty"` + Severity string `json:"severity,omitempty"` + Mode string `json:"mode,omitempty"` + StartMs int64 `json:"start_ms"` + // EndMs is 0 while the verdict still holds. + EndMs int64 `json:"end_ms,omitempty"` + UpdatedMs int64 `json:"updated_ms"` + Samples int64 `json:"samples"` + // Requested and Actual summarise the power Core asked for and the + // device delivered over the span. A sample without a number adds + // nothing, so N may be below Samples. + Requested PowerStats `json:"requested_w"` + Actual PowerStats `json:"actual_w"` + // First and Last are the verdict as Core served it when the span + // opened and at its latest sample. + First json.RawMessage `json:"first,omitempty"` + Last json.RawMessage `json:"last,omitempty"` +} + +// PowerStats sums the watts seen over a span. +type PowerStats struct { + N int64 `json:"n"` + Min float64 `json:"min"` + Max float64 `json:"max"` + Sum float64 `json:"sum"` +} + +// Add takes one reading; nil adds nothing. +func (p *PowerStats) Add(w *float64) { + if w == nil { + return + } + if p.N == 0 || *w < p.Min { + p.Min = *w + } + if p.N == 0 || *w > p.Max { + p.Max = *w + } + p.N++ + p.Sum += *w +} + +// Mean is the average reading, false when there was none. +func (p PowerStats) Mean() (float64, bool) { + if p.N == 0 { + return 0, false + } + return p.Sum / float64(p.N), true +} + +// ControlSpanQuery picks the spans that overlap [SinceMs, UntilMs]. +type ControlSpanQuery struct { + SinceMs int64 + UntilMs int64 + // SubjectPrefix keeps subjects that start with it, e.g. "battery:". + SubjectPrefix string + Driver string + // Severities keeps these severities; empty keeps all. + Severities []string + // Limit bounds the answer; the newest spans come first. + Limit int +} + +const controlJournalColumns = `id, subject, driver, kind, status, reason, severity, mode, + start_ms, COALESCE(end_ms, 0), updated_ms, samples, + requested_n, COALESCE(requested_min, 0), COALESCE(requested_max, 0), requested_sum, + actual_n, COALESCE(actual_min, 0), COALESCE(actual_max, 0), actual_sum, + first_json, last_json` + +// SaveControlSpans writes new spans and updates known ones in one +// transaction. A span with ID 0 is inserted and gets its ID only after the +// commit, so a failed write leaves every span as it was for the next try. +func (s *Store) SaveControlSpans(ctx context.Context, spans []*ControlSpan) error { + if len(spans) == 0 { + return nil + } + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("control journal: begin: %w", err) + } + defer tx.Rollback() + ids := make([]int64, len(spans)) + for i, sp := range spans { + args := []any{ + sp.Subject, sp.Driver, sp.Kind, sp.Status, sp.Reason, sp.Severity, sp.Mode, + sp.StartMs, nullableMs(sp.EndMs), sp.UpdatedMs, sp.Samples, + sp.Requested.N, nullableStat(sp.Requested, sp.Requested.Min), nullableStat(sp.Requested, sp.Requested.Max), sp.Requested.Sum, + sp.Actual.N, nullableStat(sp.Actual, sp.Actual.Min), nullableStat(sp.Actual, sp.Actual.Max), sp.Actual.Sum, + jsonText(sp.First), jsonText(sp.Last), + } + if sp.ID == 0 { + res, err := tx.ExecContext(ctx, `INSERT INTO control_journal + (subject, driver, kind, status, reason, severity, mode, + start_ms, end_ms, updated_ms, samples, + requested_n, requested_min, requested_max, requested_sum, + actual_n, actual_min, actual_max, actual_sum, + first_json, last_json) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, args...) + if err != nil { + return fmt.Errorf("control journal: insert %s: %w", sp.Subject, err) + } + if ids[i], err = res.LastInsertId(); err != nil { + return fmt.Errorf("control journal: insert id: %w", err) + } + continue + } + if _, err := tx.ExecContext(ctx, `UPDATE control_journal SET + subject = ?, driver = ?, kind = ?, status = ?, reason = ?, severity = ?, mode = ?, + start_ms = ?, end_ms = ?, updated_ms = ?, samples = ?, + requested_n = ?, requested_min = ?, requested_max = ?, requested_sum = ?, + actual_n = ?, actual_min = ?, actual_max = ?, actual_sum = ?, + first_json = ?, last_json = ? + WHERE id = ?`, append(args, sp.ID)...); err != nil { + return fmt.Errorf("control journal: update %d: %w", sp.ID, err) + } + ids[i] = sp.ID + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("control journal: commit: %w", err) + } + for i, sp := range spans { + sp.ID = ids[i] + } + return nil +} + +// ControlSpans returns the spans that overlap the query's window, newest +// first, and whether more matched than the limit allowed. +func (s *Store) ControlSpans(ctx context.Context, q ControlSpanQuery) ([]ControlSpan, bool, error) { + if q.Limit <= 0 { + q.Limit = 200 + } + where := []string{"start_ms <= ?", "(end_ms IS NULL OR end_ms >= ?)"} + args := []any{q.UntilMs, q.SinceMs} + if q.SubjectPrefix != "" { + where = append(where, "substr(subject, 1, ?) = ?") + args = append(args, len(q.SubjectPrefix), q.SubjectPrefix) + } + if q.Driver != "" { + where = append(where, "driver = ?") + args = append(args, q.Driver) + } + if len(q.Severities) > 0 { + where = append(where, "severity IN (?"+strings.Repeat(", ?", len(q.Severities)-1)+")") + for _, sev := range q.Severities { + args = append(args, sev) + } + } + args = append(args, q.Limit+1) + rows, err := s.db.QueryContext(ctx, `SELECT `+controlJournalColumns+` + FROM control_journal WHERE `+strings.Join(where, " AND ")+` + ORDER BY start_ms DESC, id DESC LIMIT ?`, args...) + if err != nil { + return nil, false, fmt.Errorf("control journal: query: %w", err) + } + defer rows.Close() + out := []ControlSpan{} + for rows.Next() { + var sp ControlSpan + var first, last string + if err := rows.Scan(&sp.ID, &sp.Subject, &sp.Driver, &sp.Kind, &sp.Status, &sp.Reason, &sp.Severity, &sp.Mode, + &sp.StartMs, &sp.EndMs, &sp.UpdatedMs, &sp.Samples, + &sp.Requested.N, &sp.Requested.Min, &sp.Requested.Max, &sp.Requested.Sum, + &sp.Actual.N, &sp.Actual.Min, &sp.Actual.Max, &sp.Actual.Sum, + &first, &last); err != nil { + return nil, false, fmt.Errorf("control journal: scan: %w", err) + } + sp.First, sp.Last = rawJSON(first), rawJSON(last) + out = append(out, sp) + } + if err := rows.Err(); err != nil { + return nil, false, fmt.Errorf("control journal: rows: %w", err) + } + more := len(out) > q.Limit + if more { + out = out[:q.Limit] + } + return out, more, nil +} + +// CloseOpenControlSpans ends every span a previous run left open at the +// last moment that run saw it hold. Nothing is known after that, so the +// span does not run on through the restart. +func (s *Store) CloseOpenControlSpans(ctx context.Context) (int64, error) { + res, err := s.db.ExecContext(ctx, `UPDATE control_journal SET end_ms = updated_ms WHERE end_ms IS NULL`) + if err != nil { + return 0, fmt.Errorf("control journal: close open spans: %w", err) + } + return res.RowsAffected() +} + +// PruneControlSpans deletes spans that ended before beforeMs. +func (s *Store) PruneControlSpans(ctx context.Context, beforeMs int64) (int64, error) { + res, err := s.db.ExecContext(ctx, `DELETE FROM control_journal WHERE end_ms IS NOT NULL AND end_ms < ?`, beforeMs) + if err != nil { + return 0, fmt.Errorf("control journal: prune: %w", err) + } + return res.RowsAffected() +} + +// NotificationsBetween returns notification attempts in [sinceMs, untilMs], +// newest first, and whether more matched than the limit allowed. +func (s *Store) NotificationsBetween(ctx context.Context, sinceMs, untilMs int64, limit int) ([]NotificationEntry, bool, error) { + if limit <= 0 { + limit = 100 + } + rows, err := s.db.QueryContext(ctx, + `SELECT id, ts_ms, event_type, driver, title, body, priority, status, error + FROM notification_log + WHERE ts_ms >= ? AND ts_ms <= ? + ORDER BY ts_ms DESC, id DESC + LIMIT ?`, sinceMs, untilMs, limit+1) + if err != nil { + return nil, false, err + } + defer rows.Close() + out := []NotificationEntry{} + for rows.Next() { + var e NotificationEntry + if err := rows.Scan(&e.ID, &e.TsMs, &e.EventType, &e.Driver, &e.Title, + &e.Body, &e.Priority, &e.Status, &e.Error); err != nil { + return nil, false, err + } + out = append(out, e) + } + if err := rows.Err(); err != nil { + return nil, false, err + } + more := len(out) > limit + if more { + out = out[:limit] + } + return out, more, nil +} + +func nullableMs(ms int64) any { + if ms == 0 { + return nil + } + return ms +} + +func nullableStat(p PowerStats, v float64) any { + if p.N == 0 { + return nil + } + return v +} + +func jsonText(raw json.RawMessage) string { + if len(raw) == 0 || !json.Valid(raw) { + return "{}" + } + return string(raw) +} + +func rawJSON(text string) json.RawMessage { + if text == "" || text == "{}" { + return nil + } + return json.RawMessage(text) +} diff --git a/go/internal/state/control_journal_test.go b/go/internal/state/control_journal_test.go new file mode 100644 index 000000000..b1613e81b --- /dev/null +++ b/go/internal/state/control_journal_test.go @@ -0,0 +1,115 @@ +package state + +import ( + "context" + "encoding/json" + "testing" +) + +func jw(v float64) *float64 { return &v } + +func TestControlSpansRoundTripAndWindow(t *testing.T) { + st := openTestStore(t) + ctx := context.Background() + limited := &ControlSpan{Subject: "battery:sungrow", Driver: "sungrow", Kind: "battery", + Status: "limited", Reason: "power_below_target", Severity: "warning", Mode: "planner_arbitrage", + StartMs: 1_000, UpdatedMs: 1_000, Samples: 1, First: json.RawMessage(`{"requested_w":4200}`)} + limited.Requested.Add(jw(4200)) + limited.Actual.Add(jw(1200)) + if err := st.SaveControlSpans(ctx, []*ControlSpan{limited}); err != nil { + t.Fatal(err) + } + if limited.ID == 0 { + t.Fatal("insert did not set the id") + } + limited.Samples, limited.UpdatedMs, limited.EndMs = 3, 9_000, 10_000 + limited.Requested.Add(jw(4300)) + limited.Actual.Add(nil) + limited.Last = json.RawMessage(`{"requested_w":4300}`) + quiet := &ControlSpan{Subject: "source:myuplink", Driver: "myuplink", Kind: "source", + Status: "stale", Severity: "warning", StartMs: 20_000, UpdatedMs: 20_000, Samples: 1} + if err := st.SaveControlSpans(ctx, []*ControlSpan{limited, quiet}); err != nil { + t.Fatal(err) + } + + got, more, err := st.ControlSpans(ctx, ControlSpanQuery{SinceMs: 5_000, UntilMs: 6_000}) + if err != nil || more || len(got) != 1 { + t.Fatalf("window inside the span: got %+v more=%v err=%v", got, more, err) + } + sp := got[0] + if sp.EndMs != 10_000 || sp.Samples != 3 || sp.Requested.N != 2 || sp.Requested.Max != 4300 || sp.Actual.N != 1 || sp.Actual.Min != 1200 { + t.Fatalf("stats did not round-trip: %+v", sp) + } + if mean, ok := sp.Requested.Mean(); !ok || mean != 4250 { + t.Fatalf("requested mean = %v, %v", mean, ok) + } + if string(sp.First) != `{"requested_w":4200}` || string(sp.Last) != `{"requested_w":4300}` { + t.Fatalf("verdict text lost: first %s last %s", sp.First, sp.Last) + } + + // An open span overlaps every window after its start. + got, _, _ = st.ControlSpans(ctx, ControlSpanQuery{SinceMs: 50_000, UntilMs: 60_000}) + if len(got) != 1 || got[0].Subject != "source:myuplink" || got[0].EndMs != 0 { + t.Fatalf("open span: %+v", got) + } + got, _, _ = st.ControlSpans(ctx, ControlSpanQuery{SinceMs: 0, UntilMs: 60_000, SubjectPrefix: "battery:", Severities: []string{"warning"}}) + if len(got) != 1 || got[0].Subject != "battery:sungrow" { + t.Fatalf("prefix and severity filter: %+v", got) + } + got, more, _ = st.ControlSpans(ctx, ControlSpanQuery{SinceMs: 0, UntilMs: 60_000, Limit: 1}) + if len(got) != 1 || !more || got[0].Subject != "source:myuplink" { + t.Fatalf("limit keeps the newest and says more: %+v more=%v", got, more) + } +} + +// A restart leaves spans open. Nothing is known after the last update, so +// they end there rather than run on through the outage. +func TestCloseOpenControlSpansEndsAtLastUpdate(t *testing.T) { + st := openTestStore(t) + ctx := context.Background() + open := &ControlSpan{Subject: "site:dispatch", Kind: "dispatch", Status: "free", StartMs: 1_000, UpdatedMs: 7_000} + if err := st.SaveControlSpans(ctx, []*ControlSpan{open}); err != nil { + t.Fatal(err) + } + if n, err := st.CloseOpenControlSpans(ctx); err != nil || n != 1 { + t.Fatalf("closed %d, err %v", n, err) + } + got, _, _ := st.ControlSpans(ctx, ControlSpanQuery{SinceMs: 0, UntilMs: 10_000}) + if len(got) != 1 || got[0].EndMs != 7_000 { + t.Fatalf("span not closed at its last update: %+v", got) + } + if n, err := st.PruneControlSpans(ctx, 8_000); err != nil || n != 1 { + t.Fatalf("pruned %d, err %v", n, err) + } +} + +// A failed write leaves new spans without an id, so the next try inserts +// them again instead of updating rows that were never written. +func TestSaveControlSpansKeepsIDsOnFailure(t *testing.T) { + st := openTestStore(t) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + sp := &ControlSpan{Subject: "battery:sungrow", Status: "following", StartMs: 1, UpdatedMs: 1} + if err := st.SaveControlSpans(ctx, []*ControlSpan{sp}); err == nil { + t.Fatal("a cancelled write must fail") + } + if sp.ID != 0 { + t.Fatalf("id %d set by a failed write", sp.ID) + } + if err := st.SaveControlSpans(context.Background(), []*ControlSpan{sp}); err != nil || sp.ID == 0 { + t.Fatalf("retry: id %d err %v", sp.ID, err) + } +} + +func TestNotificationsBetween(t *testing.T) { + st := openTestStore(t) + for i, ts := range []int64{1_000, 5_000, 9_000} { + if err := st.RecordNotification(NotificationEntry{TsMs: ts, EventType: "control.not_following", Status: "sent", Priority: i}); err != nil { + t.Fatal(err) + } + } + got, more, err := st.NotificationsBetween(context.Background(), 2_000, 9_000, 1) + if err != nil || !more || len(got) != 1 || got[0].TsMs != 9_000 { + t.Fatalf("got %+v more=%v err=%v", got, more, err) + } +} diff --git a/go/internal/state/store.go b/go/internal/state/store.go index c41697867..b2cdd95cd 100644 --- a/go/internal/state/store.go +++ b/go/internal/state/store.go @@ -991,6 +991,34 @@ func (s *Store) migrate() error { error TEXT NOT NULL DEFAULT '' )`, `CREATE INDEX IF NOT EXISTS idx_notification_log_ts ON notification_log(ts_ms DESC)`, + // control_journal: each stretch a subject held one control verdict + // (control_journal.go). Small; kept in recovery snapshots. + `CREATE TABLE IF NOT EXISTS control_journal ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + subject TEXT NOT NULL, + driver TEXT NOT NULL DEFAULT '', + kind TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL, + reason TEXT NOT NULL DEFAULT '', + severity TEXT NOT NULL DEFAULT '', + mode TEXT NOT NULL DEFAULT '', + start_ms INTEGER NOT NULL, + end_ms INTEGER, + updated_ms INTEGER NOT NULL, + samples INTEGER NOT NULL DEFAULT 0, + requested_n INTEGER NOT NULL DEFAULT 0, + requested_min REAL, + requested_max REAL, + requested_sum REAL NOT NULL DEFAULT 0, + actual_n INTEGER NOT NULL DEFAULT 0, + actual_min REAL, + actual_max REAL, + actual_sum REAL NOT NULL DEFAULT 0, + first_json TEXT NOT NULL DEFAULT '{}', + last_json TEXT NOT NULL DEFAULT '{}' + ) STRICT`, + `CREATE INDEX IF NOT EXISTS idx_control_journal_start ON control_journal(start_ms DESC)`, + `CREATE INDEX IF NOT EXISTS idx_control_journal_open ON control_journal(end_ms) WHERE end_ms IS NULL`, // Independently installed Lua drivers. Content lives on disk; SQLite // records activation history and the exact previous artifact used for diff --git a/go/protocol/agents.go b/go/protocol/agents.go index ba82f253b..36b1d1092 100644 --- a/go/protocol/agents.go +++ b/go/protocol/agents.go @@ -419,18 +419,69 @@ type LogsGet struct { AfterMs int64 `cbor:"afterMs,omitempty"` } -// EventsGet asks for recent events, with the same cursors as LogsGet. +// EventsGet asks what the box recorded about control between AfterMs and +// BeforeMs, the last 24 hours when both are 0: each stretch in which a +// device function, a source, the site's dispatch or the mode held one +// verdict, each scored plan slot, and the notifications the box sent. Src +// keeps one source by its agent id; MinSeverity "warning" keeps warnings and +// alarms, "alarm" alarms only. N bounds the spans, newest first. type EventsGet struct { - N uint64 `cbor:"n,omitempty"` - BeforeMs int64 `cbor:"beforeMs,omitempty"` - AfterMs int64 `cbor:"afterMs,omitempty"` -} - -// Events is the answer to events.get: FTW Cloud's event bodies, oldest -// first. More says older events remain. + N uint64 `cbor:"n,omitempty"` + BeforeMs int64 `cbor:"beforeMs,omitempty"` + AfterMs int64 `cbor:"afterMs,omitempty"` + Src string `cbor:"src,omitempty"` + MinSeverity string `cbor:"minSeverity,omitempty"` +} + +// EventSpan is one stretch in which a subject held one control verdict. +// Kind is battery, ev, pv, v2x_charger or device for a controlled function, +// source for a driver's freshness, dispatch for whether the site may act, +// mode for the operating mode, or slot for a scored plan slot. Status, +// reason and severity are Core's own words, as in control_feedback. +type EventSpan struct { + Kind string `cbor:"kind"` + Src string `cbor:"src,omitempty"` + Status string `cbor:"status"` + Reason string `cbor:"reason,omitempty"` + Severity string `cbor:"severity,omitempty"` + Mode string `cbor:"mode,omitempty"` + StartMs int64 `cbor:"startMs"` + // EndMs is 0 while the verdict still holds. + EndMs int64 `cbor:"endMs"` + Samples int64 `cbor:"samples"` + // RequestedW and ActualW summarise the power Core asked for and the + // power measured over the span; absent when never known. + RequestedW *PowerRange `cbor:"requestedW,omitempty"` + ActualW *PowerRange `cbor:"actualW,omitempty"` + // Last holds the numbers of the latest sample by field name, such as + // sent_w, readback_w, offered_a, battery_soc or planned_wh. + Last map[string]float64 `cbor:"last,omitempty"` +} + +// PowerRange is the least, most and mean watts over a span. +type PowerRange struct { + Min float64 `cbor:"min"` + Max float64 `cbor:"max"` + Mean float64 `cbor:"mean"` +} + +// EventNotice is one notification the box sent or tried to send. +type EventNotice struct { + Ts int64 `cbor:"ts"` + Type string `cbor:"type"` + Src string `cbor:"src,omitempty"` + Status string `cbor:"status"` + Priority int `cbor:"priority,omitempty"` +} + +// Events is the answer to events.get: spans newest first, the notices in +// the same window, and More when either was cut. Events is kept for +// readers built before spans; the box leaves it empty. type Events struct { - Events []cbor.RawMessage `cbor:"events"` - More bool `cbor:"more"` + Events []cbor.RawMessage `cbor:"events"` + Spans []EventSpan `cbor:"spans,omitempty"` + Notices []EventNotice `cbor:"notices,omitempty"` + More bool `cbor:"more"` } // DriverState is one driver in drivers.get, by its opaque source id. From b867b4d30f1889867c11ed431597c475f0884fb5 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Sun, 8 Nov 2026 08:39:03 +0100 Subject: [PATCH 2/2] =?UTF-8?q?fix(control):=20journal=20review=20?= =?UTF-8?q?=E2=80=94=20one=20flush=20at=20a=20time,=20no=20NaN,=20fewer=20?= =?UTF-8?q?writes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Shutdown now stops the sampling goroutine before it closes and writes the spans, and flushes take turns, so a new span is inserted once. CloseAll stops later samples. A reading that is not a finite number adds nothing. An open span writes its numbers every 5 minutes, without first_json. When a batch fails, each span is tried alone, and one no write takes is dropped after 20 tries. Notices are filtered by driver in SQL; severity and limit are checked. Slot spans are the fleet's. Co-Authored-By: Claude Opus 5.5 --- docs/agents.md | 4 + go/cmd/ftw/control_journal.go | 56 ++++++--- go/cmd/ftw/control_journal_test.go | 10 +- go/cmd/ftw/main.go | 8 +- go/internal/api/api_control_journal.go | 53 +++++--- go/internal/api/api_control_journal_test.go | 11 +- go/internal/journal/journal.go | 90 +++++++++---- go/internal/journal/journal_test.go | 133 ++++++++++++++++++-- go/internal/journal/slot.go | 9 +- go/internal/state/control_journal.go | 23 ++-- go/internal/state/control_journal_test.go | 52 +++++++- 11 files changed, 357 insertions(+), 92 deletions(-) diff --git a/docs/agents.md b/docs/agents.md index e120d884f..3ff28dcf0 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -418,6 +418,10 @@ The registry's `reads` table names each read and the scope it needs. of the latest verdict by field name. Text leaves only as Core's own codes, never names. `minSeverity: "warning"` keeps warnings and alarms; `src` keeps one source. `events` stays empty; it is kept for older readers. + A source span is the driver's health as Core's watchdog sees it. A slot + span is the battery fleet's planned and delivered energy over one plan + slot, as average watts. A warning starts a span at once; a change to an + info verdict counts once it has held 15 s, from when it was first seen. - The box answers `diag.run` with `E_UNKNOWN_OP` until it builds it. `cars.get` carries no name, maker, VIN, identifier or source, and the redactor takes those out of `logs.get` and `config.get`; a box without the car store answers it `E_UNAVAILABLE`. `plan.get` names each diff --git a/go/cmd/ftw/control_journal.go b/go/cmd/ftw/control_journal.go index c7057d37f..ad5227757 100644 --- a/go/cmd/ftw/control_journal.go +++ b/go/cmd/ftw/control_journal.go @@ -19,15 +19,50 @@ const ( controlJournalKeep = 366 * 24 * time.Hour ) +// controlJournal runs the recorder on its own goroutine until shutdown. +type controlJournal struct { + rec *journal.Recorder + stop context.CancelFunc + done chan struct{} +} + // newControlJournal ends the spans a previous run left open, at the last -// moment that run saw them hold, and returns the recorder for this one. -func newControlJournal(st *state.Store) *journal.Recorder { +// moment that run saw them hold, and returns the journal for this one. +func newControlJournal(st *state.Store) *controlJournal { if n, err := st.CloseOpenControlSpans(context.Background()); err != nil { slog.Warn("control journal: could not close spans the last run left open", "err", err) } else if n > 0 { slog.Info("control journal: closed spans the last run left open", "spans", n) } - return journal.New(st) + return &controlJournal{rec: journal.New(st)} +} + +// start samples, writes and prunes until ctx ends or shutdown is called. +func (j *controlJournal) start(ctx context.Context, sample func(time.Time) []journal.Verdict, st *state.Store) { + ctx, j.stop = context.WithCancel(ctx) + j.done = make(chan struct{}) + go func() { + defer close(j.done) + runControlJournal(ctx, j.rec, sample, st) + }() +} + +// shutdown stops the sampling, then ends every open span now and writes +// them, so a clean stop leaves no span running through the time the box +// was down. +func (j *controlJournal) shutdown() { + if j.stop != nil { + j.stop() + select { + case <-j.done: + case <-time.After(2 * controlJournalEvery): + } + } + now := time.Now() + j.rec.CloseAll(now) + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + j.rec.Flush(ctx, now) } // runControlJournal samples, writes and prunes until ctx ends. @@ -57,16 +92,6 @@ func runControlJournal(ctx context.Context, rec *journal.Recorder, sample func(t } } -// stopControlJournal ends every open span now and writes them, so a clean -// stop leaves no span running through the time the box was down. -func stopControlJournal(rec *journal.Recorder) { - now := time.Now() - rec.CloseAll(now) - ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) - defer cancel() - rec.Flush(ctx, now) -} - // stateJournal serves events.get from the control journal and the // notification history. type stateJournal struct{ st *state.Store } @@ -78,7 +103,7 @@ func (j stateJournal) Journal(ctx context.Context, q appproto.JournalQuery) ([]a if err != nil { return nil, nil, false, err } - notices, noticesMore, err := j.st.NotificationsBetween(ctx, q.SinceMs, q.UntilMs, q.Limit) + notices, noticesMore, err := j.st.NotificationsBetween(ctx, q.SinceMs, q.UntilMs, q.Driver, q.Limit) if err != nil { return nil, nil, false, err } @@ -92,9 +117,6 @@ func (j stateJournal) Journal(ctx context.Context, q appproto.JournalQuery) ([]a } var ns []appproto.JournalNotice for _, n := range notices { - if q.Driver != "" && n.Driver != q.Driver { - continue - } ns = append(ns, appproto.JournalNotice{TsMs: n.TsMs, Type: n.EventType, Driver: n.Driver, Status: n.Status, Priority: n.Priority}) } return out, ns, more || noticesMore, nil diff --git a/go/cmd/ftw/control_journal_test.go b/go/cmd/ftw/control_journal_test.go index d77f85f7d..d142fd4dd 100644 --- a/go/cmd/ftw/control_journal_test.go +++ b/go/cmd/ftw/control_journal_test.go @@ -27,10 +27,14 @@ func openJournalStore(t *testing.T) *state.Store { // left open. func TestStopControlJournalClosesAndWrites(t *testing.T) { st := openJournalStore(t) - rec := newControlJournal(st) + j := newControlJournal(st) now := time.Now() - rec.Observe(now, []journal.Verdict{{Subject: "site:dispatch", Kind: "dispatch", Status: "free", Severity: "info"}}) - stopControlJournal(rec) + sample := func(time.Time) []journal.Verdict { + return []journal.Verdict{{Subject: "site:dispatch", Kind: "dispatch", Status: "free", Severity: "info"}} + } + j.rec.Observe(now, sample(now)) + j.start(context.Background(), sample, st) + j.shutdown() spans, _, err := st.ControlSpans(context.Background(), state.ControlSpanQuery{SinceMs: 0, UntilMs: now.Add(time.Minute).UnixMilli()}) if err != nil || len(spans) != 1 || spans[0].EndMs == 0 { t.Fatalf("spans %+v err %v", spans, err) diff --git a/go/cmd/ftw/main.go b/go/cmd/ftw/main.go index 77f505ebb..9d57e036e 100644 --- a/go/cmd/ftw/main.go +++ b/go/cmd/ftw/main.go @@ -623,7 +623,7 @@ func main() { // in control?", and each scored plan slot, so a warning can be explained // after it has passed (GET /api/control/journal, events.get). ctrlJournal := newControlJournal(st) - ctrl.SlotScored = func(d control.SlotDelivery) { ctrlJournal.Record(journal.SlotSpan(d)) } + ctrl.SlotScored = func(d control.SlotDelivery) { ctrlJournal.rec.Record(journal.SlotSpan(d)) } storedTrust, _ := st.LoadConfig(config.StateKeyForecastTrust) storedExport, _ := st.LoadConfig(config.StateKeyBatteryExport) storedSafetyK, _ := st.LoadConfig(config.StateKeySafetyK) @@ -2731,7 +2731,7 @@ func main() { // The control.not_following rule reads the box's own answer to "Are we // in control?", the one GET /api/status serves. notifSvc.SetControlReader(srv.ControlStates) - go runControlJournal(ctx, ctrlJournal, srv.JournalVerdicts, st) + ctrlJournal.start(ctx, srv.JournalVerdicts, st) // FuseReader: on each HealthTick the fuse_over_limit rule reads // the site meter's live per-phase currents from telemetry and // compares against cfg.Fuse.MaxAmps. Closes over cfg + cfgMu so @@ -2939,7 +2939,7 @@ func main() { slog.Info("shutting down") cloudLink.Bye(cloudwire.ByeShutdown, cloudByeWait) markNativeCleanExit(nativeRoot) - stopControlJournal(ctrlJournal) + ctrlJournal.shutdown() flushHistoryOnStop(st) if err := st.RecordEvent("shutdown"); err != nil { slog.Warn("failed to persist shutdown event", "err", err) @@ -2955,7 +2955,7 @@ func main() { // back "box out of reach" until the box is back. cloudLink.Bye(cloudwire.ByeUpdate, cloudByeWait) markNativeCleanExit(nativeRoot) - stopControlJournal(ctrlJournal) + ctrlJournal.shutdown() flushHistoryOnStop(st) if err := st.RecordEvent("restart"); err != nil { slog.Warn("failed to persist restart event", "err", err) diff --git a/go/internal/api/api_control_journal.go b/go/internal/api/api_control_journal.go index 4eaae6ecd..f0d07d4de 100644 --- a/go/internal/api/api_control_journal.go +++ b/go/internal/api/api_control_journal.go @@ -57,8 +57,11 @@ func (s *Server) JournalVerdicts(now time.Time) []journal.Verdict { return out } -// sourceVerdict is a driver's freshness as the journal keeps it. The error -// text changes from poll to poll, so it is detail, not the reason. +// sourceVerdict is a driver's health as Core's watchdog sees it: live, +// lagging after a failed read, offline once the watchdog gave up, or fault. +// It is the driver's state, not a reading's age, which control feedback +// judges per device. The error text changes from poll to poll, so it is +// detail, not the reason. func sourceVerdict(name string, h telemetry.DriverHealth) journal.Verdict { v := journal.Verdict{Subject: "source:" + name, Driver: name, Kind: "source", Status: "live", Severity: "info"} switch { @@ -136,22 +139,32 @@ func toJournalSpan(sp state.ControlSpan, nowMs int64) journalSpan { } // journalSeverities turns ?severity= into the severities to keep: a single -// level keeps it and those above it; a comma list keeps exactly those. -func journalSeverities(q string) []string { +// level keeps it and those above it; a comma list keeps exactly those. ok is +// false for a word that is not a severity. +func journalSeverities(q string) (keep []string, ok bool) { + known := map[string]bool{"info": true, "warning": true, "alarm": true} if q == "" { - return nil + return nil, true } if strings.Contains(q, ",") { - return strings.Split(q, ",") + for _, sev := range strings.Split(q, ",") { + sev = strings.TrimSpace(sev) + if !known[sev] { + return nil, false + } + keep = append(keep, sev) + } + return keep, true } switch q { case "info": - return nil + return nil, true case "warning": - return []string{"warning", "alarm"} - default: - return []string{q} + return []string{"warning", "alarm"}, true + case "alarm": + return []string{"alarm"}, true } + return nil, false } // GET /api/control/journal answers "what happened" over a window: each @@ -159,7 +172,9 @@ func journalSeverities(q string) []string { // one verdict, with the power asked for and delivered, each scored plan // slot, and the notifications the box sent. since and until are Unix ms // (the last 24 hours by default); subject keeps subjects with that prefix, -// e.g. battery: or source:easee; severity=warning keeps warnings and alarms. +// e.g. battery: or source:easee; driver keeps one driver's spans and +// notifications; severity=warning keeps warning and alarm spans. Notices are +// what the owner was told and are not filtered by severity. func (s *Server) handleControlJournal(w http.ResponseWriter, r *http.Request) { if s.deps.State == nil { writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "the control journal needs the state store"}) @@ -184,19 +199,27 @@ func (s *Server) handleControlJournal(w http.ResponseWriter, r *http.Request) { } limit := 200 if v := q.Get("limit"); v != "" { - if n, err := strconv.Atoi(v); err == nil && n > 0 { - limit = min(n, 1000) + n, err := strconv.Atoi(v) + if err != nil || n <= 0 { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "limit must be a positive number"}) + return } + limit = min(n, 1000) + } + severities, ok := journalSeverities(q.Get("severity")) + if !ok { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "severity must be info, warning or alarm, or a comma list of them"}) + return } spans, more, err := s.deps.State.ControlSpans(r.Context(), state.ControlSpanQuery{ SinceMs: sinceMs, UntilMs: untilMs, SubjectPrefix: q.Get("subject"), Driver: q.Get("driver"), - Severities: journalSeverities(q.Get("severity")), Limit: limit, + Severities: severities, Limit: limit, }) if err != nil { writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()}) return } - notices, noticesMore, err := s.deps.State.NotificationsBetween(r.Context(), sinceMs, untilMs, limit) + notices, noticesMore, err := s.deps.State.NotificationsBetween(r.Context(), sinceMs, untilMs, q.Get("driver"), limit) if err != nil { writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()}) return diff --git a/go/internal/api/api_control_journal_test.go b/go/internal/api/api_control_journal_test.go index d0d7d5e91..35ad0a7cd 100644 --- a/go/internal/api/api_control_journal_test.go +++ b/go/internal/api/api_control_journal_test.go @@ -118,8 +118,15 @@ func TestControlJournalAnswersAWindow(t *testing.T) { if len(spans) != 1 || spans[0].Status != "following" || spans[0].EndMs != nil { t.Fatalf("open span: %s", out["spans"]) } - if _, code := get("since=10&until=5"); code != http.StatusBadRequest { - t.Fatalf("reversed window: status %d", code) + for _, bad := range []string{"since=10&until=5", "severity=loud", "severity=warning,lod", "limit=-1", "limit=many"} { + if _, code := get(bad); code != http.StatusBadRequest { + t.Fatalf("%s: status %d", bad, code) + } + } + out, _ = get("since=" + msText(t0) + "&until=" + msText(t0+60*minute) + "&severity=" + "info,%20warning") + _ = json.Unmarshal(out["spans"], &spans) + if len(spans) != 2 { + t.Fatalf("a spaced comma list: %s", out["spans"]) } } diff --git a/go/internal/journal/journal.go b/go/internal/journal/journal.go index dbe8fd1ba..d7b22f490 100644 --- a/go/internal/journal/journal.go +++ b/go/internal/journal/journal.go @@ -42,8 +42,12 @@ type Store interface { const ( // StatsEvery is how often a span that still holds writes its running - // numbers. A changed verdict is written at the next flush. - StatsEvery = time.Minute + // numbers; a crash loses at most this much of them. A changed verdict is + // written at the next flush. + StatsEvery = 5 * time.Minute + // giveUpAfter is how many flushes in a row may fail to write one span + // alone before it is dropped, so one bad row cannot hold back the rest. + giveUpAfter = 20 // maxPending bounds what a failing database leaves in memory. maxPending = 5000 // failLogEvery spaces the warnings while writes keep failing. @@ -52,16 +56,23 @@ const ( // starts a span. Each new order makes a device "waiting" for a second // or two; that is not a change worth a span. A warning or an alarm // starts one at once, however short, and a settled change is dated - // from when it was first seen. + // from when it was first seen. The cost: info verdicts that keep + // changing faster than Settle never settle, and the span they would + // have replaced runs on. Settle = 15 * time.Second ) -// Recorder turns samples into spans and writes them. Observe and Flush run -// on one goroutine; Record may be called from any. +// Recorder turns samples into spans and writes them. Any goroutine may call +// it; flushes take turns, so two never write the same new span twice. type Recorder struct { - mu sync.Mutex - store Store - open map[string]*state.ControlSpan + mu sync.Mutex + flushMu sync.Mutex + store Store + // stopped is set by CloseAll: later samples start nothing. + stopped bool + // failures counts flushes in a row that could not write a span alone. + failures map[*state.ControlSpan]int + open map[string]*state.ControlSpan // candidate holds a change to an info verdict until it settles. candidate map[string]*state.ControlSpan // pending holds spans with changes not yet written, each with the @@ -79,7 +90,7 @@ type Recorder struct { // New returns a recorder that writes to store. func New(store Store) *Recorder { return &Recorder{store: store, open: map[string]*state.ControlSpan{}, candidate: map[string]*state.ControlSpan{}, - pending: map[*state.ControlSpan]uint64{}} + pending: map[*state.ControlSpan]uint64{}, failures: map[*state.ControlSpan]int{}} } // Observe takes one sample of every subject. A subject missing from the @@ -91,6 +102,9 @@ func (r *Recorder) Observe(now time.Time, verdicts []Verdict) { nowMs := now.UnixMilli() r.mu.Lock() defer r.mu.Unlock() + if r.stopped { + return + } seen := map[string]bool{} for _, v := range verdicts { if v.Subject == "" || seen[v.Subject] { @@ -172,13 +186,15 @@ func (r *Recorder) Record(sp state.ControlSpan) { r.markLocked(&sp) } -// CloseAll ends every open span now, for a clean shutdown. +// CloseAll ends every open span now, for a clean shutdown. Samples after it +// start nothing. func (r *Recorder) CloseAll(now time.Time) { if r == nil { return } r.mu.Lock() defer r.mu.Unlock() + r.stopped = true for subject, sp := range r.open { r.closeLocked(sp, now.UnixMilli()) delete(r.open, subject) @@ -187,11 +203,15 @@ func (r *Recorder) CloseAll(now time.Time) { } // Flush writes changed spans, and once a StatsEvery the running numbers of -// spans that still hold. A failed write keeps everything for the next one. +// spans that still hold. A failed write keeps everything for the next one; +// when the whole batch fails, each span is tried alone so one bad row +// cannot hold back the rest. func (r *Recorder) Flush(ctx context.Context, now time.Time) { if r == nil || r.store == nil { return } + r.flushMu.Lock() + defer r.flushMu.Unlock() type entry struct { orig *state.ControlSpan version uint64 @@ -224,33 +244,55 @@ func (r *Recorder) Flush(ctx context.Context, now time.Time) { return } + saved := make([]bool, len(copies)) err := r.store.SaveControlSpans(ctx, copies) + if err == nil { + for i := range saved { + saved[i] = true + } + } else { + for i, c := range copies { + saved[i] = r.store.SaveControlSpans(ctx, []*state.ControlSpan{c}) == nil + } + } r.mu.Lock() defer r.mu.Unlock() - if err != nil { + failed := 0 + for i, e := range batch { + if !saved[i] { + failed++ + r.failures[e.orig]++ + if r.failures[e.orig] >= giveUpAfter && e.orig.EndMs != 0 { + slog.Error("control journal: dropping a span no write would take", + "subject", e.orig.Subject, "start_ms", e.orig.StartMs, "tries", r.failures[e.orig], "err", err) + delete(r.pending, e.orig) + delete(r.failures, e.orig) + r.dropped++ + } + continue + } + delete(r.failures, e.orig) + if e.orig.ID == 0 { + e.orig.ID = copies[i].ID + } + if v, ok := r.pending[e.orig]; ok && v == e.version { + delete(r.pending, e.orig) + } + } + if failed > 0 { if r.failingAt.IsZero() { r.failingAt = now } if now.Sub(r.loggedAt) >= failLogEvery { r.loggedAt = now slog.Warn("control journal: write failed; keeping spans for the next try", - "pending", len(r.pending), "since", r.failingAt, "err", err) + "failed", failed, "pending", len(r.pending), "since", r.failingAt, "err", err) } - return - } - if !r.failingAt.IsZero() { + } else if !r.failingAt.IsZero() { slog.Info("control journal: writes recovered", "failing_since", r.failingAt, "dropped", r.dropped) r.failingAt, r.loggedAt, r.dropped = time.Time{}, time.Time{}, 0 } - for i, e := range batch { - if e.orig.ID == 0 { - e.orig.ID = copies[i].ID - } - if v, ok := r.pending[e.orig]; ok && v == e.version { - delete(r.pending, e.orig) - } - } r.compactLocked() } diff --git a/go/internal/journal/journal_test.go b/go/internal/journal/journal_test.go index 27cdbabc8..1f541ebd4 100644 --- a/go/internal/journal/journal_test.go +++ b/go/internal/journal/journal_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "sync" "testing" "time" @@ -139,34 +140,38 @@ func TestRecorderKeepsSpansThroughFailedWrites(t *testing.T) { } } -// A span that holds writes its running numbers once a minute, so a crash -// loses at most a minute of them. -func TestRecorderWritesHoldingSpansOnceAMinute(t *testing.T) { +// A span that holds writes its running numbers once a StatsEvery, so a +// crash loses at most that much of them. +func TestRecorderWritesHoldingSpansOnceAStatsEvery(t *testing.T) { st := &fakeStore{} r := New(st) t0 := time.Unix(20_000, 0) r.Observe(t0, []Verdict{battery("following", "plan", "info", 500, 500)}) r.Flush(context.Background(), t0) - for s := 5; s <= 55; s += 5 { - r.Observe(t0.Add(time.Duration(s)*time.Second), []Verdict{battery("following", "plan", "info", 500, 500)}) - r.Flush(context.Background(), t0.Add(time.Duration(s)*time.Second)) + step := 5 * time.Second + n := int(StatsEvery / step) + for i := 1; i < n; i++ { + at := t0.Add(time.Duration(i) * step) + r.Observe(at, []Verdict{battery("following", "plan", "info", 500, 500)}) + r.Flush(context.Background(), at) } if got := st.bySubject("battery:sungrow")[0].Samples; got != 1 { - t.Fatalf("samples written before a minute: %d", got) + t.Fatalf("samples written before StatsEvery: %d", got) } - r.Observe(t0.Add(65*time.Second), []Verdict{battery("following", "plan", "info", 500, 500)}) - r.Flush(context.Background(), t0.Add(65*time.Second)) - if got := st.bySubject("battery:sungrow")[0]; got.Samples != 13 || got.UpdatedMs != t0.Add(65*time.Second).UnixMilli() { - t.Fatalf("after a minute: %+v", got) + at := t0.Add(StatsEvery) + r.Observe(at, []Verdict{battery("following", "plan", "info", 500, 500)}) + r.Flush(context.Background(), at) + if got := st.bySubject("battery:sungrow")[0]; got.Samples != int64(n+1) || got.UpdatedMs != at.UnixMilli() { + t.Fatalf("after StatsEvery: %+v", got) } } func TestRecorderRecordsAFinishedSpan(t *testing.T) { st := &fakeStore{} r := New(st) - r.Record(state.ControlSpan{Subject: "slot:battery", Kind: "slot", Status: "under_delivered", Severity: "warning", StartMs: 1, EndMs: 900_001, UpdatedMs: 900_001}) + r.Record(state.ControlSpan{Subject: "slot:batteries", Kind: "slot", Status: "under_delivered", Severity: "warning", StartMs: 1, EndMs: 900_001, UpdatedMs: 900_001}) r.Flush(context.Background(), time.Unix(1_000, 0)) - if got := st.bySubject("slot:battery"); len(got) != 1 || got[0].Status != "under_delivered" { + if got := st.bySubject("slot:batteries"); len(got) != 1 || got[0].Status != "under_delivered" { t.Fatalf("slot span: %+v", got) } } @@ -197,3 +202,105 @@ func TestRecorderSettlesInfoChangesButNotWarnings(t *testing.T) { t.Fatalf("a 3 s warning must be its own span, ended when the recovery was first seen: %+v", got) } } + +// blockingStore holds each write until released, and fails spans whose +// subject it refuses. +type blockingStore struct { + mu sync.Mutex + inserts map[string]int + nextID int64 + gate chan struct{} + refuse string +} + +func (b *blockingStore) SaveControlSpans(_ context.Context, spans []*state.ControlSpan) error { + if b.gate != nil { + <-b.gate + } + b.mu.Lock() + defer b.mu.Unlock() + for _, sp := range spans { + if sp.Subject == b.refuse { + return errors.New("NOT NULL constraint failed") + } + } + for _, sp := range spans { + if sp.ID == 0 { + b.nextID++ + sp.ID = b.nextID + b.inserts[sp.Subject]++ + } + } + return nil +} + +// A stop flushes from another goroutine while the journal's own flush may +// be writing. A new span is inserted once, and samples after the stop +// start nothing. +func TestRecorderFlushesTakeTurns(t *testing.T) { + st := &blockingStore{inserts: map[string]int{}, gate: make(chan struct{})} + r := New(st) + t0 := time.Unix(40_000, 0) + r.Observe(t0, []Verdict{battery("following", "plan", "info", 500, 500)}) + var wg sync.WaitGroup + for range 2 { + wg.Add(1) + go func() { defer wg.Done(); r.Flush(context.Background(), t0) }() + } + time.Sleep(20 * time.Millisecond) + close(st.gate) + wg.Wait() + if n := st.inserts["battery:sungrow"]; n != 1 { + t.Fatalf("span inserted %d times", n) + } + r.CloseAll(t0.Add(time.Second)) + r.Observe(t0.Add(2*time.Second), []Verdict{battery("following", "plan", "info", 500, 500)}) + if open := r.Open(); len(open) != 0 { + t.Fatalf("a sample after the stop opened %+v", open) + } +} + +// One span no write will take does not hold back the others, and is +// dropped once it has failed long enough. +func TestRecorderIsolatesASpanNoWriteTakes(t *testing.T) { + st := &blockingStore{inserts: map[string]int{}, refuse: "source:bad"} + r := New(st) + t0 := time.Unix(50_000, 0) + r.Observe(t0, []Verdict{ + {Subject: "source:bad", Kind: "source", Status: "live", Severity: "info"}, + battery("following", "plan", "info", 500, 500), + }) + r.Observe(t0.Add(5*time.Second), []Verdict{battery("following", "plan", "info", 500, 500)}) + for i := range giveUpAfter + 1 { + r.Flush(context.Background(), t0.Add(time.Duration(10+i)*time.Second)) + } + if st.inserts["battery:sungrow"] != 1 { + t.Fatalf("the good span was held back: %+v", st.inserts) + } + r.mu.Lock() + left := len(r.pending) + r.mu.Unlock() + if left != 0 { + t.Fatalf("%d spans still pending after giving up", left) + } +} + +// A warning that arrives while an info change waits starts its span at +// once, and the change that never settled leaves no trace. +func TestRecorderWarningDuringACandidate(t *testing.T) { + st := &fakeStore{} + r := New(st) + t0 := time.Unix(60_000, 0) + tick := func(s int, v Verdict) { + at := t0.Add(time.Duration(s) * time.Second) + r.Observe(at, []Verdict{v}) + r.Flush(context.Background(), at) + } + tick(0, battery("following", "plan", "info", 1000, 1000)) + tick(5, battery("waiting", "waiting_response", "info", 1000, 0)) + tick(10, battery("not_following", "no_power_response", "warning", 1000, 0)) + got := st.bySubject("battery:sungrow") + if len(got) != 2 || got[0].Status != "following" || got[0].EndMs != t0.Add(10*time.Second).UnixMilli() || got[1].Status != "not_following" { + t.Fatalf("spans %+v", got) + } +} diff --git a/go/internal/journal/slot.go b/go/internal/journal/slot.go index c4f23974b..0dc4ebdc6 100644 --- a/go/internal/journal/slot.go +++ b/go/internal/journal/slot.go @@ -7,16 +7,17 @@ import ( "github.com/srcfl/ftw/go/internal/state" ) -// SlotSpan is a scored plan slot as the journal keeps it: the plan's -// average battery power over the slot as asked for, the delivered average -// as actual, and the energy behind both. +// SlotSpan is a scored plan slot as the journal keeps it, for the battery +// fleet as a whole: the plan's energy over the slot as an average power +// asked for, the fleet's delivered energy as an average actual, and the +// energy behind both. The mode is the one in force when the slot ended. func SlotSpan(d control.SlotDelivery) state.ControlSpan { severity := "info" if d.Verdict == "under_delivered" || d.Verdict == "sign_mismatch" { severity = "warning" } sp := state.ControlSpan{ - Subject: "slot:battery", Kind: "slot", Status: d.Verdict, Severity: severity, Mode: string(d.Mode), + Subject: "slot:batteries", Kind: "slot", Status: d.Verdict, Severity: severity, Mode: string(d.Mode), StartMs: d.SlotStart.UnixMilli(), EndMs: d.SlotEnd.UnixMilli(), UpdatedMs: d.SlotEnd.UnixMilli(), Samples: 1, } if hours := d.SlotEnd.Sub(d.SlotStart).Hours(); hours > 0 { diff --git a/go/internal/state/control_journal.go b/go/internal/state/control_journal.go index 23cb1db96..ab8054964 100644 --- a/go/internal/state/control_journal.go +++ b/go/internal/state/control_journal.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "math" "strings" ) @@ -15,7 +16,7 @@ import ( type ControlSpan struct { ID int64 `json:"id"` // Subject names what held the verdict, e.g. "battery:sungrow", - // "source:easee", "site:dispatch", "site:mode" or "slot:battery". + // "source:easee", "site:dispatch", "site:mode" or "slot:batteries". Subject string `json:"subject"` Driver string `json:"driver,omitempty"` Kind string `json:"kind"` @@ -47,9 +48,10 @@ type PowerStats struct { Sum float64 `json:"sum"` } -// Add takes one reading; nil adds nothing. +// Add takes one reading; nil or a value that is not a finite number adds +// nothing. func (p *PowerStats) Add(w *float64) { - if w == nil { + if w == nil || math.IsNaN(*w) || math.IsInf(*w, 0) { return } if p.N == 0 || *w < p.Min { @@ -126,13 +128,15 @@ func (s *Store) SaveControlSpans(ctx context.Context, spans []*ControlSpan) erro } continue } + // first_json never changes after the insert, so an update leaves it. + update := append(append([]any{}, args[:len(args)-2]...), args[len(args)-1], sp.ID) if _, err := tx.ExecContext(ctx, `UPDATE control_journal SET subject = ?, driver = ?, kind = ?, status = ?, reason = ?, severity = ?, mode = ?, start_ms = ?, end_ms = ?, updated_ms = ?, samples = ?, requested_n = ?, requested_min = ?, requested_max = ?, requested_sum = ?, actual_n = ?, actual_min = ?, actual_max = ?, actual_sum = ?, - first_json = ?, last_json = ? - WHERE id = ?`, append(args, sp.ID)...); err != nil { + last_json = ? + WHERE id = ?`, update...); err != nil { return fmt.Errorf("control journal: update %d: %w", sp.ID, err) } ids[i] = sp.ID @@ -221,17 +225,18 @@ func (s *Store) PruneControlSpans(ctx context.Context, beforeMs int64) (int64, e } // NotificationsBetween returns notification attempts in [sinceMs, untilMs], -// newest first, and whether more matched than the limit allowed. -func (s *Store) NotificationsBetween(ctx context.Context, sinceMs, untilMs int64, limit int) ([]NotificationEntry, bool, error) { +// newest first, for one driver when driver is set, and whether more matched +// than the limit allowed. +func (s *Store) NotificationsBetween(ctx context.Context, sinceMs, untilMs int64, driver string, limit int) ([]NotificationEntry, bool, error) { if limit <= 0 { limit = 100 } rows, err := s.db.QueryContext(ctx, `SELECT id, ts_ms, event_type, driver, title, body, priority, status, error FROM notification_log - WHERE ts_ms >= ? AND ts_ms <= ? + WHERE ts_ms >= ? AND ts_ms <= ? AND (? = '' OR driver = ?) ORDER BY ts_ms DESC, id DESC - LIMIT ?`, sinceMs, untilMs, limit+1) + LIMIT ?`, sinceMs, untilMs, driver, driver, limit+1) if err != nil { return nil, false, err } diff --git a/go/internal/state/control_journal_test.go b/go/internal/state/control_journal_test.go index b1613e81b..8004beea1 100644 --- a/go/internal/state/control_journal_test.go +++ b/go/internal/state/control_journal_test.go @@ -3,6 +3,7 @@ package state import ( "context" "encoding/json" + "math" "testing" ) @@ -108,8 +109,57 @@ func TestNotificationsBetween(t *testing.T) { t.Fatal(err) } } - got, more, err := st.NotificationsBetween(context.Background(), 2_000, 9_000, 1) + got, more, err := st.NotificationsBetween(context.Background(), 2_000, 9_000, "", 1) if err != nil || !more || len(got) != 1 || got[0].TsMs != 9_000 { t.Fatalf("got %+v more=%v err=%v", got, more, err) } } + +// A reading that is not a number must not reach the table: one NaN sum +// failed the NOT NULL check and held back every later write. +func TestPowerStatsIgnoresNonFiniteReadings(t *testing.T) { + st := openTestStore(t) + sp := &ControlSpan{Subject: "battery:sungrow", Status: "following", StartMs: 1, UpdatedMs: 1} + nan, inf := math.NaN(), math.Inf(1) + sp.Requested.Add(&nan) + sp.Requested.Add(&inf) + sp.Requested.Add(jw(500)) + if err := st.SaveControlSpans(context.Background(), []*ControlSpan{sp}); err != nil { + t.Fatal(err) + } + if sp.Requested.N != 1 || sp.Requested.Sum != 500 { + t.Fatalf("stats %+v", sp.Requested) + } +} + +// An update keeps the verdict the span opened with. +func TestSaveControlSpansKeepsFirstOnUpdate(t *testing.T) { + st := openTestStore(t) + ctx := context.Background() + sp := &ControlSpan{Subject: "battery:sungrow", Status: "limited", StartMs: 1, UpdatedMs: 1, + First: json.RawMessage(`{"requested_w":4298}`), Last: json.RawMessage(`{"requested_w":4298}`)} + if err := st.SaveControlSpans(ctx, []*ControlSpan{sp}); err != nil { + t.Fatal(err) + } + sp.First, sp.Last = nil, json.RawMessage(`{"requested_w":4162}`) + if err := st.SaveControlSpans(ctx, []*ControlSpan{sp}); err != nil { + t.Fatal(err) + } + got, _, _ := st.ControlSpans(ctx, ControlSpanQuery{SinceMs: 0, UntilMs: 10}) + if string(got[0].First) != `{"requested_w":4298}` || string(got[0].Last) != `{"requested_w":4162}` { + t.Fatalf("first %s last %s", got[0].First, got[0].Last) + } +} + +func TestNotificationsBetweenForOneDriver(t *testing.T) { + st := openTestStore(t) + for _, d := range []string{"sungrow", "easee", "sungrow"} { + if err := st.RecordNotification(NotificationEntry{TsMs: 100, EventType: "control.not_following", Driver: d, Status: "sent"}); err != nil { + t.Fatal(err) + } + } + got, more, err := st.NotificationsBetween(context.Background(), 0, 1_000, "sungrow", 1) + if err != nil || len(got) != 1 || !more || got[0].Driver != "sungrow" { + t.Fatalf("got %+v more %v err %v", got, more, err) + } +}