From a2f7941ff4d4b6f9b520d78a5e83e5031a59ebe7 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Sun, 16 Aug 2026 20:03:09 +0200 Subject: [PATCH] fix(drivers): preempt blocked commands for defaults --- .changeset/preempt-blocked-driver-defaults.md | 5 + go/internal/drivers/lua.go | 22 +- go/internal/drivers/registry.go | 99 ++++-- .../drivers/registry_command_deadline_test.go | 13 +- .../drivers/registry_ev_command_owner_test.go | 30 +- go/internal/drivers/registry_restart_test.go | 14 +- .../registry_runtime_preemption_test.go | 286 ++++++++++++++++++ 7 files changed, 419 insertions(+), 50 deletions(-) create mode 100644 .changeset/preempt-blocked-driver-defaults.md create mode 100644 go/internal/drivers/registry_runtime_preemption_test.go diff --git a/.changeset/preempt-blocked-driver-defaults.md b/.changeset/preempt-blocked-driver-defaults.md new file mode 100644 index 000000000..eb94b87d5 --- /dev/null +++ b/.changeset/preempt-blocked-driver-defaults.md @@ -0,0 +1,5 @@ +--- +"ftw": patch +--- + +Core now cancels blocked Lua work, read-only HTTP, or sleep before it runs the driver's autonomous default. Mutating HTTP stays ordered until the host transport returns, and a dedicated default queue keeps the safety request ahead of stale control commands without calling one driver in parallel. diff --git a/go/internal/drivers/lua.go b/go/internal/drivers/lua.go index d8ab865a6..46804c334 100644 --- a/go/internal/drivers/lua.go +++ b/go/internal/drivers/lua.go @@ -609,6 +609,13 @@ func luaReturnError(name string, ret lua.LValue) error { // ---- host.* API exposed to Lua ---- +func luaCallContext(L *lua.LState) context.Context { + if ctx := L.Context(); ctx != nil { + return ctx + } + return context.Background() +} + func registerHost(L *lua.LState, env *HostEnv) { host := L.NewTable() @@ -720,7 +727,12 @@ func registerHost(L *lua.LState, env *HostEnv) { return 1 } if ms > 0 { - time.Sleep(time.Duration(ms) * time.Millisecond) + timer := time.NewTimer(time.Duration(ms) * time.Millisecond) + defer timer.Stop() + select { + case <-timer.C: + case <-luaCallContext(L).Done(): + } } return 0 })) @@ -1280,7 +1292,7 @@ func registerHost(L *lua.LState, env *HostEnv) { L.Push(lua.LString("http: " + reason)) return 2 } - req, err := net_http.NewRequest("GET", url, nil) + req, err := net_http.NewRequestWithContext(luaCallContext(L), "GET", url, nil) if err != nil { L.Push(lua.LNil) L.Push(lua.LString(err.Error())) @@ -1332,6 +1344,10 @@ func registerHost(L *lua.LState, env *HostEnv) { return 2 } payload := L.CheckString(2) + // Do not cancel a mutating request when the Lua command context ends. + // Once the device may have received the write, a following default must + // stay behind this request in the registry actor. The host client's + // 15-second timeout still bounds transport failure. req, err := net_http.NewRequest("POST", url, strings.NewReader(payload)) if err != nil { L.Push(lua.LNil) @@ -1393,6 +1409,8 @@ func registerHost(L *lua.LState, env *HostEnv) { return 2 } payload := L.CheckString(2) + // Keep the same ordering rule as POST: the registry actor must not send + // a default while this older mutating request can still finish normally. req, err := net_http.NewRequest("PATCH", url, strings.NewReader(payload)) if err != nil { L.Push(lua.LNil) diff --git a/go/internal/drivers/registry.go b/go/internal/drivers/registry.go index a16cce588..fe993b2a9 100644 --- a/go/internal/drivers/registry.go +++ b/go/internal/drivers/registry.go @@ -263,9 +263,10 @@ type runningDriver struct { evPauseCycleID uint64 evPausePending bool // 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 { @@ -657,6 +658,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{}), } @@ -820,7 +822,45 @@ func (r *Registry) runLoop(rd *runningDriver) { } return commandOutcome } + handleDefault := func(cmd driverCmd) { + invalidateCommandSequence() + cmdCtx := cmd.ctx + if cmdCtx == nil { + cmdCtx = ctx + } + // Once accepted, a safety default must survive its caller timing out + // behind an older command. A fresh bounded attempt keeps that request + // durable without letting the actor block forever. + 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 { + // Service an accepted autonomous default before normal control. The + // blocked status still rejects a normal command if both queues become + // ready between this check and the main select. + select { + case cmd := <-rd.defaultCh: + handleDefault(cmd) + continue + default: + } select { case skipDefault := <-rd.stop: if !skipDefault { @@ -856,6 +896,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 reportOutcome := cmd.outcome != nil @@ -925,6 +967,15 @@ func (r *Registry) runLoop(rd *runningDriver) { invalidateCommandSequence() } commandCtx, finishCommand := rd.beginCommand(cmdCtx) + // SendDefault may close the control window after the earlier queue + // check but before this actor installs activeCancel. Recheck once the + // cancel hook exists; after this point a racing default cancels the + // context passed to the runtime. + if rd.controlIsBlocked() { + finishCommand() + err = ErrControlBlocked + break + } if rd.policy != nil && rd.policy.IsControlV2() { var result DriverCommandResultV1 var leaseExpiresAt time.Time @@ -961,17 +1012,6 @@ func (r *Registry) runLoop(rd *runningDriver) { rd.evPauseRevision = rd.commandRevision rd.evPauseCycleID = cmd.cycleID } - case "default": - invalidateCommandSequence() - err = r.defaultDriver(cmdCtx, rd, "host_request") - if err == nil { - clearLease() - rd.markDefaultConfirmed() - clearRecoveryTimer() - r.clearRecoveryRequired(rd.cfg.Name, rd) - } else { - scheduleRecovery() - } } if reportOutcome { cmd.outcome(err) @@ -1279,12 +1319,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() @@ -1295,11 +1333,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. Canceling the active Lua call + // lets a context-aware host operation return to this same actor before the + // queued default runs; the registry never calls one driver in parallel. + rd.markDefaultRecoveryPending() + rd.cancelActiveCommand() 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: + // Another accepted default already supplies the durable safety request. + // Wait only for room or this caller's deadline. + 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 1e8579ae3..afd671a2e 100644 --- a/go/internal/drivers/registry_command_deadline_test.go +++ b/go/internal/drivers/registry_command_deadline_test.go @@ -54,12 +54,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_ev_command_owner_test.go b/go/internal/drivers/registry_ev_command_owner_test.go index 11352954d..610aa4bf8 100644 --- a/go/internal/drivers/registry_ev_command_owner_test.go +++ b/go/internal/drivers/registry_ev_command_owner_test.go @@ -96,6 +96,7 @@ func newEVCommandOwnerRegistry(t *testing.T, blocked bool) (*Registry, *runningD lifecycleCtx: lifecycleCtx, lifecycleCancel: lifecycleCancel, cmdCh: make(chan driverCmd, 8), + defaultCh: make(chan driverCmd, 1), stop: make(chan bool, 1), done: make(chan struct{}), } @@ -128,26 +129,29 @@ func TestDefaultBoundaryInvalidatesEarlierEVPause(t *testing.T) { defaultDone := make(chan error, 1) go func() { defaultDone <- r.SendDefault(ctx, "charger") }() deadline := time.Now().Add(2 * time.Second) - for len(rd.cmdCh) == 0 && time.Now().Before(deadline) { + for len(rd.defaultCh) == 0 && time.Now().Before(deadline) { time.Sleep(time.Millisecond) } - if len(rd.cmdCh) == 0 { + if len(rd.defaultCh) == 0 { close(runtime.pauseRelease) t.Fatal("default was not queued behind the parked pause") } close(runtime.pauseRelease) - for name, done := range map[string]<-chan error{ - "pause": pauseDone, - "default": defaultDone, - } { - select { - case err := <-done: - if err != nil { - t.Fatalf("%s: %v", name, err) - } - case <-time.After(2 * time.Second): - t.Fatalf("%s did not finish", name) + select { + case err := <-pauseDone: + if !errors.Is(err, ErrCommandMayHaveRun) || !errors.Is(err, context.Canceled) { + t.Fatalf("pause = %v, want command-may-have-run plus canceled", err) + } + case <-time.After(2 * time.Second): + t.Fatal("pause did not finish") + } + select { + case err := <-defaultDone: + if err != nil { + t.Fatalf("default: %v", err) } + case <-time.After(2 * time.Second): + t.Fatal("default did not finish") } // Prove the actor's default revision, not the outer health check, owns // the rejection: telemetry may recover before this stale continuation. diff --git a/go/internal/drivers/registry_restart_test.go b/go/internal/drivers/registry_restart_test.go index ed0865f19..afab5cd5d 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{}), } diff --git a/go/internal/drivers/registry_runtime_preemption_test.go b/go/internal/drivers/registry_runtime_preemption_test.go new file mode 100644 index 000000000..b683ea2ed --- /dev/null +++ b/go/internal/drivers/registry_runtime_preemption_test.go @@ -0,0 +1,286 @@ +package drivers + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/srcfl/ftw/go/internal/config" + "github.com/srcfl/ftw/go/internal/telemetry" +) + +type blockedRuntime struct { + env *HostEnv + entered chan struct{} + release chan struct{} + defaulted chan struct{} + enteredOnce sync.Once + defaultOnce sync.Once + commandCalls atomic.Int32 +} + +func (r *blockedRuntime) Init(context.Context, []byte) error { return nil } +func (r *blockedRuntime) Poll(context.Context) (time.Duration, error) { + return time.Hour, nil +} +func (r *blockedRuntime) Command(context.Context, []byte) error { + r.commandCalls.Add(1) + r.enteredOnce.Do(func() { close(r.entered) }) + <-r.release + return nil +} +func (r *blockedRuntime) DefaultMode(context.Context) error { + r.defaultOnce.Do(func() { close(r.defaulted) }) + return nil +} +func (r *blockedRuntime) Cleanup(context.Context) error { return nil } +func (r *blockedRuntime) Env() *HostEnv { return r.env } + +func TestSendDefaultSurvivesFullStaleCommandQueue(t *testing.T) { + tel := telemetry.NewStore() + r := NewRegistry(tel) + runtime := &blockedRuntime{ + env: NewHostEnv("blocked-queue", tel), + entered: make(chan struct{}), + release: make(chan struct{}), + defaulted: make(chan struct{}), + } + lifecycleCtx, lifecycleCancel := context.WithCancel(context.Background()) + rd := &runningDriver{ + driver: runtime, + env: runtime.env, + cfg: config.Driver{Name: "blocked-queue"}, + generation: 1, + defaultConfirmed: true, + lifecycleCtx: lifecycleCtx, + lifecycleCancel: lifecycleCancel, + cmdCh: make(chan driverCmd, 2), + defaultCh: make(chan driverCmd, 1), + stop: make(chan bool, 1), + done: make(chan struct{}), + } + r.rec[rd.cfg.Name] = rd + go r.runLoop(rd) + var releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(runtime.release) }) } + t.Cleanup(func() { + release() + if _, ok := r.ControlStatus(rd.cfg.Name); ok { + r.remove(rd.cfg.Name, true) + } + }) + + activeDone := make(chan error, 1) + go func() { + activeDone <- r.Send(context.Background(), rd.cfg.Name, []byte(`{"action":"active"}`)) + }() + select { + case <-runtime.entered: + case <-time.After(time.Second): + t.Fatal("active command did not enter the runtime") + } + for i := 0; i < cap(rd.cmdCh); i++ { + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + err := r.Send(ctx, rd.cfg.Name, []byte(`{"action":"stale"}`)) + cancel() + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("stale command %d = %v, want deadline exceeded", i, err) + } + } + + defaultCtx, cancelDefault := context.WithTimeout(context.Background(), 20*time.Millisecond) + err := r.SendDefault(defaultCtx, rd.cfg.Name) + cancelDefault() + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("SendDefault = %v, want caller deadline while active command is blocked", err) + } + status, ok := r.ControlStatus(rd.cfg.Name) + if !ok || !status.Blocked || status.DefaultConfirmed || !status.RecoveryPending { + t.Fatalf("status after accepted default = %+v, running=%v", status, ok) + } + + release() + select { + case err := <-activeDone: + if !errors.Is(err, ErrCommandMayHaveRun) || !errors.Is(err, context.Canceled) { + t.Fatalf("active command = %v, want command-may-have-run plus canceled", err) + } + case <-time.After(time.Second): + t.Fatal("active command did not return") + } + select { + case <-runtime.defaulted: + case <-time.After(time.Second): + t.Fatal("accepted default did not run") + } + if calls := runtime.commandCalls.Load(); calls != 1 { + t.Fatalf("runtime command calls = %d, want only the active command", calls) + } + deadline := time.Now().Add(time.Second) + for { + status, ok = r.ControlStatus(rd.cfg.Name) + if ok && !status.Blocked && status.DefaultConfirmed && !status.RecoveryPending { + break + } + if time.Now().After(deadline) { + t.Fatalf("status after default = %+v, running=%v", status, ok) + } + time.Sleep(time.Millisecond) + } +} + +func TestSendDefaultWaitsForMutatingHTTPBeforeRunningDefault(t *testing.T) { + var defaultCalls atomic.Int32 + var commandInFlight atomic.Bool + var commandApplied atomic.Bool + var defaultBeforeCommandApplied atomic.Bool + commandEntered := make(chan struct{}) + defaultAfterCommand := make(chan struct{}) + releaseCommand := make(chan struct{}) + var enteredOnce sync.Once + var defaultAfterOnce sync.Once + var releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(releaseCommand) }) } + t.Cleanup(release) + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + if req.URL.Path == "/default" { + defaultCalls.Add(1) + if commandInFlight.Load() && !commandApplied.Load() { + defaultBeforeCommandApplied.Store(true) + } + if commandApplied.Load() { + defaultAfterOnce.Do(func() { close(defaultAfterCommand) }) + } + w.WriteHeader(http.StatusNoContent) + return + } + commandInFlight.Store(true) + enteredOnce.Do(func() { close(commandEntered) }) + <-releaseCommand + commandApplied.Store(true) + commandInFlight.Store(false) + w.WriteHeader(http.StatusNoContent) + })) + t.Cleanup(srv.Close) + + driverPath := writeTestDriver(t, ` +function driver_init(config) end +function driver_poll() return 60000 end +function driver_command(action, w, cmd) + local _, err = host.http_patch("`+srv.URL+`/command", "{}") + return err +end +function driver_default_mode() + local _, err = host.http_patch("`+srv.URL+`/default", "{}") + return err +end +`) + r := NewRegistry(telemetry.NewStore()) + cfg := config.Driver{ + Name: "blocked-http", + Lua: driverPath, + Capabilities: config.Capabilities{ + HTTP: &config.HTTPCapability{AllowWrite: true}, + }, + } + if err := r.Add(context.Background(), cfg); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + release() + r.ShutdownAll() + }) + if got := defaultCalls.Load(); got != 1 { + t.Fatalf("startup default calls = %d, want 1", got) + } + + commandDone := make(chan error, 1) + go func() { + commandDone <- r.Send(context.Background(), cfg.Name, []byte(`{"action":"battery","power_w":1000}`)) + }() + select { + case <-commandEntered: + case <-time.After(time.Second): + t.Fatal("command did not enter host.http_patch") + } + + defaultCtx, cancelDefault := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancelDefault() + defaultDone := make(chan error, 1) + go func() { defaultDone <- r.SendDefault(defaultCtx, cfg.Name) }() + select { + case err := <-defaultDone: + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("SendDefault before command completion = %v, want deadline exceeded", err) + } + case <-time.After(time.Second): + t.Fatal("SendDefault did not honor its caller deadline") + } + if defaultBeforeCommandApplied.Load() { + t.Fatal("default reached the device before the older mutating request completed") + } + if got := defaultCalls.Load(); got != 1 { + t.Fatalf("default calls before command completion = %d, want only startup default", got) + } + + release() + select { + case err := <-commandDone: + if !errors.Is(err, ErrCommandMayHaveRun) || !strings.Contains(err.Error(), context.Canceled.Error()) { + t.Fatalf("command result = %v, want command-may-have-run plus canceled", err) + } + case <-time.After(time.Second): + t.Fatal("command did not return after its HTTP request completed") + } + select { + case <-defaultAfterCommand: + case <-time.After(time.Second): + t.Fatal("accepted default did not run after the mutating request completed") + } + deadline := time.Now().Add(time.Second) + for defaultCalls.Load() < 3 && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + if got := defaultCalls.Load(); got < 3 { + t.Fatalf("default calls = %d, want startup, recovery, and requested default", got) + } + if defaultBeforeCommandApplied.Load() { + t.Fatal("default reached the device before the older mutating request completed") + } + status, ok := r.ControlStatus(cfg.Name) + if !ok || status.Blocked || !status.DefaultConfirmed || status.RecoveryPending { + t.Fatalf("status after confirmed default = %+v, running=%v", status, ok) + } +} + +func TestLuaHostSleepHonorsCommandContext(t *testing.T) { + driverPath := writeTestDriver(t, ` +function driver_command(action, w, cmd) + host.sleep(60000) + return true +end +`) + driver, err := NewLuaDriver(driverPath, NewHostEnv("blocked-sleep", telemetry.NewStore())) + if err != nil { + t.Fatal(err) + } + t.Cleanup(driver.Cleanup) + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + started := time.Now() + err = driver.Command(ctx, []byte(`{"action":"test"}`)) + if err == nil || !strings.Contains(err.Error(), context.DeadlineExceeded.Error()) { + t.Fatalf("Command = %v, want deadline exceeded", err) + } + if elapsed := time.Since(started); elapsed > time.Second { + t.Fatalf("context-canceled host.sleep returned after %s", elapsed) + } +}