From 328e08769e2eb9cb6acdb18897990da4b62544c4 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Tue, 4 Aug 2026 15:34:10 +0200 Subject: [PATCH] fix(drivers): prioritize autonomous defaults over stale commands --- .changeset/stale-driver-command-safety.md | 5 + go/internal/drivers/registry.go | 92 ++++++--- .../drivers/registry_command_deadline_test.go | 13 +- .../drivers/registry_command_safety_test.go | 179 ++++++++++++++++++ go/internal/drivers/registry_restart_test.go | 14 +- 5 files changed, 269 insertions(+), 34 deletions(-) create mode 100644 .changeset/stale-driver-command-safety.md create mode 100644 go/internal/drivers/registry_command_safety_test.go diff --git a/.changeset/stale-driver-command-safety.md b/.changeset/stale-driver-command-safety.md new file mode 100644 index 00000000..1937176f --- /dev/null +++ b/.changeset/stale-driver-command-safety.md @@ -0,0 +1,5 @@ +--- +"ftw": patch +--- + +Autonomous driver defaults now bypass queued stale commands, and expired commands are discarded before they reach hardware. diff --git a/go/internal/drivers/registry.go b/go/internal/drivers/registry.go index 55f858c8..85f7a6b3 100644 --- a/go/internal/drivers/registry.go +++ b/go/internal/drivers/registry.go @@ -244,9 +244,10 @@ type runningDriver struct { shutdownMu sync.Mutex shutdownDefaultErr error // Poll loop coordination - cmdCh chan driverCmd - stop chan bool - done chan struct{} + cmdCh chan driverCmd + defaultCh chan driverCmd + stop chan bool + done chan struct{} } func (rd *runningDriver) controlStatus() DriverControlStatus { @@ -601,6 +602,7 @@ func (r *Registry) add(ctx context.Context, cfg config.Driver, startupDefault bo lifecycleCtx: lifecycleCtx, lifecycleCancel: lifecycleCancel, cmdCh: make(chan driverCmd, 8), + defaultCh: make(chan driverCmd, 1), stop: make(chan bool, 1), done: make(chan struct{}), } @@ -756,7 +758,48 @@ func (r *Registry) runLoop(rd *runningDriver) { } return commandOutcome } + handleDefault := func(cmd driverCmd) { + cmdCtx := cmd.ctx + if cmdCtx == nil { + cmdCtx = ctx + } + // Once a safety default has been accepted into the dedicated queue, + // the caller's deadline must not turn it into a no-op while the + // driver is finishing an older command. Preserve the caller context + // when the request starts in time, but give an already-expired queued + // default a fresh bounded attempt; recovery will retry on failure. + var cancel context.CancelFunc + if cmdCtx.Err() != nil { + cmdCtx, cancel = context.WithTimeout(context.Background(), defaultRecoveryTimeout) + } + err := r.defaultDriver(cmdCtx, rd, "host_request") + if cancel != nil { + cancel() + } + if err == nil { + clearLease() + rd.markDefaultConfirmed() + clearRecoveryTimer() + r.clearRecoveryRequired(rd.cfg.Name, rd) + } else { + scheduleRecovery() + } + if cmd.result != nil { + cmd.result <- err + } + } for { + // A pending autonomous default must get service before any queued + // control. SendDefault marks the generation blocked before enqueue, + // so a normal command selected in the next select is discarded by + // the controlIsBlocked check below even if both channels become ready + // at the same time. + select { + case cmd := <-rd.defaultCh: + handleDefault(cmd) + continue + default: + } select { case skipDefault := <-rd.stop: if !skipDefault { @@ -791,6 +834,8 @@ func (r *Registry) runLoop(rd *runningDriver) { _ = rd.env.TCP.Close() } return + case cmd := <-rd.defaultCh: + handleDefault(cmd) case cmd := <-rd.cmdCh: var err error cmdCtx := cmd.ctx @@ -841,16 +886,6 @@ func (r *Registry) runLoop(rd *runningDriver) { } } finishCommand() - case "default": - err = r.defaultDriver(cmdCtx, rd, "host_request") - if err == nil { - clearLease() - rd.markDefaultConfirmed() - clearRecoveryTimer() - r.clearRecoveryRequired(rd.cfg.Name, rd) - } else { - scheduleRecovery() - } } if cmd.state != nil { cmd.state.finish(err) @@ -1136,12 +1171,10 @@ func (r *Registry) SendWithGeneration(ctx context.Context, name string, payload } } -// SendDefault sends the default/watchdog command to a driver. Symmetric -// with Send: both the channel-push and the result-wait honour ctx. A -// driver whose cmdCh is full (because its goroutine is slow / stuck mid -// I/O) would otherwise block the caller forever; the watchdog-fallback -// path runs on every dispatch tick, so an unblocked send into a wedged -// driver deadlocks the entire control loop. +// SendDefault sends the default/watchdog command to a driver. Defaults use a +// dedicated one-slot queue so stale normal commands cannot prevent the +// autonomous path from being accepted. Once accepted, the generation stays +// blocked until the default succeeds or the recovery timer retries it. func (r *Registry) SendDefault(ctx context.Context, name string) error { if ctx == nil { ctx = context.Background() @@ -1152,11 +1185,26 @@ func (r *Registry) SendDefault(ctx context.Context, name string) error { if !ok { return fmt.Errorf("driver %q not found", name) } + if err := ctx.Err(); err != nil { + return err + } + // Close the control window before enqueueing. A command that races this + // transition may still enter cmdCh, but runLoop will discard it without + // crossing the driver boundary. + rd.markDefaultRecoveryPending() resCh := make(chan error, 1) + cmd := driverCmd{kind: "default", ctx: ctx, result: resCh} select { - case rd.cmdCh <- driverCmd{kind: "default", ctx: ctx, result: resCh}: - case <-ctx.Done(): - return ctx.Err() + case rd.defaultCh <- cmd: + default: + // If another default is already pending, wait only until this + // caller's deadline. The queued default remains the durable safety + // request and the recovery timer will retry it if needed. + select { + case rd.defaultCh <- cmd: + case <-ctx.Done(): + return ctx.Err() + } } select { case err := <-resCh: diff --git a/go/internal/drivers/registry_command_deadline_test.go b/go/internal/drivers/registry_command_deadline_test.go index 46643687..b2cd6213 100644 --- a/go/internal/drivers/registry_command_deadline_test.go +++ b/go/internal/drivers/registry_command_deadline_test.go @@ -52,12 +52,13 @@ func TestSendReturnsAtDeadlineWhileDriverIsWedged(t *testing.T) { release: make(chan struct{}), } rd := &runningDriver{ - driver: rt, - env: rt.env, - cfg: config.Driver{Name: "d1"}, - cmdCh: make(chan driverCmd, 1), - stop: make(chan bool, 1), - done: make(chan struct{}), + driver: rt, + env: rt.env, + cfg: config.Driver{Name: "d1"}, + cmdCh: make(chan driverCmd, 1), + defaultCh: make(chan driverCmd, 1), + stop: make(chan bool, 1), + done: make(chan struct{}), } r.rec["d1"] = rd go r.runLoop(rd) diff --git a/go/internal/drivers/registry_command_safety_test.go b/go/internal/drivers/registry_command_safety_test.go new file mode 100644 index 00000000..2c4f6166 --- /dev/null +++ b/go/internal/drivers/registry_command_safety_test.go @@ -0,0 +1,179 @@ +package drivers + +import ( + "context" + "errors" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/srcfl/ftw/go/internal/config" + "github.com/srcfl/ftw/go/internal/telemetry" +) + +// staleQueueRuntime models a legacy Lua driver_command that has crossed the +// driver boundary, may have changed hardware, and then blocks without +// watching its context. The registry must not let commands that timed out +// while waiting in cmdCh reach this runtime after recovery. +type staleQueueRuntime struct { + env *HostEnv + entered chan struct{} + release chan struct{} + enteredOnce sync.Once + releaseOnce sync.Once + defaulted chan struct{} + defaultedOnce sync.Once + defaultFailures int32 + defaultAttempts atomic.Int32 + eventsMu sync.Mutex + events []string +} + +func (r *staleQueueRuntime) Init(ctx context.Context, configJSON []byte) error { return nil } +func (r *staleQueueRuntime) Poll(ctx context.Context) (time.Duration, error) { + return time.Hour, nil +} +func (r *staleQueueRuntime) Command(ctx context.Context, cmdJSON []byte) error { + payload := string(cmdJSON) + r.eventsMu.Lock() + r.events = append(r.events, "command:"+payload) + r.eventsMu.Unlock() + first := false + r.enteredOnce.Do(func() { + first = true + close(r.entered) + }) + if first { + <-r.release + } + return nil +} +func (r *staleQueueRuntime) DefaultMode(ctx context.Context) error { + r.eventsMu.Lock() + r.events = append(r.events, "default") + r.eventsMu.Unlock() + attempt := r.defaultAttempts.Add(1) + if attempt <= r.defaultFailures { + return errors.New("simulated default failure") + } + r.defaultedOnce.Do(func() { close(r.defaulted) }) + return nil +} +func (r *staleQueueRuntime) Cleanup(ctx context.Context) error { return nil } +func (r *staleQueueRuntime) Env() *HostEnv { return r.env } + +func (r *staleQueueRuntime) releaseCommand() { + r.releaseOnce.Do(func() { close(r.release) }) +} + +func (r *staleQueueRuntime) eventSnapshot() []string { + r.eventsMu.Lock() + defer r.eventsMu.Unlock() + return append([]string(nil), r.events...) +} + +func TestRegistryDefaultBypassesStaleCommandQueue(t *testing.T) { + tel := telemetry.NewStore() + r := NewRegistry(tel) + rt := &staleQueueRuntime{ + env: NewHostEnv("d1", tel), + entered: make(chan struct{}), + release: make(chan struct{}), + defaulted: make(chan struct{}), + defaultFailures: 1, + } + lifecycleCtx, lifecycleCancel := context.WithCancel(context.Background()) + rd := &runningDriver{ + driver: rt, + env: rt.env, + cfg: config.Driver{Name: "d1"}, + generation: 1, + lifecycleCtx: lifecycleCtx, + lifecycleCancel: lifecycleCancel, + defaultConfirmed: true, + cmdCh: make(chan driverCmd, 8), + defaultCh: make(chan driverCmd, 1), + stop: make(chan bool, 1), + done: make(chan struct{}), + } + r.mu.Lock() + r.rec["d1"] = rd + r.mu.Unlock() + go r.runLoop(rd) + t.Cleanup(func() { + rt.releaseCommand() + if _, ok := r.ControlStatus("d1"); ok { + r.remove("d1", true) + } + }) + + activeDone := make(chan error, 1) + activeCtx, activeCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer activeCancel() + go func() { + activeDone <- r.Send(activeCtx, "d1", []byte(`{"action":"active"}`)) + }() + select { + case <-rt.entered: + case <-time.After(time.Second): + t.Fatal("blocking command did not reach the driver") + } + + // The blocked command leaves all eight normal queue slots available for + // callers whose short contexts will expire before runLoop can dequeue. + for i := 0; i < cap(rd.cmdCh); i++ { + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + err := r.Send(ctx, "d1", []byte(`{"action":"stale"}`)) + cancel() + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("stale Send %d = %v, want deadline exceeded", i, err) + } + } + if got := len(rd.cmdCh); got != cap(rd.cmdCh) { + t.Fatalf("normal queue length = %d, want %d", got, cap(rd.cmdCh)) + } + + // This is the watchdog failure from the merged P1: the caller expires + // while the active legacy command still owns runLoop. The request must be + // accepted outside cmdCh and remain durable after this call returns. + defaultCtx, defaultCancel := context.WithTimeout(context.Background(), 30*time.Millisecond) + defaultErr := r.SendDefault(defaultCtx, "d1") + defaultCancel() + if !errors.Is(defaultErr, context.DeadlineExceeded) { + t.Fatalf("SendDefault = %v, want caller deadline while command is blocked", defaultErr) + } + status, ok := r.ControlStatus("d1") + if !ok || !status.Blocked || !status.RecoveryPending || status.DefaultConfirmed { + t.Fatalf("status after accepted default = %+v, running=%v", status, ok) + } + + rt.releaseCommand() + select { + case err := <-activeDone: + if err != nil { + t.Fatalf("active Send = %v, want success after release", err) + } + case <-time.After(time.Second): + t.Fatal("blocking command did not finish after release") + } + select { + case <-rt.defaulted: + case <-time.After(2 * time.Second): + t.Fatal("accepted default was not retried to success") + } + if attempts := rt.defaultAttempts.Load(); attempts < 2 { + t.Fatalf("default attempts = %d, want failed attempt plus recovery retry", attempts) + } + + status, ok = r.ControlStatus("d1") + if !ok || status.Blocked || !status.DefaultConfirmed || status.RecoveryPending { + t.Fatalf("status after default recovery = %+v, running=%v", status, ok) + } + for _, event := range rt.eventSnapshot() { + if strings.HasPrefix(event, "command:") && !strings.HasSuffix(event, `{"action":"active"}`) { + t.Fatalf("stale command crossed driver boundary: events=%v", rt.eventSnapshot()) + } + } +} diff --git a/go/internal/drivers/registry_restart_test.go b/go/internal/drivers/registry_restart_test.go index ed0865f1..afab5cd5 100644 --- a/go/internal/drivers/registry_restart_test.go +++ b/go/internal/drivers/registry_restart_test.go @@ -130,12 +130,13 @@ func TestSendDefaultPassesCallerContextToRuntime(t *testing.T) { entered: make(chan struct{}), } rd := &runningDriver{ - driver: rt, - env: rt.env, - cfg: config.Driver{Name: "d1"}, - cmdCh: make(chan driverCmd, 1), - stop: make(chan bool, 1), - done: make(chan struct{}), + driver: rt, + env: rt.env, + cfg: config.Driver{Name: "d1"}, + cmdCh: make(chan driverCmd, 1), + defaultCh: make(chan driverCmd, 1), + stop: make(chan bool, 1), + done: make(chan struct{}), } r.rec["d1"] = rd go r.runLoop(rd) @@ -171,6 +172,7 @@ func TestRegistryCancelAfterCommandStartedRestoresDefault(t *testing.T) { lifecycleCtx: lifecycleCtx, lifecycleCancel: lifecycleCancel, cmdCh: make(chan driverCmd, 1), + defaultCh: make(chan driverCmd, 1), stop: make(chan bool, 1), done: make(chan struct{}), }