From ce8fffd49e38ca21090eed28f6f5ccc8c47b15fc Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 14 Oct 2026 15:44:55 +0000 Subject: [PATCH 1/2] fix(drivers): restart a driver that stays offline A driver that went offline or failed to start stayed down until a config reload or a manual restart. The watchdog retries with backoff, and leaves disabled, observe-only, waiting push, and running cloud drivers alone. Signed-off-by: Cursor Agent Co-authored-by: Fredrik Ahlgren --- .changeset/driver-restart-watchdog.md | 5 + go/cmd/ftw/main.go | 26 + go/internal/api/api.go | 52 +- go/internal/api/api_test.go | 54 ++ go/internal/drivers/registry.go | 5 + go/internal/drivers/restart_watchdog.go | 569 +++++++++++++++++++ go/internal/drivers/restart_watchdog_test.go | 438 ++++++++++++++ 7 files changed, 1141 insertions(+), 8 deletions(-) create mode 100644 .changeset/driver-restart-watchdog.md create mode 100644 go/internal/drivers/restart_watchdog.go create mode 100644 go/internal/drivers/restart_watchdog_test.go diff --git a/.changeset/driver-restart-watchdog.md b/.changeset/driver-restart-watchdog.md new file mode 100644 index 00000000..ed526fee --- /dev/null +++ b/.changeset/driver-restart-watchdog.md @@ -0,0 +1,5 @@ +--- +"ftw": patch +--- + +Restart a driver that stays offline or fails to start, instead of leaving it down until someone restarts it by hand. diff --git a/go/cmd/ftw/main.go b/go/cmd/ftw/main.go index da5997e1..a57f0f7d 100644 --- a/go/cmd/ftw/main.go +++ b/go/cmd/ftw/main.go @@ -746,6 +746,8 @@ func main() { continue } if err := reg.Add(ctx, d); err != nil { + // The restart watchdog retries this start. A failed boot is not + // terminal. slog.Warn("failed to spawn driver", "name", d.Name, "err", err) } } @@ -791,6 +793,30 @@ func main() { capMu := &sync.RWMutex{} cfgMu := &sync.RWMutex{} modelsMu := &sync.Mutex{} + // Restart a driver that stays offline, or that never finished starting. + // The control-loop watchdog still hands a stale driver its default mode; + // this one replaces the process when that is not enough. Disabled and + // observe-only drivers are left alone, and a cloud login is not repeated + // just because the stream went quiet. Stop it before ShutdownAll, which + // was deferred above: a cancelled watchdog must not start a driver the + // shutdown has already removed. + watchdogCtx, stopRestartWatchdog := context.WithCancel(ctx) + watchdogDone := make(chan struct{}) + go func() { + defer close(watchdogDone) + reg.RunRestartWatchdog(watchdogCtx, func() ([]config.Driver, time.Duration) { + cfgMu.RLock() + defer cfgMu.RUnlock() + timeout := time.Duration(cfg.Site.WatchdogTimeoutS) * time.Second + return config.WithBatterySoCBounds(cfg.ModbusControlDrivers(), cfg.Batteries), timeout + }) + }() + // Registered after ShutdownAll, so this runs first and the watchdog + // has left the lifecycle gates before drivers are removed. + defer func() { + stopRestartWatchdog() + <-watchdogDone + }() // Pre-declare services that the hot-reload Applier needs to touch. // The Applier closure captures these by reference; they're assigned diff --git a/go/internal/api/api.go b/go/internal/api/api.go index a802fe9a..b9771b67 100644 --- a/go/internal/api/api.go +++ b/go/internal/api/api.go @@ -735,6 +735,36 @@ func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { // ---- /api/status ---- +// driverAutoRestart is the watchdog's view of one driver: how many times +// core has restarted it, why, and when the next try is due. Empty when the +// watchdog has nothing to say. +func (s *Server) driverAutoRestart(name string) map[string]any { + if s == nil || s.deps == nil || s.deps.Registry == nil || name == "" { + return nil + } + st, ok := s.deps.Registry.RestartStatus(name) + if !ok { + return nil + } + out := map[string]any{} + if st.Restarts > 0 { + out["auto_restarts"] = st.Restarts + } + if st.LastReason != "" { + out["auto_restart_reason"] = st.LastReason + } + if st.LastError != "" { + out["auto_restart_error"] = st.LastError + } + if !st.NextAttempt.IsZero() { + out["next_auto_restart_ms"] = st.NextAttempt.UnixMilli() + } + if len(out) == 0 { + return nil + } + return out +} + func statusDriverOnline(tel *telemetry.Store, name string) bool { if tel == nil || name == "" { return false @@ -1078,6 +1108,9 @@ func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) { } } } + for k, v := range s.driverAutoRestart(name) { + d[k] = v + } drivers[name] = d } // Merge config drivers that aren't in the registry so the UI can @@ -1090,19 +1123,22 @@ func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) { if _, ok := drivers[dc.Name]; ok { continue } + // Configured but not running — spawn probably failed. Show it as + // offline so the user sees it and can restart it. + row := map[string]any{ + "status": "offline", + "not_running": true, + } if dc.Disabled { - drivers[dc.Name] = map[string]any{ + row = map[string]any{ "status": "disabled", "disabled": true, } - } else { - // Configured but not running — spawn probably failed. Show - // as offline so the user sees it and can Restart. - drivers[dc.Name] = map[string]any{ - "status": "offline", - "not_running": true, - } } + for k, v := range s.driverAutoRestart(dc.Name) { + row[k] = v + } + drivers[dc.Name] = row } s.deps.CfgMu.RUnlock() diff --git a/go/internal/api/api_test.go b/go/internal/api/api_test.go index a6b80c87..1dc4bfcf 100644 --- a/go/internal/api/api_test.go +++ b/go/internal/api/api_test.go @@ -170,6 +170,60 @@ func TestHandlePostConfigDoesNotPersistEVPasswordOnInvalidConfig(t *testing.T) { } } +func TestHandleStatusReportsAutomaticDriverRestarts(t *testing.T) { + path := filepath.Join(t.TempDir(), "fail.lua") + src := "function driver_init(config) error('start failed') end\nfunction driver_poll() return 60000 end\n" + if err := os.WriteFile(path, []byte(src), 0o644); err != nil { + t.Fatal(err) + } + tel := telemetry.NewStore() + reg := drivers.NewRegistry(tel) + cfg := config.Driver{Name: "pixii", Lua: path} + now := time.Now() + reg.MaintainDrivers(context.Background(), now, []config.Driver{cfg}, time.Minute) + + srv := New(&Deps{ + Tel: tel, + Ctrl: &control.State{}, + CtrlMu: &sync.Mutex{}, + CapMu: &sync.RWMutex{}, + CfgMu: &sync.RWMutex{}, + Cfg: &config.Config{Drivers: []config.Driver{cfg}}, + Registry: reg, + }) + body := func() map[string]map[string]any { + t.Helper() + req := httptest.NewRequest(http.MethodGet, "/api/status", nil) + rr := httptest.NewRecorder() + srv.Handler().ServeHTTP(rr, req) + if rr.Code != http.StatusOK { + t.Fatalf("status = %d, body: %s", rr.Code, rr.Body.String()) + } + var resp struct { + Drivers map[string]map[string]any `json:"drivers"` + } + if err := json.Unmarshal(rr.Body.Bytes(), &resp); err != nil { + t.Fatalf("decode: %v", err) + } + return resp.Drivers + } + + pending := body()["pixii"] + if pending["not_running"] != true { + t.Fatalf("failed driver = %#v", pending) + } + next, _ := pending["next_auto_restart_ms"].(float64) + if int64(next) != now.Add(30*time.Second).UnixMilli() { + t.Fatalf("next retry = %v, want %d", pending["next_auto_restart_ms"], now.Add(30*time.Second).UnixMilli()) + } + + reg.MaintainDrivers(context.Background(), now.Add(30*time.Second), []config.Driver{cfg}, time.Minute) + retried := body()["pixii"] + if retried["auto_restarts"] != float64(1) || retried["auto_restart_reason"] != "start_failed" || retried["auto_restart_error"] == "" { + t.Fatalf("after retry = %#v", retried) + } +} + func TestHandleStatusIgnoresOfflineDERInLiveBalance(t *testing.T) { tel := telemetry.NewStore() ctrl := &control.State{SiteMeterDriver: "site"} diff --git a/go/internal/drivers/registry.go b/go/internal/drivers/registry.go index 222e71ea..957371c8 100644 --- a/go/internal/drivers/registry.go +++ b/go/internal/drivers/registry.go @@ -115,6 +115,11 @@ type Registry struct { lifecycleHook func(name string) lifecycleMu sync.Mutex lifecycleGates map[string]*lifecycleGate + + // restartMu guards the automatic-restart backoff. It is never held + // while a driver is started or stopped. + restartMu sync.Mutex + restarts map[string]*driverRestartState } // lifecycleGate serializes every lifecycle transition for one driver name. diff --git a/go/internal/drivers/restart_watchdog.go b/go/internal/drivers/restart_watchdog.go new file mode 100644 index 00000000..b4651aca --- /dev/null +++ b/go/internal/drivers/restart_watchdog.go @@ -0,0 +1,569 @@ +package drivers + +import ( + "context" + "log/slog" + "strings" + "time" + + "github.com/srcfl/ftw/go/internal/config" + "github.com/srcfl/ftw/go/internal/telemetry" +) + +// Restart backoff after a driver has been offline, stale, or unable to +// start. The last step repeats. +var restartBackoffSteps = [...]time.Duration{ + 30 * time.Second, + time.Minute, + 2 * time.Minute, + 5 * time.Minute, +} + +// restartWatchInterval is how often the running watchdog looks. The +// shortest backoff is 30s, so this only needs to be a fraction of that. +const restartWatchInterval = 5 * time.Second + +// DriverMaintenance is one automatic start or restart MaintainDrivers performed. +type DriverMaintenance struct { + Name string + Reason string // "offline", "stale", or "start_failed" + Err error +} + +// DriverRestartStatus is the watchdog counters for one driver. +type DriverRestartStatus struct { + Restarts uint64 + LastAttempt time.Time + LastError string + LastReason string + NextAttempt time.Time +} + +// driverRestartState is the backoff memory for one driver name. seenData +// survives a restart wiping the health record, so a push driver that has +// already delivered telemetry is not later treated as "still waiting". +type driverRestartState struct { + step int + unhealthySince time.Time + nextAttempt time.Time + restarts uint64 + lastAttempt time.Time + lastError string + lastReason string + seenData bool + awaitingRecovery bool + notedCloudSkip bool +} + +type restartLive int + +const ( + liveHealthy restartLive = iota + liveWaiting + liveDown + liveStartFailed + liveCloud +) + +// RunRestartWatchdog restarts drivers that stay offline or stale, and +// retries drivers that failed to start. It stops when ctx is cancelled. +// snapshot must return the drivers that should be running (the same list +// startup and reload use) and the site watchdog timeout. +func (r *Registry) RunRestartWatchdog(ctx context.Context, snapshot func() ([]config.Driver, time.Duration)) { + if r == nil || snapshot == nil { + return + } + if ctx == nil { + ctx = context.Background() + } + tick := func() { + drivers, timeout := snapshot() + if timeout <= 0 { + timeout = 60 * time.Second + } + r.MaintainDrivers(ctx, time.Now(), drivers, timeout) + } + tick() + ticker := time.NewTicker(restartWatchInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + tick() + } + } +} + +// MaintainDrivers starts or restarts drivers that have been offline, stale, +// or unable to start. now is the decision clock. Disabled and observe-only +// drivers are left alone. A running cloud or OAuth driver is left alone too, +// so a quiet stream cannot become a re-login loop; a failed start of one is +// still retried. An MQTT, WebSocket, or TCP driver that is polling and has +// never delivered data is waiting, not dead. +func (r *Registry) MaintainDrivers(ctx context.Context, now time.Time, drivers []config.Driver, staleAfter time.Duration) []DriverMaintenance { + if r == nil { + return nil + } + if ctx == nil { + ctx = context.Background() + } + desired := orderedDesiredDrivers(drivers) + keep := make(map[string]struct{}, len(desired)) + var out []DriverMaintenance + for _, cfg := range desired { + keep[cfg.Name] = struct{}{} + if err := ctx.Err(); err != nil { + break + } + if ev, ok := r.maintainDriver(ctx, now, cfg, staleAfter); ok { + out = append(out, ev) + } + } + r.forgetRestartExcept(keep) + return out +} + +// RestartStatus reports watchdog counters for one driver. +func (r *Registry) RestartStatus(name string) (DriverRestartStatus, bool) { + if r == nil || name == "" { + return DriverRestartStatus{}, false + } + r.restartMu.Lock() + defer r.restartMu.Unlock() + st := r.restarts[name] + if st == nil { + return DriverRestartStatus{}, false + } + out := DriverRestartStatus{ + Restarts: st.restarts, + LastAttempt: st.lastAttempt, + LastError: st.lastError, + LastReason: st.lastReason, + NextAttempt: st.nextAttempt, + } + if out.NextAttempt.IsZero() && !st.unhealthySince.IsZero() { + out.NextAttempt = st.unhealthySince.Add(restartDelay(st.step)) + } + if out.Restarts == 0 && out.LastReason == "" && out.NextAttempt.IsZero() && st.unhealthySince.IsZero() { + return DriverRestartStatus{}, false + } + return out, true +} + +func (r *Registry) maintainDriver(ctx context.Context, now time.Time, cfg config.Driver, staleAfter time.Duration) (DriverMaintenance, bool) { + if cfg.Disabled || cfg.ObserveOnly || cfg.Name == "" { + r.deleteRestart(cfg.Name) + return DriverMaintenance{}, false + } + // Same gate Restart and Reload use, so a config reload cannot start + // this driver between the liveness check and the replacement. + release := r.acquireLifecycle(cfg.Name) + defer release() + + entry, _ := ParseCatalogFile(cfg.Lua) + live, h := r.classifyDriver(cfg, entry, now, staleAfter) + switch live { + case liveHealthy, liveWaiting: + r.noteHealthy(cfg.Name, h, live == liveWaiting) + return DriverMaintenance{}, false + case liveCloud: + r.noteCloudRunning(cfg.Name, h, now, staleAfter) + return DriverMaintenance{}, false + case liveDown, liveStartFailed: + // A cancelled watchdog must not stop a driver shutdown is about to + // remove. The check is after the lifecycle gate so a reload that + // held the gate cannot sneak a start in either. + if err := ctx.Err(); err != nil { + return DriverMaintenance{}, false + } + if !r.restartDue(cfg.Name, h, now, staleAfter, live) { + return DriverMaintenance{}, false + } + reason := restartReason(live, h, now, staleAfter) + attempt, next := r.scheduleRestart(cfg.Name, h, now, reason) + err := r.restartLocked(ctx, cfg) + if err != nil { + r.noteRestartError(cfg.Name, err) + slog.Warn("automatic driver restart failed", + "name", cfg.Name, "reason", reason, "restarts", attempt, "next_backoff", next.String(), "err", err) + } else { + slog.Warn("automatic driver restart", + "name", cfg.Name, "reason", reason, "restarts", attempt, "next_backoff", next.String()) + r.finishRestart(cfg.Name) + } + return DriverMaintenance{Name: cfg.Name, Reason: reason, Err: err}, true + default: + return DriverMaintenance{}, false + } +} + +// restartLocked replaces a running driver, or starts one that is not +// running. The caller holds the lifecycle gate. Shutdown still runs +// driver_default_mode, and the new generation confirms it again, which is +// the same hand-back Restart uses. +func (r *Registry) restartLocked(ctx context.Context, cfg config.Driver) error { + r.mu.Lock() + _, running := r.rec[cfg.Name] + r.mu.Unlock() + if running { + r.removeLocked(cfg.Name, false) + } + if cfg.Disabled || cfg.ObserveOnly { + return nil + } + return r.add(ctx, cfg, true) +} + +func (r *Registry) finishRestart(name string) { + var h *telemetry.DriverHealth + if r.tel != nil { + h = r.tel.DriverHealth(name) + } + r.noteHealthy(name, h, false) +} + +func (r *Registry) classifyDriver(cfg config.Driver, entry CatalogEntry, now time.Time, staleAfter time.Duration) (restartLive, *telemetry.DriverHealth) { + r.mu.Lock() + _, running := r.rec[cfg.Name] + r.mu.Unlock() + if !running { + return liveStartFailed, nil + } + var h *telemetry.DriverHealth + if r.tel != nil { + h = r.tel.DriverHealth(cfg.Name) + } + // A running cloud or OAuth driver is not torn down for staleness. + // driver_init would log in again, and for a rotating refresh token + // that is the loop registry.Add warns about. + if cloudOrOAuth(cfg, entry) { + return liveCloud, h + } + seen := r.driverSeenData(cfg.Name) + if h != nil && h.LastSuccess != nil { + seen = true + } + if pushTransport(cfg, entry) && pushWaiting(h, seen) { + return liveWaiting, h + } + if h == nil || !driverDown(h, now, staleAfter) { + return liveHealthy, h + } + return liveDown, h +} + +func cloudOrOAuth(cfg config.Driver, entry CatalogEntry) bool { + if strings.TrimSpace(cfg.CredentialOwner) != "" { + return true + } + if strings.TrimSpace(entry.AuthPostPath) != "" || len(entry.AuthPostPaths) > 0 { + return true + } + if len(entry.HTTPHosts) == 0 { + return false + } + for _, p := range entry.Protocols { + switch strings.ToLower(strings.TrimSpace(p)) { + case "modbus", "mqtt", "serial", "tcp": + // A local bus owns the session. HTTP beside it is not a cloud login. + return false + } + } + for _, p := range entry.Protocols { + switch strings.ToLower(strings.TrimSpace(p)) { + case "http", "websocket": + return true + } + } + return false +} + +func pushTransport(cfg config.Driver, entry CatalogEntry) bool { + if cfg.EffectiveMQTT() != nil || cfg.Capabilities.WebSocket != nil || cfg.Capabilities.TCP != nil { + return true + } + for _, p := range entry.Protocols { + switch strings.ToLower(strings.TrimSpace(p)) { + case "mqtt", "websocket", "tcp": + return true + } + } + return false +} + +// pushWaiting is an MQTT, WebSocket, or TCP driver whose poll loop is +// healthy and that has not delivered a reading yet. Silence here is the +// device not having spoken, not a dead session. +func pushWaiting(h *telemetry.DriverHealth, seenData bool) bool { + if h == nil || seenData || h.LastSuccess != nil { + return false + } + if h.ConsecutiveErrors > 0 || strings.TrimSpace(h.LastError) != "" { + return false + } + return h.TickCount > 0 +} + +func driverDown(h *telemetry.DriverHealth, now time.Time, timeout time.Duration) bool { + if h == nil { + return false + } + if h.Status == telemetry.StatusOffline { + return true + } + return healthIsStale(h, now, timeout) +} + +func healthIsStale(h *telemetry.DriverHealth, now time.Time, timeout time.Duration) bool { + if h == nil { + return false + } + eff := effectiveWatchdog(h, timeout) + since := h.StartedAt + if h.LastSuccess != nil { + since = *h.LastSuccess + } + if since.IsZero() { + return true + } + return now.Sub(since) > eff +} + +func effectiveWatchdog(h *telemetry.DriverHealth, timeout time.Duration) time.Duration { + if h != nil && h.WatchdogTimeoutOverride > 0 { + return h.WatchdogTimeoutOverride + } + if timeout <= 0 { + return 60 * time.Second + } + return timeout +} + +func restartReason(live restartLive, h *telemetry.DriverHealth, now time.Time, timeout time.Duration) string { + if live == liveStartFailed { + return "start_failed" + } + if h != nil && h.Status == telemetry.StatusOffline { + return "offline" + } + if h != nil && healthIsStale(h, now, timeout) { + return "stale" + } + return "offline" +} + +func restartDelay(step int) time.Duration { + if step < 0 { + step = 0 + } + if step >= len(restartBackoffSteps) { + return restartBackoffSteps[len(restartBackoffSteps)-1] + } + return restartBackoffSteps[step] +} + +// restartDue reports whether the current backoff has elapsed. The first +// time a start failure or a fresh-but-offline driver is seen, it records +// the observation and waits. A reading that is already stale long enough +// is due immediately. +func (r *Registry) restartDue(name string, h *telemetry.DriverHealth, now time.Time, timeout time.Duration, live restartLive) bool { + var st driverRestartState + r.updateRestart(name, true, func(s *driverRestartState) { + if h != nil && h.LastSuccess != nil { + s.seenData = true + } + readingStale := h != nil && healthIsStale(h, now, timeout) + if live == liveStartFailed || (live == liveDown && !readingStale) { + if s.unhealthySince.IsZero() { + s.unhealthySince = now + } + } + st = *s + }) + if !st.nextAttempt.IsZero() && now.Before(st.nextAttempt) { + return false + } + since := unhealthyAt(st, h, now, timeout, live) + if since.IsZero() || since.After(now) { + return false + } + return now.Sub(since) >= restartDelay(st.step) +} + +func unhealthyAt(st driverRestartState, h *telemetry.DriverHealth, now time.Time, timeout time.Duration, live restartLive) time.Time { + if live != liveStartFailed && h != nil && healthIsStale(h, now, timeout) { + eff := effectiveWatchdog(h, timeout) + since := h.StartedAt + if h.LastSuccess != nil { + since = *h.LastSuccess + } + if !since.IsZero() { + return since.Add(eff) + } + } + if !st.unhealthySince.IsZero() { + return st.unhealthySince + } + return now +} + +func (r *Registry) scheduleRestart(name string, h *telemetry.DriverHealth, now time.Time, reason string) (uint64, time.Duration) { + var restarts uint64 + var next time.Duration + r.updateRestart(name, true, func(s *driverRestartState) { + if h != nil && h.LastSuccess != nil { + s.seenData = true + } + s.step++ + if s.step >= len(restartBackoffSteps) { + s.step = len(restartBackoffSteps) - 1 + } + next = restartDelay(s.step) + s.nextAttempt = now.Add(next) + s.restarts++ + restarts = s.restarts + s.lastAttempt = now + s.lastReason = reason + s.lastError = "" + s.awaitingRecovery = true + s.notedCloudSkip = false + if reason == "start_failed" { + s.unhealthySince = now + } else { + s.unhealthySince = time.Time{} + } + }) + return restarts, next +} + +func (r *Registry) noteRestartError(name string, err error) { + if err == nil { + return + } + r.updateRestart(name, true, func(s *driverRestartState) { + s.lastError = err.Error() + }) +} + +func (r *Registry) noteHealthy(name string, h *telemetry.DriverHealth, waiting bool) { + var recovered bool + var restarts uint64 + ok := r.updateRestart(name, false, func(s *driverRestartState) { + if h != nil && h.LastSuccess != nil { + s.seenData = true + } + if waiting || s.lastAttempt.IsZero() || telemetrySince(h, s.lastAttempt) { + recovered = s.awaitingRecovery + restarts = s.restarts + clearRestartBackoff(s) + } else { + s.unhealthySince = time.Time{} + } + }) + if ok && recovered { + slog.Info("driver recovered after automatic restart", "name", name, "restarts", restarts) + } +} + +func (r *Registry) noteCloudRunning(name string, h *telemetry.DriverHealth, now time.Time, timeout time.Duration) { + down := h != nil && driverDown(h, now, timeout) + var skipLog bool + r.updateRestart(name, down, func(s *driverRestartState) { + if h != nil && h.LastSuccess != nil { + s.seenData = true + } + if !down { + if s.lastAttempt.IsZero() || telemetrySince(h, s.lastAttempt) { + clearRestartBackoff(s) + } + s.notedCloudSkip = false + return + } + if !s.notedCloudSkip { + s.notedCloudSkip = true + skipLog = true + } + }) + if skipLog { + slog.Info("cloud driver left running", "name", name, "reason", "automatic restart would re-login") + } +} + +func telemetrySince(h *telemetry.DriverHealth, since time.Time) bool { + if h == nil || h.LastSuccess == nil || h.Status == telemetry.StatusOffline { + return false + } + if !since.IsZero() && h.LastSuccess.Before(since) { + return false + } + return true +} + +func clearRestartBackoff(s *driverRestartState) { + s.step = 0 + s.unhealthySince = time.Time{} + s.nextAttempt = time.Time{} + s.lastError = "" + s.awaitingRecovery = false + s.notedCloudSkip = false +} + +func (r *Registry) driverSeenData(name string) bool { + r.restartMu.Lock() + defer r.restartMu.Unlock() + st := r.restarts[name] + return st != nil && st.seenData +} + +func (r *Registry) updateRestart(name string, create bool, fn func(*driverRestartState)) bool { + r.restartMu.Lock() + defer r.restartMu.Unlock() + if r.restarts == nil { + r.restarts = map[string]*driverRestartState{} + } + st := r.restarts[name] + if st == nil { + if !create { + return false + } + st = &driverRestartState{} + r.restarts[name] = st + } + fn(st) + return true +} + +func (r *Registry) deleteRestart(name string) { + r.restartMu.Lock() + defer r.restartMu.Unlock() + delete(r.restarts, name) +} + +func (r *Registry) forgetRestartExcept(keep map[string]struct{}) { + r.restartMu.Lock() + defer r.restartMu.Unlock() + for name := range r.restarts { + if _, ok := keep[name]; !ok { + delete(r.restarts, name) + } + } +} + +func orderedDesiredDrivers(in []config.Driver) []config.Driver { + index := make(map[string]int, len(in)) + out := make([]config.Driver, 0, len(in)) + for _, cfg := range in { + if cfg.Name == "" { + continue + } + if i, ok := index[cfg.Name]; ok { + out[i] = cfg + continue + } + index[cfg.Name] = len(out) + out = append(out, cfg) + } + return out +} diff --git a/go/internal/drivers/restart_watchdog_test.go b/go/internal/drivers/restart_watchdog_test.go new file mode 100644 index 00000000..9ef696c2 --- /dev/null +++ b/go/internal/drivers/restart_watchdog_test.go @@ -0,0 +1,438 @@ +package drivers + +import ( + "context" + "testing" + "time" + + "github.com/srcfl/ftw/go/internal/config" + "github.com/srcfl/ftw/go/internal/telemetry" +) + +const restartWatchSessionDriver = ` +function driver_init(config) + host.set_poll_interval(60000) + if config ~= nil and config.fail == "yes" then + error("start failed") + end + if config ~= nil and config.emit == "yes" then + host.emit_metric("alive", 1) + end +end +function driver_poll() return 60000 end +function driver_default_mode() + host.emit_metric("default_mode", 1) +end +` + +const restartWatchCloudDriver = ` +DRIVER = { + id = "cloud_login", + protocols = { "http" }, + http_hosts = { "api.example.com" }, + auth_post_path = "/oauth/token", +} +function driver_init(config) + host.set_poll_interval(60000) + if config ~= nil and config.fail == "yes" then + error("token rejected") + end +end +function driver_poll() return 60000 end +` + +const restartWatchPushDriver = ` +function driver_init(config) + host.set_poll_interval(20) +end +function driver_poll() return 20 end +` + +func TestRestartBackoffSteps(t *testing.T) { + want := []time.Duration{ + 30 * time.Second, + time.Minute, + 2 * time.Minute, + 5 * time.Minute, + 5 * time.Minute, + } + for step, delay := range want { + if got := restartDelay(step); got != delay { + t.Fatalf("step %d = %s, want %s", step, got, delay) + } + } +} + +func TestCloudAndPushClassification(t *testing.T) { + if !cloudOrOAuth(config.Driver{CredentialOwner: "owner"}, CatalogEntry{}) { + t.Fatal("credential owner should be treated as OAuth") + } + if !cloudOrOAuth(config.Driver{}, CatalogEntry{AuthPostPath: "/oauth/token"}) { + t.Fatal("auth_post_path should be treated as OAuth") + } + if !cloudOrOAuth(config.Driver{}, CatalogEntry{Protocols: []string{"http"}, HTTPHosts: []string{"api.easee.com"}}) { + t.Fatal("vendor HTTP host should be treated as cloud") + } + if cloudOrOAuth(config.Driver{}, CatalogEntry{Protocols: []string{"http"}}) { + t.Fatal("HTTP without a vendor host is a local device") + } + if cloudOrOAuth(config.Driver{}, CatalogEntry{Protocols: []string{"modbus", "http"}, HTTPHosts: []string{"api.example.com"}}) { + t.Fatal("a modbus driver is not cloud-only") + } + mqtt := config.Driver{Capabilities: config.Capabilities{MQTT: &config.MQTTConfig{Host: "broker"}}} + if !pushTransport(mqtt, CatalogEntry{}) { + t.Fatal("MQTT capability should be a push transport") + } + if !pushTransport(config.Driver{}, CatalogEntry{Protocols: []string{"tcp"}}) { + t.Fatal("TCP protocol should be a push transport") + } + if !pushTransport(config.Driver{}, CatalogEntry{Protocols: []string{"websocket"}}) { + t.Fatal("websocket protocol should be a push transport") + } + if pushTransport(config.Driver{}, CatalogEntry{Protocols: []string{"modbus"}}) { + t.Fatal("modbus is not a push transport") + } +} + +func TestRestartWatchdogRestartsOfflineDriverThenRecovers(t *testing.T) { + tel := telemetry.NewStore() + r := NewRegistry(tel) + t.Cleanup(r.ShutdownAll) + path := writeTestDriver(t, restartWatchSessionDriver) + cfg := config.Driver{Name: "battery", Lua: path, Config: map[string]any{"emit": "yes"}} + if err := r.Add(context.Background(), cfg); err != nil { + t.Fatal(err) + } + now := time.Now() + markDriverDown(tel, "battery", now, 2*time.Minute) + + got := r.MaintainDrivers(context.Background(), now, []config.Driver{cfg}, time.Minute) + if len(got) != 1 || got[0].Reason != "offline" || got[0].Err != nil { + t.Fatalf("offline restart = %+v", got) + } + if _, _, ok := tel.LatestMetric("battery", "default_mode"); !ok { + t.Fatal("restart did not run driver_default_mode") + } + health := tel.DriverHealth("battery") + if health == nil || health.Status == telemetry.StatusOffline || health.LastSuccess == nil { + t.Fatalf("restart did not come back with fresh telemetry: %+v", health) + } + if again := r.MaintainDrivers(context.Background(), now, []config.Driver{cfg}, time.Minute); len(again) != 0 { + t.Fatalf("recovered driver restarted again: %+v", again) + } + + // Backoff was reset by the recovery, so a new outage is eligible at 30s + // of staleness rather than the next grown delay. + markDriverStale(tel, "battery", now, 2*time.Minute) + got = r.MaintainDrivers(context.Background(), now, []config.Driver{cfg}, time.Minute) + if len(got) != 1 || got[0].Reason != "stale" || got[0].Err != nil { + t.Fatalf("stale restart after recovery = %+v", got) + } + status, ok := r.RestartStatus("battery") + if !ok || status.Restarts != 2 { + t.Fatalf("status = %+v ok=%v, want 2 restarts", status, ok) + } +} + +func TestRestartWatchdogBackoffGrowsAndResets(t *testing.T) { + tel := telemetry.NewStore() + r := NewRegistry(tel) + t.Cleanup(r.ShutdownAll) + path := writeTestDriver(t, restartWatchSessionDriver) + cfg := config.Driver{Name: "battery", Lua: path, Config: map[string]any{"fail": "yes"}} + base := time.Now() + // Waits before each new attempt: 30s, 1m, 2m, 5m, then 5m again. + steps := []struct { + at time.Duration + want uint64 + }{ + {0, 0}, + {29 * time.Second, 0}, + {30 * time.Second, 1}, + {89 * time.Second, 1}, + {90 * time.Second, 2}, + {210*time.Second - time.Second, 2}, + {210 * time.Second, 3}, + {510*time.Second - time.Second, 3}, + {510 * time.Second, 4}, + {810*time.Second - time.Second, 4}, + {810 * time.Second, 5}, + } + for _, step := range steps { + r.MaintainDrivers(context.Background(), base.Add(step.at), []config.Driver{cfg}, time.Minute) + if got := restartCount(r, "battery"); got != step.want { + t.Fatalf("at %s restarts = %d, want %d", step.at, got, step.want) + } + if step.want > 0 && len(r.Names()) != 0 { + t.Fatalf("at %s driver is running after a failed start", step.at) + } + } + + successAt := base.Add(810*time.Second + 5*time.Minute) + cfg.Config = map[string]any{"emit": "yes"} + if got := r.MaintainDrivers(context.Background(), successAt, []config.Driver{cfg}, time.Minute); len(got) != 1 || got[0].Err != nil { + t.Fatalf("successful retry = %+v", got) + } + markDriverRecovered(tel, "battery", successAt) + if got := r.MaintainDrivers(context.Background(), successAt, []config.Driver{cfg}, time.Minute); len(got) != 0 { + t.Fatalf("recovered driver restarted: %+v", got) + } + + r.Remove("battery") + cfg.Config = map[string]any{"fail": "yes"} + r.MaintainDrivers(context.Background(), successAt, []config.Driver{cfg}, time.Minute) + if got := restartCount(r, "battery"); got != 6 { + t.Fatalf("observation after recovery changed restarts to %d", got) + } + r.MaintainDrivers(context.Background(), successAt.Add(29*time.Second), []config.Driver{cfg}, time.Minute) + if got := restartCount(r, "battery"); got != 6 { + t.Fatalf("backoff reset waited %s, restarts = %d", 29*time.Second, got) + } + r.MaintainDrivers(context.Background(), successAt.Add(30*time.Second), []config.Driver{cfg}, time.Minute) + if got := restartCount(r, "battery"); got != 7 { + t.Fatalf("after recovery the next retry waited past 30s, restarts = %d", got) + } +} + +func TestRestartWatchdogRetriesStartFailure(t *testing.T) { + tel := telemetry.NewStore() + r := NewRegistry(tel) + t.Cleanup(r.ShutdownAll) + path := writeTestDriver(t, restartWatchSessionDriver) + cfg := config.Driver{Name: "battery", Lua: path, Config: map[string]any{"fail": "yes"}} + now := time.Now() + if got := r.MaintainDrivers(context.Background(), now, []config.Driver{cfg}, time.Minute); len(got) != 0 { + t.Fatalf("first look retried immediately: %+v", got) + } + status, ok := r.RestartStatus("battery") + if !ok || !status.NextAttempt.Equal(now.Add(30*time.Second)) || status.Restarts != 0 { + t.Fatalf("pending retry = %+v ok=%v", status, ok) + } + got := r.MaintainDrivers(context.Background(), now.Add(30*time.Second), []config.Driver{cfg}, time.Minute) + if len(got) != 1 || got[0].Reason != "start_failed" || got[0].Err == nil { + t.Fatalf("start retry = %+v", got) + } + if len(r.Names()) != 0 { + t.Fatalf("failed start left a running driver: %v", r.Names()) + } + status, ok = r.RestartStatus("battery") + if !ok || status.Restarts != 1 || status.LastReason != "start_failed" || status.LastError == "" { + t.Fatalf("status after failed start = %+v ok=%v", status, ok) + } +} + +func TestRestartWatchdogLeavesDisabledAndObserveOnly(t *testing.T) { + tel := telemetry.NewStore() + r := NewRegistry(tel) + t.Cleanup(r.ShutdownAll) + path := writeTestDriver(t, restartWatchSessionDriver) + now := time.Now() + disabled := config.Driver{Name: "battery", Lua: path, Disabled: true, Config: map[string]any{"emit": "yes"}} + if got := r.MaintainDrivers(context.Background(), now.Add(time.Hour), []config.Driver{disabled}, time.Minute); len(got) != 0 || len(r.Names()) != 0 { + t.Fatalf("disabled driver was started: events=%+v names=%v", got, r.Names()) + } + + enabled := disabled + enabled.Disabled = false + if err := r.Add(context.Background(), enabled); err != nil { + t.Fatal(err) + } + markDriverDown(tel, "battery", now, 10*time.Minute) + if got := r.MaintainDrivers(context.Background(), now, []config.Driver{disabled}, time.Minute); len(got) != 0 { + t.Fatalf("disabled running driver was restarted: %+v", got) + } + if len(r.Names()) != 1 { + t.Fatalf("disabled entry stopped the driver: %v", r.Names()) + } + + r.Remove("battery") + observe := config.Driver{Name: "battery", Lua: path, ObserveOnly: true, Config: map[string]any{"emit": "yes"}} + if err := r.Add(context.Background(), observe); err != nil { + t.Fatal(err) + } + markDriverDown(tel, "battery", now, 10*time.Minute) + if got := r.MaintainDrivers(context.Background(), now, []config.Driver{observe}, time.Minute); len(got) != 0 { + t.Fatalf("observe-only driver was restarted: %+v", got) + } + if _, ok := r.RestartStatus("battery"); ok { + t.Fatal("observe-only driver has restart state") + } +} + +func TestRestartWatchdogDoesNotReloginCloudDriver(t *testing.T) { + tel := telemetry.NewStore() + r := NewRegistry(tel) + t.Cleanup(r.ShutdownAll) + path := writeTestDriver(t, restartWatchCloudDriver) + cfg := config.Driver{Name: "myuplink", Lua: path} + if err := r.Add(context.Background(), cfg); err != nil { + t.Fatal(err) + } + now := time.Now() + markDriverDown(tel, "myuplink", now, 30*time.Minute) + if got := r.MaintainDrivers(context.Background(), now.Add(time.Hour), []config.Driver{cfg}, time.Minute); len(got) != 0 { + t.Fatalf("stale cloud driver was restarted: %+v", got) + } + if restartCount(r, "myuplink") != 0 || len(r.Names()) != 1 { + t.Fatalf("cloud driver restarts=%d names=%v", restartCount(r, "myuplink"), r.Names()) + } + + r.Remove("myuplink") + cfg.Config = map[string]any{"fail": "yes"} + base := time.Now() + r.MaintainDrivers(context.Background(), base, []config.Driver{cfg}, time.Minute) + got := r.MaintainDrivers(context.Background(), base.Add(30*time.Second), []config.Driver{cfg}, time.Minute) + if len(got) != 1 || got[0].Reason != "start_failed" || got[0].Err == nil { + t.Fatalf("cloud start retry = %+v", got) + } +} + +func TestRestartWatchdogLeavesMQTTWaitingForData(t *testing.T) { + tel := telemetry.NewStore() + r := NewRegistry(tel) + t.Cleanup(r.ShutdownAll) + path := writeTestDriver(t, restartWatchPushDriver) + cfg := config.Driver{ + Name: "ferroamp", + Lua: path, + Capabilities: config.Capabilities{ + MQTT: &config.MQTTConfig{Host: "127.0.0.1", Port: 1883}, + }, + } + if err := r.Add(context.Background(), cfg); err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(2 * time.Second) + for { + health := tel.DriverHealth("ferroamp") + if health != nil && health.TickCount > 0 { + break + } + if time.Now().After(deadline) { + t.Fatal("mqtt driver never polled") + } + time.Sleep(10 * time.Millisecond) + } + now := time.Now() + health := tel.DriverHealthMut("ferroamp") + health.StartedAt = now.Add(-10 * time.Minute) + health.LastSuccess = nil + health.ConsecutiveErrors = 0 + health.LastError = "" + health.SetOffline() + if got := r.MaintainDrivers(context.Background(), now, []config.Driver{cfg}, time.Minute); len(got) != 0 { + t.Fatalf("mqtt driver waiting for data was restarted: %+v", got) + } + + past := now.Add(-10 * time.Minute) + health = tel.DriverHealthMut("ferroamp") + health.StartedAt = past + health.LastSuccess = &past + health.SetOffline() + got := r.MaintainDrivers(context.Background(), now, []config.Driver{cfg}, time.Minute) + if len(got) != 1 || got[0].Err != nil { + t.Fatalf("mqtt driver that had data and went stale was not restarted: %+v", got) + } +} + +func TestRestartWatchdogStopsWhenContextCancelled(t *testing.T) { + tel := telemetry.NewStore() + r := NewRegistry(tel) + t.Cleanup(r.ShutdownAll) + path := writeTestDriver(t, restartWatchSessionDriver) + cfg := config.Driver{Name: "battery", Lua: path, Config: map[string]any{"emit": "yes"}} + if err := r.Add(context.Background(), cfg); err != nil { + t.Fatal(err) + } + now := time.Now() + markDriverDown(tel, "battery", now, 10*time.Minute) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if got := r.MaintainDrivers(ctx, now, []config.Driver{cfg}, time.Minute); len(got) != 0 { + t.Fatalf("cancelled watchdog restarted: %+v", got) + } + if restartCount(r, "battery") != 0 || len(r.Names()) != 1 { + t.Fatalf("cancelled watchdog changed the driver: restarts=%d names=%v", restartCount(r, "battery"), r.Names()) + } +} + +func TestRestartWatchdogWaitsForLifecycleLock(t *testing.T) { + tel := telemetry.NewStore() + r := NewRegistry(tel) + t.Cleanup(r.ShutdownAll) + path := writeTestDriver(t, restartWatchSessionDriver) + cfg := config.Driver{Name: "battery", Lua: path, Config: map[string]any{"emit": "yes"}} + if err := r.Add(context.Background(), cfg); err != nil { + t.Fatal(err) + } + now := time.Now() + markDriverDown(tel, "battery", now, 10*time.Minute) + release := r.acquireLifecycle("battery") + released := false + t.Cleanup(func() { + if !released { + release() + } + }) + done := make(chan struct{}) + go func() { + defer close(done) + r.MaintainDrivers(context.Background(), now, []config.Driver{cfg}, time.Minute) + }() + select { + case <-done: + t.Fatal("watchdog restarted while the driver lifecycle lock was held") + case <-time.After(200 * time.Millisecond): + } + if restartCount(r, "battery") != 0 { + t.Fatal("restart counted while lifecycle lock was held") + } + release() + released = true + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("watchdog did not restart after the lifecycle lock was released") + } + if restartCount(r, "battery") != 1 { + t.Fatalf("restarts = %d, want 1", restartCount(r, "battery")) + } +} + +func markDriverDown(tel *telemetry.Store, name string, now time.Time, age time.Duration) { + health := tel.DriverHealthMut(name) + past := now.Add(-age) + health.StartedAt = past + health.LastSuccess = &past + health.SetOffline() + health.ConsecutiveErrors = 0 + health.LastError = "" +} + +func markDriverStale(tel *telemetry.Store, name string, now time.Time, age time.Duration) { + health := tel.DriverHealthMut(name) + past := now.Add(-age) + health.StartedAt = past + health.LastSuccess = &past + health.Status = telemetry.StatusOk + health.ConsecutiveErrors = 0 + health.LastError = "" +} + +func markDriverRecovered(tel *telemetry.Store, name string, at time.Time) { + health := tel.DriverHealthMut(name) + health.StartedAt = at + success := at + health.LastSuccess = &success + health.Status = telemetry.StatusOk + health.ConsecutiveErrors = 0 + health.LastError = "" +} + +func restartCount(r *Registry, name string) uint64 { + status, ok := r.RestartStatus(name) + if !ok { + return 0 + } + return status.Restarts +} From a9dd4345eb98e0c80e734d3590cde25e76941a45 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 14 Oct 2026 19:24:50 +0000 Subject: [PATCH 2/2] fix(drivers): skip a restart when config changed under the gate The watchdog copied the driver list, dropped the config lock, and could then start that copy after a reload had already disabled or edited the driver. It now re-reads the installed config after the lifecycle gate and leaves a disabled, observe-only, or changed driver to reload. Signed-off-by: Cursor Agent Co-authored-by: Fredrik Ahlgren --- go/cmd/ftw/main.go | 5 ++ go/internal/drivers/restart_watchdog.go | 60 +++++++++++++- go/internal/drivers/restart_watchdog_test.go | 87 ++++++++++++++++++++ 3 files changed, 148 insertions(+), 4 deletions(-) diff --git a/go/cmd/ftw/main.go b/go/cmd/ftw/main.go index a57f0f7d..dd55f2fb 100644 --- a/go/cmd/ftw/main.go +++ b/go/cmd/ftw/main.go @@ -805,6 +805,11 @@ func main() { go func() { defer close(watchdogDone) reg.RunRestartWatchdog(watchdogCtx, func() ([]config.Driver, time.Duration) { + // Copy under the lock, then drop it. MaintainDrivers reads this + // again after it takes the driver's lifecycle gate and skips the + // restart when the driver is now disabled, observe-only, or its + // config has changed. Holding cfgMu across the restart would + // block a settings save for the whole reconnect. cfgMu.RLock() defer cfgMu.RUnlock() timeout := time.Duration(cfg.Site.WatchdogTimeoutS) * time.Second diff --git a/go/internal/drivers/restart_watchdog.go b/go/internal/drivers/restart_watchdog.go index b4651aca..4a681ab7 100644 --- a/go/internal/drivers/restart_watchdog.go +++ b/go/internal/drivers/restart_watchdog.go @@ -68,7 +68,10 @@ const ( // RunRestartWatchdog restarts drivers that stay offline or stale, and // retries drivers that failed to start. It stops when ctx is cancelled. // snapshot must return the drivers that should be running (the same list -// startup and reload use) and the site watchdog timeout. +// startup and reload use) and the site watchdog timeout. It is called +// again after each driver's lifecycle gate is held. That second read +// must not take a lifecycle gate itself: the gate is already held, and +// config reload takes the config lock before the gate. func (r *Registry) RunRestartWatchdog(ctx context.Context, snapshot func() ([]config.Driver, time.Duration)) { if r == nil || snapshot == nil { return @@ -81,7 +84,10 @@ func (r *Registry) RunRestartWatchdog(ctx context.Context, snapshot func() ([]co if timeout <= 0 { timeout = 60 * time.Second } - r.MaintainDrivers(ctx, time.Now(), drivers, timeout) + r.maintainDrivers(ctx, time.Now(), drivers, timeout, func() []config.Driver { + current, _ := snapshot() + return current + }) } tick() ticker := time.NewTicker(restartWatchInterval) @@ -103,6 +109,15 @@ func (r *Registry) RunRestartWatchdog(ctx context.Context, snapshot func() ([]co // still retried. An MQTT, WebSocket, or TCP driver that is polling and has // never delivered data is waiting, not dead. func (r *Registry) MaintainDrivers(ctx context.Context, now time.Time, drivers []config.Driver, staleAfter time.Duration) []DriverMaintenance { + return r.maintainDrivers(ctx, now, drivers, staleAfter, nil) +} + +// maintainDrivers is MaintainDrivers with a second look at the installed +// config. current is read after the per-driver lifecycle gate is held, so a +// reload that landed while this pass was between its snapshot and the gate +// is visible. A nil current trusts the snapshot; the running watchdog always +// passes one. +func (r *Registry) maintainDrivers(ctx context.Context, now time.Time, drivers []config.Driver, staleAfter time.Duration, current func() []config.Driver) []DriverMaintenance { if r == nil { return nil } @@ -117,7 +132,7 @@ func (r *Registry) MaintainDrivers(ctx context.Context, now time.Time, drivers [ if err := ctx.Err(); err != nil { break } - if ev, ok := r.maintainDriver(ctx, now, cfg, staleAfter); ok { + if ev, ok := r.maintainDriver(ctx, now, cfg, staleAfter, current); ok { out = append(out, ev) } } @@ -152,7 +167,7 @@ func (r *Registry) RestartStatus(name string) (DriverRestartStatus, bool) { return out, true } -func (r *Registry) maintainDriver(ctx context.Context, now time.Time, cfg config.Driver, staleAfter time.Duration) (DriverMaintenance, bool) { +func (r *Registry) maintainDriver(ctx context.Context, now time.Time, cfg config.Driver, staleAfter time.Duration, current func() []config.Driver) (DriverMaintenance, bool) { if cfg.Disabled || cfg.ObserveOnly || cfg.Name == "" { r.deleteRestart(cfg.Name) return DriverMaintenance{}, false @@ -178,6 +193,15 @@ func (r *Registry) maintainDriver(ctx context.Context, now time.Time, cfg config if err := ctx.Err(); err != nil { return DriverMaintenance{}, false } + // The snapshot was copied before this gate. A reload can have + // installed a disabled, observe-only, or edited driver in that + // window and already decided not to start it. Later passes leave a + // running disabled driver alone, so this start would stick. + var fresh bool + cfg, fresh = r.revalidatedDriver(cfg, current) + if !fresh { + return DriverMaintenance{}, false + } if !r.restartDue(cfg.Name, h, now, staleAfter, live) { return DriverMaintenance{}, false } @@ -199,6 +223,34 @@ func (r *Registry) maintainDriver(ctx context.Context, now time.Time, cfg config } } +// revalidatedDriver compares the snapshot with the config installed now. +// The caller holds the driver's lifecycle gate. A nil current means the +// caller has no newer view and the snapshot stands. Otherwise a driver +// that is gone, disabled, observe-only, or edited since the snapshot is +// left for reload: this pass does not start the copy it captured earlier. +func (r *Registry) revalidatedDriver(snapshot config.Driver, current func() []config.Driver) (config.Driver, bool) { + if current == nil { + return snapshot, true + } + live, ok := findDriver(current(), snapshot.Name) + if !ok || live.Disabled || live.ObserveOnly { + r.deleteRestart(snapshot.Name) + reason := "removed" + if ok && live.Disabled { + reason = "disabled" + } else if ok && live.ObserveOnly { + reason = "observe_only" + } + slog.Info("automatic driver restart skipped", "name", snapshot.Name, "reason", reason) + return config.Driver{}, false + } + if !sameDriverConfig(snapshot, live) { + slog.Info("automatic driver restart skipped", "name", snapshot.Name, "reason", "config_changed") + return config.Driver{}, false + } + return live, true +} + // restartLocked replaces a running driver, or starts one that is not // running. The caller holds the lifecycle gate. Shutdown still runs // driver_default_mode, and the new generation confirms it again, which is diff --git a/go/internal/drivers/restart_watchdog_test.go b/go/internal/drivers/restart_watchdog_test.go index 9ef696c2..c17c4c68 100644 --- a/go/internal/drivers/restart_watchdog_test.go +++ b/go/internal/drivers/restart_watchdog_test.go @@ -2,6 +2,7 @@ package drivers import ( "context" + "sync/atomic" "testing" "time" @@ -335,6 +336,92 @@ func TestRestartWatchdogLeavesMQTTWaitingForData(t *testing.T) { } } +func TestRestartWatchdogRechecksConfigAfterLifecycleGate(t *testing.T) { + path := writeTestDriver(t, restartWatchSessionDriver) + enabled := config.Driver{Name: "battery", Lua: path, Config: map[string]any{"fail": "yes"}} + cases := []struct { + name string + current func(config.Driver) config.Driver + wantStart bool + }{ + { + name: "disabled", + current: func(d config.Driver) config.Driver { d.Disabled = true; return d }, + wantStart: false, + }, + { + name: "observe-only", + current: func(d config.Driver) config.Driver { d.ObserveOnly = true; return d }, + wantStart: false, + }, + { + name: "config changed", + current: func(d config.Driver) config.Driver { + d.Config = map[string]any{"fail": "yes", "host": "10.0.0.9"} + return d + }, + wantStart: false, + }, + { + name: "unchanged", + current: func(d config.Driver) config.Driver { return d }, + wantStart: true, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + tel := telemetry.NewStore() + r := NewRegistry(tel) + t.Cleanup(r.ShutdownAll) + now := time.Now() + // A failed start is due 30s later. The second pass still carries + // the enabled snapshot; the installed config may have moved. + r.MaintainDrivers(context.Background(), now, []config.Driver{enabled}, time.Minute) + + release := r.acquireLifecycle(enabled.Name) + released := false + t.Cleanup(func() { + if !released { + release() + } + }) + var looked atomic.Int32 + done := make(chan []DriverMaintenance, 1) + go func() { + done <- r.maintainDrivers(context.Background(), now.Add(30*time.Second), []config.Driver{enabled}, time.Minute, func() []config.Driver { + looked.Add(1) + return []config.Driver{tc.current(enabled)} + }) + }() + select { + case got := <-done: + t.Fatalf("watchdog finished while the lifecycle gate was held: %+v", got) + case <-time.After(200 * time.Millisecond): + } + if looked.Load() != 0 { + t.Fatal("config was revalidated before the lifecycle gate") + } + release() + released = true + var got []DriverMaintenance + select { + case got = <-done: + case <-time.After(3 * time.Second): + t.Fatal("watchdog did not finish after the lifecycle gate was released") + } + if tc.wantStart { + if len(got) != 1 || got[0].Err == nil || got[0].Reason != "start_failed" { + t.Fatalf("unchanged config was not restarted: %+v", got) + } + return + } + if len(got) != 0 || len(r.Names()) != 0 || looked.Load() == 0 { + t.Fatalf("stale snapshot started the driver: events=%+v names=%v revalidated=%d", got, r.Names(), looked.Load()) + } + }) + } +} + func TestRestartWatchdogStopsWhenContextCancelled(t *testing.T) { tel := telemetry.NewStore() r := NewRegistry(tel)