diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index 524bf700..31f6f00a 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -1825,10 +1825,42 @@ func (d *Daemon) NodeID() uint32 { // them for SYN-handler enforcement. Called at startup and after IPC joins. func (d *Daemon) loadNetworkPolicies() { nets := d.nodeNetworks() - policies := make(map[uint16][]uint16, len(nets)) + + // Snapshot the current cache so a transient GetNetworkPolicy error + // carries the last-known policy FORWARD instead of dropping it. + // Dropping an entry makes isPortAllowed fall through to allow-all + // (empty list == no restriction), so a restricted network would go + // silently unrestricted on any registry hiccup — a fail-OPEN hole. + // A restriction is relaxed only by an explicit successful response + // that no longer lists it, never by a failed lookup. + d.netPolicyMu.RLock() + prior := make(map[uint16][]uint16, len(d.netPolicies)) + for k, v := range d.netPolicies { + prior[k] = v + } + d.netPolicyMu.RUnlock() + + results := make([]netPolicyResult, 0, len(nets)) for _, netID := range nets { - resp, err := d.regConn.GetNetworkPolicy(netID) + var resp map[string]interface{} + var err error + // Bounded retry so a transient blip on the very first load (when + // there is no prior to fall back to) still self-heals quickly. + for attempt := 0; attempt < 3; attempt++ { + if resp, err = d.regConn.GetNetworkPolicy(netID); err == nil { + break + } + time.Sleep(200 * time.Millisecond) + } if err != nil { + _, hadPrior := prior[netID] + slog.Warn("network policy load failed — retaining last-known policy, not failing open", + "net_id", netID, "had_prior", hadPrior, "err", err) + d.publishEvent("network.policy_load_failed", map[string]any{ + "net_id": netID, + "had_prior": hadPrior, + }) + results = append(results, netPolicyResult{netID: netID, err: err}) continue } portsRaw, _ := resp["allowed_ports"].([]interface{}) @@ -1838,15 +1870,45 @@ func (d *Daemon) loadNetworkPolicies() { ports = append(ports, uint16(f)) } } - if len(ports) > 0 { - policies[netID] = ports - } + results = append(results, netPolicyResult{netID: netID, ports: ports}) } + + newPolicies := mergeNetworkPolicies(prior, results) d.netPolicyMu.Lock() - d.netPolicies = policies + d.netPolicies = newPolicies d.netPolicyMu.Unlock() } +// netPolicyResult is one network's port-policy fetch outcome, fed to +// mergeNetworkPolicies. +type netPolicyResult struct { + netID uint16 + ports []uint16 + err error +} + +// mergeNetworkPolicies computes the new port-policy cache from the fetch +// results and the prior cache. A FAILED fetch retains the prior policy +// (fail-CLOSED — a transient registry error must not drop a restriction +// and silently relax the network to allow-all); a SUCCESSFUL fetch is +// stored verbatim, including an empty list, which explicitly means "no +// restriction" and is the only thing that relaxes an existing policy. +// Pure function so the fail-closed invariant is unit-testable without a +// live registry. +func mergeNetworkPolicies(prior map[uint16][]uint16, results []netPolicyResult) map[uint16][]uint16 { + policies := make(map[uint16][]uint16, len(results)) + for _, r := range results { + if r.err != nil { + if p, ok := prior[r.netID]; ok { + policies[r.netID] = p + } + continue + } + policies[r.netID] = r.ports + } + return policies +} + // isPortAllowed checks whether dstPort is permitted by the network's AllowedPorts // policy. Returns true if no restriction is set (empty list = all ports allowed). func (d *Daemon) isPortAllowed(netID uint16, port uint16) bool { diff --git a/pkg/daemon/rxwatchdog.go b/pkg/daemon/rxwatchdog.go index 3bc1b37e..128a4123 100644 --- a/pkg/daemon/rxwatchdog.go +++ b/pkg/daemon/rxwatchdog.go @@ -6,6 +6,8 @@ import ( "log/slog" "math/rand" "os" + "strconv" + "strings" "sync/atomic" "time" ) @@ -82,6 +84,21 @@ const ( // Non-zero so launchd KeepAlive/SuccessfulExit=false respawns; a // distinctive value so `pilotctl`/operators can attribute the restart. rxWedgeExitCode = 86 + + // rxWedgeLoopWindow bounds how far back the restart-loop circuit + // breaker counts prior hard exits. Exits older than this don't count. + rxWedgeLoopWindow = 2 * time.Hour + + // rxWedgeLoopMax is how many hard exits within rxWedgeLoopWindow are + // tolerated before the breaker OPENS and further exits are withheld. + // The hard exit assumes a restart CLEARS the wedge (stale NAT/relay + // mapping). When it doesn't — a NAT/firewall/topology change a + // restart can't fix — the daemon would otherwise boot-loop every + // ~30-40 min forever (wedge → 3 soft attempts → exit → respawn → + // wedge → …), each restart dropping every live tunnel for nothing. + // After this many restarts the breaker keeps the daemon UP and + // soft-recovering instead, and surfaces the condition for an operator. + rxWedgeLoopMax = 3 ) // rxWatchdogExit is swapped by tests to observe the hard escalation @@ -107,6 +124,7 @@ const ( rxActionSoftRecover rxWatchdogAction = "soft-recover" rxActionExitWithheld rxWatchdogAction = "exit-withheld" rxActionExit rxWatchdogAction = "exit" + rxActionRestartLoop rxWatchdogAction = "restart-loop-withheld" ) func (d *Daemon) rxWatchdogLoop() { @@ -261,6 +279,30 @@ func (d *Daemon) rxWatchdogTick(st *rxWatchdogState, now time.Time) (action rxWa "silent_for", silence.Truncate(time.Second).String(), "uptime", uptime.Truncate(time.Second).String()) default: + // Restart-loop circuit breaker. The hard exit is premised on a + // restart CLEARING the wedge; when it doesn't (a NAT/firewall/ + // topology change no restart can fix), exiting just boot-loops + // the daemon and drops every live tunnel each cycle. If we have + // already respawned rxWedgeLoopMax times within rxWedgeLoopWindow + // (tracked in a sidecar file next to the identity so it survives + // the respawn), stop exiting: stay up, keep soft-recovering, and + // surface the condition for an operator. + exitLog := rxWedgeExitLogPath(d.config.IdentityPath) + if recent := recentRxWedgeExits(exitLog, now, rxWedgeLoopWindow); len(recent) >= rxWedgeLoopMax { + slog.Error("transport wedged but restart-loop detected — withholding exit, staying up to soft-recover", + "reason", wedgeReason, + "recent_exits", len(recent), + "window", rxWedgeLoopWindow.String(), + "silent_for", silence.Truncate(time.Second).String()) + d.publishEvent("tunnel.rx_wedge_restart_loop", map[string]any{ + "reason": wedgeReason, + "recent_exits": len(recent), + "window_seconds": int64(rxWedgeLoopWindow.Seconds()), + "silent_for_seconds": int64(silence.Seconds()), + }) + st.softAttempts = 0 + return rxActionRestartLoop + } slog.Error("transport wedged — exiting for supervisor respawn", "reason", wedgeReason, "silent_for", silence.Truncate(time.Second).String(), @@ -277,6 +319,9 @@ func (d *Daemon) rxWatchdogTick(st *rxWatchdogState, now time.Time) (action rxWa "soft_attempts": st.softAttempts, "exit_code": rxWedgeExitCode, }) + // Record this exit BEFORE calling exit so the respawned process + // sees it in the recent-exits count. + recordRxWedgeExit(exitLog, now, rxWedgeLoopWindow) rxWatchdogExit(rxWedgeExitCode) return rxActionExit } @@ -285,3 +330,66 @@ func (d *Daemon) rxWatchdogTick(st *rxWatchdogState, now time.Time) (action rxWa st.softAttempts = 0 return rxActionExitWithheld } + +// rxWedgeExitLogPath returns the sidecar file that records recent hard- +// exit timestamps, stored next to the identity file so the count +// survives the respawn. An empty identityPath (no persistence) returns +// "", which disables the restart-loop breaker — the daemon behaves +// exactly as before (always allowed to exit). +func rxWedgeExitLogPath(identityPath string) string { + if identityPath == "" { + return "" + } + return identityPath + ".rxwedge" +} + +// recentRxWedgeExits reads hard-exit timestamps from path and returns +// those within window of now. A missing or unreadable file yields an +// empty slice (breaker closed — exit allowed); malformed lines are +// skipped. Never errors: a read problem must not itself block a +// legitimate wedge exit. +func recentRxWedgeExits(path string, now time.Time, window time.Duration) []int64 { + if path == "" { + return nil + } + // #nosec G304 -- path is derived from the daemon's own configured + // IdentityPath (rxWedgeExitLogPath), not from any peer/user input. + data, err := os.ReadFile(path) + if err != nil { + return nil + } + cutoff := now.Add(-window).UnixNano() + var recent []int64 + for _, line := range strings.Split(strings.TrimSpace(string(data)), "\n") { + line = strings.TrimSpace(line) + if line == "" { + continue + } + ts, err := strconv.ParseInt(line, 10, 64) + if err != nil { + continue + } + if ts >= cutoff { + recent = append(recent, ts) + } + } + return recent +} + +// recordRxWedgeExit appends now to the exit log (pruned to window) so the +// respawned process sees an accurate recent-exit count. Best-effort: a +// failed write just under-counts, degrading to the prior always-restart +// behavior — never worse. +func recordRxWedgeExit(path string, now time.Time, window time.Duration) { + if path == "" { + return + } + recent := recentRxWedgeExits(path, now, window) + recent = append(recent, now.UnixNano()) + var b strings.Builder + for _, ts := range recent { + b.WriteString(strconv.FormatInt(ts, 10)) + b.WriteByte('\n') + } + _ = os.WriteFile(path, []byte(b.String()), 0o600) +} diff --git a/pkg/daemon/transport/wss/wss.go b/pkg/daemon/transport/wss/wss.go index b45e9f9b..13619f01 100644 --- a/pkg/daemon/transport/wss/wss.go +++ b/pkg/daemon/transport/wss/wss.go @@ -55,6 +55,16 @@ import ( // long enough not to flood the link. const DefaultIdlePingInterval = 30 * time.Second +// DefaultIdlePingTimeout bounds how long a single idle keepalive ping +// may wait for its pong before the connection is treated as half-open. +// On a silently-dead conn (stale NAT/relay mapping, half-open TCP) the +// pong never arrives; without this bound the ping blocks forever, the +// keepalive loop wedges, and the daemon never notices the beacon is +// gone. Kept well under DefaultIdlePingInterval so a slow-but-live link +// is not falsely reconnected. This is the WSS-transport analogue of the +// L4 rx-watchdog: detect a silently-dead inbound path and recover it. +const DefaultIdlePingTimeout = 10 * time.Second + // DefaultDialTimeout caps the time we spend on a single dial attempt // (DNS + TCP + TLS + WS upgrade + auth challenge). Beyond this we // fail fast and let the reconnect loop try again. @@ -110,6 +120,10 @@ type Config struct { // IdlePingInterval overrides DefaultIdlePingInterval. IdlePingInterval time.Duration + // IdlePingTimeout overrides DefaultIdlePingTimeout — the per-ping + // deadline that turns a silently-dead conn into a reconnect. + IdlePingTimeout time.Duration + // DialTimeout overrides DefaultDialTimeout. DialTimeout time.Duration @@ -193,6 +207,9 @@ func Dial(ctx context.Context, cfg Config) (*Transport, error) { if cfg.IdlePingInterval == 0 { cfg.IdlePingInterval = DefaultIdlePingInterval } + if cfg.IdlePingTimeout == 0 { + cfg.IdlePingTimeout = DefaultIdlePingTimeout + } if cfg.DialTimeout == 0 { cfg.DialTimeout = DefaultDialTimeout } @@ -552,6 +569,16 @@ func (t *Transport) drainReads(conn *websocket.Conn) { // to keep the WSS connection alive through proxy idle timeouts. Runs until // Close() fires via lifetimeCtx. Skips pings while the supervisor is // reconnecting (conn == nil). +// +// Each ping is bounded by IdlePingTimeout. A pong that never arrives — +// the signature of a silently half-open conn (stale NAT/relay mapping, +// half-open TCP the kernel hasn't torn down) — would otherwise block +// this loop forever while it holds writeMu, freezing every Send behind +// it and leaving the dead beacon conn undetected. On a ping error we +// force the conn closed: that unblocks the supervisor's drainReads Read, +// which drives the reconnect. Without this, a half-open beacon link is +// invisible to the transport (Send only fails once there is traffic to +// send, and drainReads sits blocked on a Read that never returns). func (t *Transport) idlePing() { ticker := time.NewTicker(t.cfg.IdlePingInterval) defer ticker.Stop() @@ -569,11 +596,27 @@ func (t *Transport) idlePing() { if conn == nil { continue } + pingCtx, cancel := context.WithTimeout(t.lifetimeCtx, t.cfg.IdlePingTimeout) t.writeMu.Lock() - err := conn.Ping(t.lifetimeCtx) + err := conn.Ping(pingCtx) t.writeMu.Unlock() + cancel() if err != nil { - slog.Debug("wss transport: idle ping failed", "err", err) + if t.closed.Load() { + return + } + // Half-open conn: the pong timed out. Force the conn + // closed so the supervisor's blocked drainReads Read + // returns an error and the reconnect sequence fires. + // Also clear t.conn so Send fails fast meanwhile. + slog.Warn("wss transport: idle ping timed out — forcing reconnect", + "err", err, "timeout", t.cfg.IdlePingTimeout) + t.connMu.Lock() + if t.conn == conn { + t.conn = nil + } + t.connMu.Unlock() + _ = conn.Close(websocket.StatusPolicyViolation, "idle ping timeout") } } } diff --git a/pkg/daemon/transport/wss/zz_wss_test.go b/pkg/daemon/transport/wss/zz_wss_test.go index d0b4573f..f09bdf23 100644 --- a/pkg/daemon/transport/wss/zz_wss_test.go +++ b/pkg/daemon/transport/wss/zz_wss_test.go @@ -14,6 +14,7 @@ import ( "net/http" "net/http/httptest" "strings" + "sync/atomic" "testing" "time" @@ -59,6 +60,19 @@ type fakeBeacon struct { // the underlying TCP without completing the WS upgrade — same // shape as nginx returning 502 while the beacon process restarts. failNextDials int + + // stallConns, when >0, makes handle() go SILENTLY HALF-OPEN after + // completing auth on that many connections (counter decrements): it + // stops calling conn.Read, so the coder/websocket library never + // auto-responds to the client's keepalive pings, while the TCP stays + // up. This is the half-open beacon link the F3 idle-ping-timeout fix + // must detect and reconnect around. + stallConns atomic.Int32 + + // authCount counts connections that completed auth. Lets tests wait + // for a reconnect (authCount advancing past 1) without polling + // client internals. + authCount atomic.Int32 } func newFakeBeacon(t *testing.T, expectedNodeID uint32) *fakeBeacon { @@ -167,6 +181,20 @@ func (fb *fakeBeacon) handle(w http.ResponseWriter, r *http.Request) { if err := conn.Write(ctx, websocket.MessageText, okBytes); err != nil { return } + fb.authCount.Add(1) + + // Half-open simulation: go silent (never Read → never auto-pong) + // while holding the TCP open, so the client's bounded idle ping must + // time out and force a reconnect. Bounded so the goroutine can't + // outlive the test if the client's close doesn't cancel r.Context(). + if fb.stallConns.Load() > 0 { + fb.stallConns.Add(-1) + select { + case <-r.Context().Done(): + case <-time.After(3 * time.Second): + } + return + } // Echo loop: binary frames in → binary frames out (prefix "echo:") frameCount := 0 @@ -644,6 +672,77 @@ func TestIdlePing_KeepsConnectionAlive(t *testing.T) { } } +// TestIdlePing_HalfOpenConnTriggersReconnect exercises the F3 fix: when +// a beacon conn goes silently half-open (keepalive pings get no pong), +// the bounded idle ping times out, forces the dead conn closed, and the +// supervisor reconnects — restoring Send/Recv. Before the fix the ping +// blocked on lifetimeCtx forever while holding writeMu, so the half-open +// link was never detected and every Send wedged behind it. +func TestIdlePing_HalfOpenConnTriggersReconnect(t *testing.T) { + t.Parallel() + id := mustID(t) + fb := newFakeBeacon(t, 1) + fb.stallConns.Store(1) // first post-auth conn goes half-open + + tr, err := wss.Dial(context.Background(), wss.Config{ + URL: fb.url(), + TLSConfig: fb.tlsConfig(), + Identity: id, + NodeID: 1, + IdlePingInterval: 40 * time.Millisecond, + IdlePingTimeout: 60 * time.Millisecond, + }) + if err != nil { + t.Fatalf("Dial: %v", err) + } + defer tr.Close() + + // The first conn never pongs: idle ping fires (~40ms), waits up to + // 60ms, times out, forces reconnect. A second successful auth means + // a fresh healthy conn is up. + deadline := time.Now().Add(5 * time.Second) + for fb.authCount.Load() < 2 { + if time.Now().After(deadline) { + t.Fatalf("no reconnect after half-open conn (authCount=%d) — idle ping did not detect the dead link", + fb.authCount.Load()) + } + time.Sleep(20 * time.Millisecond) + } + + // Round-trip on the reconnected conn. Retry Send past the brief + // ErrReconnecting window; guard Recv with its own timeout so a + // regression can't hang the test. + var sendOK bool + for time.Now().Before(deadline) { + if _, err := tr.Send([]byte("after-reconnect"), nil); err == nil { + sendOK = true + break + } + time.Sleep(20 * time.Millisecond) + } + if !sendOK { + t.Fatal("Send never succeeded after reconnect") + } + + type recvRes struct { + b []byte + err error + } + ch := make(chan recvRes, 1) + go func() { b, _, err := tr.Recv(); ch <- recvRes{b, err} }() + select { + case r := <-ch: + if r.err != nil { + t.Fatalf("Recv after reconnect: %v", r.err) + } + if want := "echo:after-reconnect"; string(r.b) != want { + t.Errorf("Recv = %q; want %q", r.b, want) + } + case <-time.After(2 * time.Second): + t.Fatal("Recv timed out after reconnect — transport did not recover") + } +} + func mustID(t *testing.T) *crypto.Identity { t.Helper() id, err := crypto.GenerateIdentity() diff --git a/pkg/daemon/zz_netpolicy_failclosed_test.go b/pkg/daemon/zz_netpolicy_failclosed_test.go new file mode 100644 index 00000000..783dfd6b --- /dev/null +++ b/pkg/daemon/zz_netpolicy_failclosed_test.go @@ -0,0 +1,62 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package daemon + +import ( + "errors" + "testing" +) + +// TestMergeNetworkPolicies_FailClosed pins the security-critical +// invariant behind loadNetworkPolicies: a failed GetNetworkPolicy must +// NOT drop an existing restriction (which isPortAllowed would read as +// allow-all), while a successful fetch — including an empty list — is +// authoritative and does relax the policy. +func TestMergeNetworkPolicies_FailClosed(t *testing.T) { + boom := errors.New("registry unreachable") + + t.Run("error retains prior restriction (no fail-open)", func(t *testing.T) { + prior := map[uint16][]uint16{7: {80, 443}} + got := mergeNetworkPolicies(prior, []netPolicyResult{{netID: 7, err: boom}}) + if p, ok := got[7]; !ok || len(p) != 2 || p[0] != 80 || p[1] != 443 { + t.Fatalf("restricted net 7 must retain [80 443] on load error, got %v (present=%v)", p, ok) + } + }) + + t.Run("successful empty relaxes restriction", func(t *testing.T) { + prior := map[uint16][]uint16{7: {80}} + got := mergeNetworkPolicies(prior, []netPolicyResult{{netID: 7, ports: nil}}) + if p := got[7]; len(p) != 0 { + t.Fatalf("net 7 should be relaxed to no-restriction on successful empty, got %v", p) + } + }) + + t.Run("successful update replaces prior", func(t *testing.T) { + prior := map[uint16][]uint16{7: {80}} + got := mergeNetworkPolicies(prior, []netPolicyResult{{netID: 7, ports: []uint16{22}}}) + if p := got[7]; len(p) != 1 || p[0] != 22 { + t.Fatalf("net 7 should become [22], got %v", p) + } + }) + + t.Run("error with no prior stays absent (allow-all, unchanged behavior)", func(t *testing.T) { + got := mergeNetworkPolicies(map[uint16][]uint16{}, []netPolicyResult{{netID: 9, err: boom}}) + if _, ok := got[9]; ok { + t.Fatalf("net 9 with no prior + load error must stay absent, got present") + } + }) + + t.Run("mixed: one fails (retained), one succeeds (updated)", func(t *testing.T) { + prior := map[uint16][]uint16{7: {80, 443}, 8: {22}} + got := mergeNetworkPolicies(prior, []netPolicyResult{ + {netID: 7, err: boom}, + {netID: 8, ports: []uint16{2222}}, + }) + if p := got[7]; len(p) != 2 { + t.Fatalf("net 7 (errored) must retain prior [80 443], got %v", p) + } + if p := got[8]; len(p) != 1 || p[0] != 2222 { + t.Fatalf("net 8 (ok) must update to [2222], got %v", p) + } + }) +} diff --git a/pkg/daemon/zz_rx_watchdog_restartloop_test.go b/pkg/daemon/zz_rx_watchdog_restartloop_test.go new file mode 100644 index 00000000..364f1d6c --- /dev/null +++ b/pkg/daemon/zz_rx_watchdog_restartloop_test.go @@ -0,0 +1,156 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package daemon + +import ( + "os" + "path/filepath" + "strconv" + "strings" + "testing" + "time" +) + +// primeExitLog writes the given exit ages (relative to now) into the +// watchdog exit-log file so a test can drive the restart-loop breaker. +func primeExitLog(t *testing.T, path string, now time.Time, ages ...time.Duration) { + t.Helper() + var b strings.Builder + for _, age := range ages { + b.WriteString(strconv.FormatInt(now.Add(-age).UnixNano(), 10)) + b.WriteByte('\n') + } + if err := os.WriteFile(path, []byte(b.String()), 0o600); err != nil { + t.Fatalf("prime exit log: %v", err) + } +} + +// TestRxWatchdogRestartLoopBreakerOpens: once rxWedgeLoopMax hard exits +// have already happened within the window, a further wedge withholds the +// exit (restarting isn't clearing it) and keeps the daemon up. +func TestRxWatchdogRestartLoopBreakerOpens(t *testing.T) { + d := newRxWatchdogTestDaemon(t) + exitCode := swapExitForTest(t) + now := time.Now() + d.config.IdentityPath = filepath.Join(t.TempDir(), "id") + d.lastRegistryOKNano.Store(now.UnixNano()) // fresh registry: wedge signature + + // rxWedgeLoopMax recent exits already recorded. + ages := make([]time.Duration, rxWedgeLoopMax) + for i := range ages { + ages[i] = time.Duration(i+1) * time.Minute + } + primeExitLog(t, rxWedgeExitLogPath(d.config.IdentityPath), now, ages...) + + st := wedgeState(d, now) + if got := d.rxWatchdogTick(st, now); got != rxActionRestartLoop { + t.Fatalf("action = %q, want %q (breaker should be open)", got, rxActionRestartLoop) + } + if *exitCode != 0 { + t.Fatalf("breaker open but exit fired (code %d)", *exitCode) + } + if st.softAttempts != 0 { + t.Fatalf("withheld exit must reset softAttempts, got %d", st.softAttempts) + } +} + +// TestRxWatchdogRestartLoopBreakerClosedExitsAndRecords: below the +// threshold the exit still fires, and it is recorded so the respawned +// process sees an incremented count. +func TestRxWatchdogRestartLoopBreakerClosedExitsAndRecords(t *testing.T) { + d := newRxWatchdogTestDaemon(t) + exitCode := swapExitForTest(t) + now := time.Now() + d.config.IdentityPath = filepath.Join(t.TempDir(), "id") + d.lastRegistryOKNano.Store(now.UnixNano()) + + path := rxWedgeExitLogPath(d.config.IdentityPath) + // One fewer than the max: this exit is allowed and becomes the max-th. + ages := make([]time.Duration, rxWedgeLoopMax-1) + for i := range ages { + ages[i] = time.Duration(i+1) * time.Minute + } + primeExitLog(t, path, now, ages...) + + st := wedgeState(d, now) + if got := d.rxWatchdogTick(st, now); got != rxActionExit { + t.Fatalf("action = %q, want %q (breaker should be closed)", got, rxActionExit) + } + if *exitCode != rxWedgeExitCode { + t.Fatalf("exit code = %d, want %d", *exitCode, rxWedgeExitCode) + } + // The exit must have been recorded: count is now rxWedgeLoopMax. + if n := len(recentRxWedgeExits(path, now, rxWedgeLoopWindow)); n != rxWedgeLoopMax { + t.Fatalf("recorded exits = %d, want %d", n, rxWedgeLoopMax) + } +} + +// TestRxWatchdogRestartLoopIgnoresOldExits: exits older than the window +// do not count, so a daemon that wedged long ago still gets to restart. +func TestRxWatchdogRestartLoopIgnoresOldExits(t *testing.T) { + d := newRxWatchdogTestDaemon(t) + exitCode := swapExitForTest(t) + now := time.Now() + d.config.IdentityPath = filepath.Join(t.TempDir(), "id") + d.lastRegistryOKNano.Store(now.UnixNano()) + + // rxWedgeLoopMax exits, all OUTSIDE the window. + ages := make([]time.Duration, rxWedgeLoopMax) + for i := range ages { + ages[i] = rxWedgeLoopWindow + time.Duration(i+1)*time.Minute + } + primeExitLog(t, rxWedgeExitLogPath(d.config.IdentityPath), now, ages...) + + st := wedgeState(d, now) + if got := d.rxWatchdogTick(st, now); got != rxActionExit { + t.Fatalf("action = %q, want %q (stale exits must not open the breaker)", got, rxActionExit) + } + if *exitCode != rxWedgeExitCode { + t.Fatalf("exit should fire when prior exits are outside window") + } +} + +// TestRxWatchdogRestartLoopDisabledWithoutPersistence: an empty identity +// path (no persistence) leaves the breaker off — behaviour is exactly as +// before (exit always allowed). +func TestRxWatchdogRestartLoopDisabledWithoutPersistence(t *testing.T) { + d := newRxWatchdogTestDaemon(t) + exitCode := swapExitForTest(t) + now := time.Now() + // d.config.IdentityPath stays "". + d.lastRegistryOKNano.Store(now.UnixNano()) + + st := wedgeState(d, now) + if got := d.rxWatchdogTick(st, now); got != rxActionExit { + t.Fatalf("action = %q, want %q (no persistence disables the breaker)", got, rxActionExit) + } + if *exitCode != rxWedgeExitCode { + t.Fatalf("exit should fire when persistence is off") + } +} + +// TestRecentRxWedgeExits_WindowAndMalformed pins the sidecar-file parser. +func TestRecentRxWedgeExits_WindowAndMalformed(t *testing.T) { + now := time.Now() + path := filepath.Join(t.TempDir(), "id.rxwedge") + content := strings.Join([]string{ + strconv.FormatInt(now.Add(-1*time.Minute).UnixNano(), 10), // in window + "garbage-not-a-number", // skipped + strconv.FormatInt(now.Add(-3*time.Hour).UnixNano(), 10), // outside window + "", // skipped + strconv.FormatInt(now.Add(-5*time.Minute).UnixNano(), 10), // in window + }, "\n") + if err := os.WriteFile(path, []byte(content), 0o600); err != nil { + t.Fatal(err) + } + if n := len(recentRxWedgeExits(path, now, rxWedgeLoopWindow)); n != 2 { + t.Fatalf("recent count = %d, want 2 (2 in-window, 1 stale, 2 junk)", n) + } + // Missing file and empty path both yield zero, never error. + if n := len(recentRxWedgeExits(filepath.Join(t.TempDir(), "nope"), now, rxWedgeLoopWindow)); n != 0 { + t.Fatalf("missing file should yield 0, got %d", n) + } + if n := len(recentRxWedgeExits("", now, rxWedgeLoopWindow)); n != 0 { + t.Fatalf("empty path should yield 0, got %d", n) + } +}