diff --git a/.changeset/dispatch-send-timeout.md b/.changeset/dispatch-send-timeout.md new file mode 100644 index 00000000..ca71cf3d --- /dev/null +++ b/.changeset/dispatch-send-timeout.md @@ -0,0 +1,5 @@ +--- +"ftw": patch +--- + +Bound every control-loop driver command with a 2 s deadline. Battery dispatch, PV curtailment and loadpoint sends previously waited on `Registry.Send` with the long-lived loop context, so one driver wedged mid-poll could stall dispatch to every other driver indefinitely. A timed-out send is logged and recovery is left to the existing watchdog/staleness paths, matching how autonomous default commands are already bounded. diff --git a/go/cmd/ftw/dispatch_send.go b/go/cmd/ftw/dispatch_send.go new file mode 100644 index 00000000..59cb1286 --- /dev/null +++ b/go/cmd/ftw/dispatch_send.go @@ -0,0 +1,33 @@ +package main + +import ( + "context" + "log/slog" + "time" +) + +// driverCommandTimeout bounds one control-loop command delivery. Registry.Send +// blocks until the driver's runLoop processes the payload, so without a +// deadline a single driver wedged mid-poll would stall dispatch to every +// other driver for the rest of the tick (and beyond). 2 s matches +// driverDefaultTimeout: the same driver-side work is being waited on, and +// anything slower is indistinguishable from a stuck driver at the default +// 2 s control interval. +const driverCommandTimeout = 2 * time.Second + +// commandSender is the slice of drivers.Registry the dispatch path needs. +type commandSender interface { + Send(ctx context.Context, name string, payload []byte) error +} + +// sendDriverCommand delivers one dispatch payload with a bounded deadline. +// Errors (including deadline expiry) are logged, not returned: the control +// loop treats a failed send like any other driver hiccup — the watchdog and +// staleness paths own recovery. +func sendDriverCommand(ctx context.Context, reg commandSender, name, kind string, payload []byte) { + cmdCtx, cancel := context.WithTimeout(ctx, driverCommandTimeout) + defer cancel() + if err := reg.Send(cmdCtx, name, payload); err != nil { + slog.Warn(kind+" send", "name", name, "timeout", driverCommandTimeout, "err", err) + } +} diff --git a/go/cmd/ftw/dispatch_send_test.go b/go/cmd/ftw/dispatch_send_test.go new file mode 100644 index 00000000..ac9f2723 --- /dev/null +++ b/go/cmd/ftw/dispatch_send_test.go @@ -0,0 +1,71 @@ +package main + +import ( + "context" + "errors" + "testing" + "time" +) + +// blockingSender blocks every Send until its context expires, like a driver +// wedged mid-poll whose cmdCh is full. +type blockingSender struct { + sawDeadline chan bool +} + +func (b *blockingSender) Send(ctx context.Context, name string, payload []byte) error { + _, ok := ctx.Deadline() + b.sawDeadline <- ok + <-ctx.Done() + return ctx.Err() +} + +func TestSendDriverCommandBoundsBlockedDriver(t *testing.T) { + s := &blockingSender{sawDeadline: make(chan bool, 1)} + done := make(chan struct{}) + start := time.Now() + go func() { + sendDriverCommand(context.Background(), s, "stuck", "driver", []byte(`{"action":"battery","power_w":0}`)) + close(done) + }() + select { + case <-done: + case <-time.After(driverCommandTimeout + 2*time.Second): + t.Fatalf("sendDriverCommand did not return; blocked driver stalls dispatch") + } + if elapsed := time.Since(start); elapsed < driverCommandTimeout/2 { + t.Fatalf("returned after %v, before the driver had its full window", elapsed) + } + if !<-s.sawDeadline { + t.Fatal("Send received a context without a deadline") + } +} + +type recordingSender struct { + name string + payload []byte + err error +} + +func (r *recordingSender) Send(ctx context.Context, name string, payload []byte) error { + r.name = name + r.payload = append([]byte(nil), payload...) + return r.err +} + +func TestSendDriverCommandPassesThrough(t *testing.T) { + s := &recordingSender{} + sendDriverCommand(context.Background(), s, "bat1", "driver", []byte(`{"action":"battery","power_w":1500}`)) + if s.name != "bat1" { + t.Fatalf("sent to %q, want bat1", s.name) + } + if string(s.payload) != `{"action":"battery","power_w":1500}` { + t.Fatalf("payload = %s", s.payload) + } +} + +func TestSendDriverCommandLogsButSwallowsErrors(t *testing.T) { + s := &recordingSender{err: errors.New("boom")} + // Must not panic or propagate: recovery belongs to watchdog/staleness. + sendDriverCommand(context.Background(), s, "bat1", "driver", []byte(`{}`)) +} diff --git a/go/cmd/ftw/main.go b/go/cmd/ftw/main.go index 6b989e5e..a8d067b3 100644 --- a/go/cmd/ftw/main.go +++ b/go/cmd/ftw/main.go @@ -1504,7 +1504,16 @@ func main() { RequestActive: reqActive, }, true } - lpController = loadpoint.NewController(lpMgr, planAdapter, telAdapter, reg.Send) + // Bound every loadpoint send the same way the battery/curtail + // dispatch is bounded: the controller passes long-lived (or + // background) contexts, and an unbounded Registry.Send on a + // wedged driver would stall the EV tick. + lpSend := func(ctx context.Context, name string, payload []byte) error { + cmdCtx, cancel := context.WithTimeout(ctx, driverCommandTimeout) + defer cancel() + return reg.Send(cmdCtx, name, payload) + } + lpController = loadpoint.NewController(lpMgr, planAdapter, telAdapter, lpSend) // Wire the site fuse so the per-phase EV clamp and the // phase-split derivation can use the actual site voltage and // breaker rating instead of hard-coding 230 V × 16 A. @@ -2637,9 +2646,7 @@ func main() { continue } payload, _ := json.Marshal(map[string]any{"action": "battery", "power_w": t.TargetW}) - if err := reg.Send(ctx, t.Driver, payload); err != nil { - slog.Warn("driver send", "name", t.Driver, "err", err) - } + sendDriverCommand(ctx, reg, t.Driver, "driver", payload) } // ---- PV curtailment dispatch ---- @@ -2665,9 +2672,7 @@ func main() { "action": "curtail_disable", }) } - if err := reg.Send(ctx, c.Driver, payload); err != nil { - slog.Warn("pv curtail send", "name", c.Driver, "err", err) - } + sendDriverCommand(ctx, reg, c.Driver, "pv curtail", payload) } // LP dispatch ran at the top of this tick — see the