fix(control): bound control-loop driver sends with a per-command deadline

.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.

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)
+	}
+}

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(`{}`))
+}

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