fix(control): keep autonomous defaults ahead of stale commands

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

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:

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)

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

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{}),
 	}