fix(control): bound every driver command with a deadline

.changeset/bound-driver-command-deadline.md

@@ -0,0 +1,5 @@
+---
+"ftw": patch
+---
+
+One slow driver can no longer stall dispatch for the whole site. Every command the control tick sends now carries its own deadline, derived from `site.control_interval_s` (half the interval, capped at 2 s and floored at 250 ms). Before this, the tick handed the driver registry the process-lifetime context, which has no deadline, and the driver goroutine runs the device call inline — so a cloud driver waiting on an HTTP or OAuth request that never answered held up every other battery on the site, and the reactive fuse guard with them. A command that runs out of time is logged at Warn with the driver name, so a chronically slow driver shows up in the log instead of quietly eating tick cadence.

go/cmd/ftw/driver_command_deadline.go

@@ -0,0 +1,76 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"time"
+)
+
+// driverCommandSender is the part of *drivers.Registry the dispatch
+// sends use. Narrow, so the deadline behaviour can be exercised without
+// a live driver host.
+type driverCommandSender interface {
+	Send(ctx context.Context, name string, payload []byte) error
+}
+
+// minDriverCommandTimeout is the floor under driverCommandTimeout. A
+// deadline shorter than a LAN Modbus write would fail every command and
+// leave the site frozen at its last output, which is worse than the
+// stall the deadline exists to prevent.
+const minDriverCommandTimeout = 250 * time.Millisecond
+
+// driverCommandTimeout bounds one dispatch command, derived from the
+// operator's control interval.
+//
+// The hazard: Registry.Send blocks until the driver goroutine has run
+// driver_command, and that goroutine performs the device I/O inline. A
+// cloud driver waiting on an HTTP or OAuth call that never answers
+// therefore holds the whole control tick — every other battery on the
+// site, and the reactive fuse guard that protects the main breaker with
+// them. The context the tick carries is the process-lifetime one, which
+// has no deadline, so nothing ever breaks that wait. sendDriverDefault
+// already bounds its call for exactly this reason, and SendDefault's own
+// doc comment names the deadlock; the dispatch path runs every control
+// interval and needs the same bound.
+//
+// Half the interval: a wedged driver then costs its own command and
+// still leaves the rest of the tick — the remaining drivers, the fuse
+// guard, the state save — inside the interval. Capped at
+// driverDefaultTimeout because a dispatch command must never be given
+// longer to give up than the safety default that has to reach the same
+// driver afterwards.
+func driverCommandTimeout(controlInterval time.Duration) time.Duration {
+	timeout := controlInterval / 2
+	if timeout > driverDefaultTimeout {
+		timeout = driverDefaultTimeout
+	}
+	if timeout < minDriverCommandTimeout {
+		timeout = minDriverCommandTimeout
+	}
+	return timeout
+}
+
+// sendDriverCommand sends one dispatch command under its own deadline.
+// A timeout is logged at Warn with the driver name so a chronically slow
+// driver shows up in the log instead of quietly eating tick cadence;
+// kind names the dispatch path ("driver send", "pv curtail send").
+//
+// A timeout is deliberately not recorded as a driver failure. The driver
+// goroutine serialises polls and commands, so a driver wedged inside
+// driver_command stops emitting telemetry as well, and the staleness
+// watchdog already walks it to its autonomous default mode. Counting the
+// timeout separately would double-book the same fault and could push a
+// merely slow cloud driver out of control on one bad round trip.
+func sendDriverCommand(ctx context.Context, reg driverCommandSender, kind, name string, payload []byte, timeout time.Duration) {
+	cmdCtx, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
+	err := reg.Send(cmdCtx, name, payload)
+	switch {
+	case err == nil:
+	case errors.Is(err, context.DeadlineExceeded):
+		slog.Warn(kind+" timed out", "name", name, "timeout", timeout)
+	default:
+		slog.Warn(kind, "name", name, "err", err)
+	}
+}

go/cmd/ftw/driver_command_deadline_test.go

@@ -0,0 +1,224 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"log/slog"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+type stubCall struct {
+	name        string
+	payload     string
+	hadDeadline bool
+}
+
+// stubSender stands in for *drivers.Registry. handler decides how a
+// given driver behaves — returning immediately, failing, or hanging the
+// way a cloud driver stuck in an unanswered HTTP call does.
+type stubSender struct {
+	mu      sync.Mutex
+	calls   []stubCall
+	handler func(ctx context.Context, name string) error
+}
+
+func (s *stubSender) Send(ctx context.Context, name string, payload []byte) error {
+	_, hadDeadline := ctx.Deadline()
+	s.mu.Lock()
+	s.calls = append(s.calls, stubCall{name: name, payload: string(payload), hadDeadline: hadDeadline})
+	s.mu.Unlock()
+	if s.handler == nil {
+		return nil
+	}
+	return s.handler(ctx, name)
+}
+
+func (s *stubSender) recorded() []stubCall {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return append([]stubCall(nil), s.calls...)
+}
+
+// captureWarnings redirects the default logger for one test so the
+// timeout warning — the operator's only signal that a driver is slow —
+// can be asserted on.
+func captureWarnings(t *testing.T) *bytes.Buffer {
+	t.Helper()
+	var buf bytes.Buffer
+	previous := slog.Default()
+	slog.SetDefault(slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelWarn})))
+	t.Cleanup(func() { slog.SetDefault(previous) })
+	return &buf
+}
+
+func TestDriverCommandTimeoutStaysUnderControlInterval(t *testing.T) {
+	cases := []struct {
+		name     string
+		interval time.Duration
+		want     time.Duration
+	}{
+		{"one second tick", time.Second, 500 * time.Millisecond},
+		{"default two second tick", 2 * time.Second, time.Second},
+		{"five second tick caps at the default timeout", 5 * time.Second, driverDefaultTimeout},
+		{"slow tick still caps", 30 * time.Second, driverDefaultTimeout},
+		{"no interval falls back to the floor", 0, minDriverCommandTimeout},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			got := driverCommandTimeout(tc.interval)
+			if got != tc.want {
+				t.Fatalf("driverCommandTimeout(%s) = %s, want %s", tc.interval, got, tc.want)
+			}
+			if got > driverDefaultTimeout {
+				t.Errorf("timeout %s exceeds the default-mode timeout %s", got, driverDefaultTimeout)
+			}
+			if got < minDriverCommandTimeout {
+				t.Errorf("timeout %s is below the floor %s", got, minDriverCommandTimeout)
+			}
+			// The point of the bound: a wedged driver must cost its
+			// own command, not the tick it rides in.
+			if tc.interval >= minDriverCommandTimeout*2 && got >= tc.interval {
+				t.Errorf("timeout %s does not fit inside the %s control interval", got, tc.interval)
+			}
+		})
+	}
+}
+
+// The bug this fix closes: dispatch handed Registry.Send the
+// process-lifetime context, which has no deadline, so a driver wedged in
+// its device I/O held the control tick for the whole site.
+func TestSendDriverCommandStopsWaitingOnWedgedDriver(t *testing.T) {
+	logs := captureWarnings(t)
+	const timeout = 50 * time.Millisecond
+	wedged := make(chan struct{})
+	defer close(wedged)
+	sender := &stubSender{handler: func(ctx context.Context, name string) error {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-wedged:
+			return nil
+		}
+	}}
+
+	done := make(chan time.Duration, 1)
+	go func() {
+		start := time.Now()
+		// context.Background() stands in for the process-lifetime
+		// context the control loop carries.
+		sendDriverCommand(context.Background(), sender, "driver send", "cloud-battery", []byte(`{"action":"battery","power_w":-2000}`), timeout)
+		done <- time.Since(start)
+	}()
+
+	select {
+	case elapsed := <-done:
+		if elapsed < timeout {
+			t.Fatalf("returned after %s, before the %s deadline", elapsed, timeout)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("sendDriverCommand never returned: a wedged driver is holding the control tick")
+	}
+
+	if got := logs.String(); !strings.Contains(got, "driver send timed out") || !strings.Contains(got, "cloud-battery") {
+		t.Errorf("timeout log does not name the slow driver: %q", got)
+	}
+}
+
+// One wedged driver must cost its own command, not the batteries queued
+// behind it — the fuse guard dispatches through this same loop.
+func TestWedgedDriverDoesNotStarveLaterDispatchTargets(t *testing.T) {
+	captureWarnings(t)
+	const timeout = 50 * time.Millisecond
+	wedged := make(chan struct{})
+	defer close(wedged)
+	sender := &stubSender{handler: func(ctx context.Context, name string) error {
+		if name != "cloud-battery" {
+			return nil
+		}
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-wedged:
+			return nil
+		}
+	}}
+
+	targets := []string{"cloud-battery", "ferroamp", "sungrow"}
+	done := make(chan time.Duration, 1)
+	go func() {
+		start := time.Now()
+		for _, name := range targets {
+			sendDriverCommand(context.Background(), sender, "driver send", name, []byte(`{"action":"battery","power_w":0}`), timeout)
+		}
+		done <- time.Since(start)
+	}()
+
+	var elapsed time.Duration
+	select {
+	case elapsed = <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("dispatch never finished: the wedged driver blocked the drivers behind it")
+	}
+	// Only the wedged driver may spend its full deadline.
+	if elapsed > 2*timeout {
+		t.Errorf("dispatch of %d drivers took %s, want roughly one %s timeout", len(targets), elapsed, timeout)
+	}
+	calls := sender.recorded()
+	if len(calls) != len(targets) {
+		t.Fatalf("sent %d commands, want %d", len(calls), len(targets))
+	}
+	for i, call := range calls {
+		if call.name != targets[i] {
+			t.Errorf("command %d went to %q, want %q", i, call.name, targets[i])
+		}
+	}
+}
+
+func TestSendDriverCommandLeavesHealthyDriverUntouched(t *testing.T) {
+	logs := captureWarnings(t)
+	sender := &stubSender{}
+	payload := []byte(`{"action":"curtail","power_w":3000}`)
+
+	start := time.Now()
+	sendDriverCommand(context.Background(), sender, "pv curtail send", "solaredge", payload, time.Second)
+	if elapsed := time.Since(start); elapsed > 100*time.Millisecond {
+		t.Fatalf("a healthy command took %s", elapsed)
+	}
+
+	calls := sender.recorded()
+	if len(calls) != 1 {
+		t.Fatalf("sent %d commands, want 1", len(calls))
+	}
+	if calls[0].name != "solaredge" || calls[0].payload != string(payload) {
+		t.Errorf("command = %+v, want the curtail payload for solaredge", calls[0])
+	}
+	if !calls[0].hadDeadline {
+		t.Error("driver received a context with no deadline")
+	}
+	if got := logs.String(); got != "" {
+		t.Errorf("healthy command logged %q, want silence", got)
+	}
+}
+
+// A driver that refuses a command still reports as before: the deadline
+// must not swallow real driver errors.
+func TestSendDriverCommandLogsDriverError(t *testing.T) {
+	logs := captureWarnings(t)
+	sender := &stubSender{handler: func(ctx context.Context, name string) error {
+		return errors.New("modbus write refused")
+	}}
+
+	sendDriverCommand(context.Background(), sender, "driver send", "ferroamp", []byte(`{"action":"battery","power_w":0}`), time.Second)
+
+	got := logs.String()
+	if !strings.Contains(got, "modbus write refused") || !strings.Contains(got, "ferroamp") {
+		t.Errorf("driver error not logged: %q", got)
+	}
+	if strings.Contains(got, "timed out") {
+		t.Errorf("a refusal was reported as a timeout: %q", got)
+	}
+}

go/cmd/ftw/main.go

@@ -2331,6 +2331,9 @@ func main() {
 	// the configreload watcher updates those fields directly, so a
 	// startup snapshot here would go stale on the first hot-reload.
 	dtS := float64(cfg.Site.ControlIntervalS)
+	// Every dispatch command carries its own deadline — see
+	// driverCommandTimeout for the stall it bounds.
+	driverCmdTimeout := driverCommandTimeout(controlInterval)
 
 	// Graceful shutdown
 	sigc := make(chan os.Signal, 1)
@@ -2649,9 +2652,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, "driver send", t.Driver, payload, driverCmdTimeout)
 			}
 
 			// ---- PV curtailment dispatch ----
@@ -2677,9 +2678,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, "pv curtail send", c.Driver, payload, driverCmdTimeout)
 			}
 
 			// LP dispatch ran at the top of this tick — see the