From a75da93381f431b30f0a97766ec017434ee63731 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Wed, 21 Oct 2026 09:52:47 +0200 Subject: [PATCH] Revert "fix(drivers): restart a driver that stays offline (#1628)" This reverts commit b271a763aa48489ab3577ab0ad2c5dc4b57084c0. #1628 has not shipped. Review found that it breaks the driver offline alert and restarts sleeping devices without limit: - A restart calls tel.Remove, which deletes the driver's health. The notifier skips a driver that has never succeeded, so the "driver offline" push never fires for a device that stays down. - Backoff ends at every 5 minutes with no cap. An inverter that sleeps at night is torn down and started again all night, and each restart sends driver_default_mode to a dead device and clears its readings. - A failed add of a cloud driver runs driver_init every 5 minutes. The problem it set out to fix stays real: a driver that fails to start at boot is never retried. A smaller change will retry only that case. Co-Authored-By: Claude Opus 5.5 --- .changeset/driver-restart-watchdog.md | 5 - go/cmd/ftw/main.go | 31 - 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 | 621 ------------------- go/internal/drivers/restart_watchdog_test.go | 525 ---------------- 7 files changed, 8 insertions(+), 1285 deletions(-) delete mode 100644 .changeset/driver-restart-watchdog.md delete mode 100644 go/internal/drivers/restart_watchdog.go delete mode 100644 go/internal/drivers/restart_watchdog_test.go diff --git a/.changeset/driver-restart-watchdog.md b/.changeset/driver-restart-watchdog.md deleted file mode 100644 index ed526fee0..000000000 --- a/.changeset/driver-restart-watchdog.md +++ /dev/null @@ -1,5 +0,0 @@ ---- -"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 883d4072d..e57a0316c 100644 --- a/go/cmd/ftw/main.go +++ b/go/cmd/ftw/main.go @@ -794,8 +794,6 @@ 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) } } @@ -840,35 +838,6 @@ func main() { controlRev := &control.Revision{} capMu := &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) { - // 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 - 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 5c226d1d8..d9e2a9b4b 100644 --- a/go/internal/api/api.go +++ b/go/internal/api/api.go @@ -799,36 +799,6 @@ 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 @@ -1166,9 +1136,6 @@ 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 @@ -1181,22 +1148,19 @@ 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 { - row = map[string]any{ + drivers[dc.Name] = 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 09d66d5e6..45f972fca 100644 --- a/go/internal/api/api_test.go +++ b/go/internal/api/api_test.go @@ -170,60 +170,6 @@ 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 TestHandleStatusKeepsOfflineFlowsUnavailable(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 de78e8586..07ea3bca4 100644 --- a/go/internal/drivers/registry.go +++ b/go/internal/drivers/registry.go @@ -122,11 +122,6 @@ 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 deleted file mode 100644 index 4a681ab72..000000000 --- a/go/internal/drivers/restart_watchdog.go +++ /dev/null @@ -1,621 +0,0 @@ -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. 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 - } - 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, func() []config.Driver { - current, _ := snapshot() - return current - }) - } - 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 { - 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 - } - 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, current); 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, current func() []config.Driver) (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 - } - // 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 - } - 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 - } -} - -// 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 -// 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 deleted file mode 100644 index c17c4c68a..000000000 --- a/go/internal/drivers/restart_watchdog_test.go +++ /dev/null @@ -1,525 +0,0 @@ -package drivers - -import ( - "context" - "sync/atomic" - "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 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) - 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 -}