diff --git a/docs/tools/lsp/index.md b/docs/tools/lsp/index.md index 58133cba7b..97adbadeff 100644 --- a/docs/tools/lsp/index.md +++ b/docs/tools/lsp/index.md @@ -197,7 +197,7 @@ Available Capabilities: LSP toolsets are managed by the same supervisor as MCP toolsets, so a crashed `gopls` (or any other language server) is reconnected automatically with exponential backoff. Use the [`lifecycle`](../../configuration/tools/index.md#toolset-lifecycle) block to tune the policy per toolset — for example, mark `gopls` as `strict` if your CI flow requires it to be available, or use `/toolset-restart gopls` from the TUI to force a reconnect when the server gets stuck. -**Startup failure behaviour:** missing-binary and bad-config failures fail fast — each turn retries immediately with no artificial delay. A language server that crash-loops is not currently paced by the backoff gate; the supervisor's own reconnect policy (controlled by the `lifecycle` block) is the primary throttle for crash recovery. +**Startup failure behaviour:** missing-binary and bad-config failures fail fast — each turn retries immediately with no artificial delay. A language server that crash-loops (3 crashes within 1 minute, by default) is different: the supervisor stops auto-restarting and reports the loop instead. The next attempt to use the toolset (a turn's start, a tool call, or `/toolset-restart`) surfaces that report rather than relaunching the server, and the backoff gate then paces subsequent attempts (15s, doubling up to 5 minutes) the same way it paces a rate-limited MCP server — so a server that dies right after every restart no longer relaunches at full speed. This applies regardless of `profile`: even a `strict` toolset is retried on the next turn once its window elapses, rather than staying down until an explicit `/toolset-restart`. ```yaml toolsets: diff --git a/pkg/tools/builtin/lsp/lsp.go b/pkg/tools/builtin/lsp/lsp.go index 43b9e049d8..617704e323 100644 --- a/pkg/tools/builtin/lsp/lsp.go +++ b/pkg/tools/builtin/lsp/lsp.go @@ -60,9 +60,10 @@ type ToolSet struct { // Verify interface compliance var ( - _ tools.ToolSet = (*ToolSet)(nil) - _ tools.Startable = (*ToolSet)(nil) - _ tools.Instructable = (*ToolSet)(nil) + _ tools.ToolSet = (*ToolSet)(nil) + _ tools.Startable = (*ToolSet)(nil) + _ tools.Instructable = (*ToolSet)(nil) + _ tools.StartReporter = (*ToolSet)(nil) ) type lspHandler struct { @@ -431,6 +432,20 @@ func (t *ToolSet) Stop(ctx context.Context) error { return t.handler.supervisor.Stop(ctx) } +// IsStarted implements tools.StartReporter: reports whether the supervisor +// requires external action (Start/Restart) to serve requests again. +// Deliberately looser than the MCP toolset's IsStarted (which tracks +// Ready/Degraded only): a transient Restarting still reports true here, +// since the supervisor already self-heals a one-off crash on its own +// watcher goroutine and every per-request call (ensureInitialized) retries +// eagerly regardless. Only a give-up — Failed (crash loop, exhausted +// restarts) or Stopped — reports false, so StartableToolSet gets involved +// (via Restart) exactly when the supervisor needs a caller-paced retry, not +// on every ordinary transient reconnect. +func (t *ToolSet) IsStarted() bool { + return !t.handler.supervisor.State().State.IsTerminal() +} + // State returns a snapshot of the underlying supervisor's lifecycle state, // suitable for the /tools dialog and lifecycle log messages. func (t *ToolSet) State() lifecycle.StateInfo { @@ -712,7 +727,17 @@ func (h *lspHandler) ensureInitialized(ctx context.Context) error { // Lazy-start through the supervisor. Concurrent ensureInitialized // callers serialize inside Supervisor.Start. + // + // A pending crash-loop report is checked first and, if present, + // returned as-is without calling Start: this per-request path bypasses + // StartableToolSet's backoff gate entirely (it isn't the wrapper's + // paced TryStart), so it must not be the one to consume — and thereby + // reconnect on behalf of — a one-shot report meant for the gate to + // pace. Only the gate's own eventual Start call clears it. if !h.supervisor.IsReady() { + if err := h.supervisor.PendingCrashLoopError(); err != nil { + return fmt.Errorf("failed to start LSP server: %w", err) + } if err := h.supervisor.Start(ctx); err != nil { return fmt.Errorf("failed to start LSP server: %w", err) } diff --git a/pkg/tools/builtin/lsp/lsp_crashloop_test.go b/pkg/tools/builtin/lsp/lsp_crashloop_test.go new file mode 100644 index 0000000000..6e4a5ba305 --- /dev/null +++ b/pkg/tools/builtin/lsp/lsp_crashloop_test.go @@ -0,0 +1,230 @@ +package lsp + +import ( + "bufio" + "encoding/json" + "fmt" + "io" + "os" + "path/filepath" + "strconv" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/docker/docker-agent/pkg/tools" + "github.com/docker/docker-agent/pkg/tools/lifecycle" +) + +// fakeLSPServerEnv names the environment variable that tells this test +// binary, when re-executed as a subprocess, to behave as a fake LSP server +// instead of running the Go test suite. Its value is the path of a log +// file the fake server appends one line to on every spawn, so tests can +// count how many times the supervisor actually spawned a new process. +const fakeLSPServerEnv = "DOCKER_AGENT_LSP_TEST_FAKE_SERVER_LOG" + +// TestMain lets this test binary re-exec itself (os.Args[0]) as a fake LSP +// server: exec.Command needs a real, portable executable, and the test +// binary itself is the simplest one available on every platform CI runs on. +func TestMain(m *testing.M) { + if logPath := os.Getenv(fakeLSPServerEnv); logPath != "" { + runFakeCrashingLSPServer(logPath) + return // unreachable: runFakeCrashingLSPServer always calls os.Exit. + } + os.Exit(m.Run()) +} + +// runFakeCrashingLSPServer answers the initialize/initialized handshake +// exactly once, records that it ran, then exits non-zero to simulate a +// crash right after startup — every time it is spawned. +func runFakeCrashingLSPServer(logPath string) { + if f, err := os.OpenFile(logPath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o600); err == nil { + fmt.Fprintln(f, "spawn") + _ = f.Close() + } + + r := bufio.NewReader(os.Stdin) + if body, err := readFramedMessage(r); err == nil { + var req struct { + ID int64 `json:"id"` + Method string `json:"method"` + } + if json.Unmarshal(body, &req) == nil && req.Method == "initialize" { + resp, _ := json.Marshal(map[string]any{ + "jsonrpc": "2.0", + "id": req.ID, + "result": map[string]any{"capabilities": map[string]any{}}, + }) + if writeFramedMessage(os.Stdout, resp) == nil { + _, _ = readFramedMessage(r) // the "initialized" notification; ignored. + } + } + } + os.Exit(1) +} + +// readFramedMessage and writeFramedMessage mirror the Content-Length +// framing lspHandler itself speaks (see readMessageLocked/writeMessageLocked). +func readFramedMessage(r *bufio.Reader) ([]byte, error) { + contentLength := 0 + for { + line, err := r.ReadString('\n') + if err != nil { + return nil, err + } + line = strings.TrimSpace(line) + if line == "" { + break + } + if after, ok := strings.CutPrefix(line, "Content-Length:"); ok { + contentLength, err = strconv.Atoi(strings.TrimSpace(after)) + if err != nil { + return nil, err + } + } + } + body := make([]byte, contentLength) + if _, err := io.ReadFull(r, body); err != nil { + return nil, err + } + return body, nil +} + +func writeFramedMessage(w io.Writer, data []byte) error { + if _, err := fmt.Fprintf(w, "Content-Length: %d\r\n\r\n", len(data)); err != nil { + return err + } + _, err := w.Write(data) + return err +} + +// countSpawns counts lines in the fake server's spawn log, i.e. how many +// times it has actually been launched as a subprocess. +func countSpawns(t *testing.T, path string) int { + t.Helper() + data, err := os.ReadFile(path) + if err != nil { + if os.IsNotExist(err) { + return 0 + } + t.Fatalf("failed to read spawn log: %v", err) + } + trimmed := strings.TrimSpace(string(data)) + if trimmed == "" { + return 0 + } + return len(strings.Split(trimmed, "\n")) +} + +// TestLSPTool_CrashLoopArmsBackoffGate drives a real ToolSet, wrapped in +// tools.StartableToolSet exactly as production wires it, against a fake +// LSP server that completes the handshake and then exits non-zero every +// time it is spawned. It proves the whole chain end to end: a sustained +// crash loop stops the supervisor's own auto-restart, TryStart surfaces +// lifecycle.ErrCrashLooping, and the backoff gate then withholds further +// spawns until its window elapses. +func TestLSPTool_CrashLoopArmsBackoffGate(t *testing.T) { + t.Parallel() + + spawnLog := filepath.Join(t.TempDir(), "spawns.log") + env := []string{fakeLSPServerEnv + "=" + spawnLog} + + policy := lifecycle.Policy{ + Backoff: lifecycle.Backoff{Initial: 2 * time.Millisecond, Max: 5 * time.Millisecond, Multiplier: 2}, + CrashLoop: lifecycle.CrashLoop{Threshold: 3, Window: time.Minute}, + } + tool := New(os.Args[0], nil, env, t.TempDir(), policy) + s := tools.NewStartable(tool, tools.WithStartRetryJitter(func(d time.Duration) time.Duration { return d })) + t.Cleanup(func() { _ = s.Stop(t.Context()) }) + + started, err := s.TryStart(t.Context()) + require.NoError(t, err) + require.True(t, started) + + // The fake server crashes right after the handshake every time; wait + // for the crash-loop detector to give up (state -> Failed) rather than + // racing the background watcher's own restart attempts. + require.Eventually(t, func() bool { + return tool.State().State == lifecycle.StateFailed + }, 10*time.Second, 5*time.Millisecond, "supervisor did not detect the crash loop") + + spawnsAtLoop := countSpawns(t, spawnLog) + assert.Equal(t, 3, spawnsAtLoop, "the loop must trip at exactly CrashLoop.Threshold spawns, no more") + + // The next TryStart reports the loop instead of relaunching. + _, err = s.TryStart(t.Context()) + require.ErrorIs(t, err, lifecycle.ErrCrashLooping) + assert.Equal(t, spawnsAtLoop, countSpawns(t, spawnLog), "the crash-loop report must not itself spawn a server") + + // Gate now armed: an immediate retry must not spawn either. + _, err = s.TryStart(t.Context()) + require.Error(t, err) + assert.Equal(t, spawnsAtLoop, countSpawns(t, spawnLog), "gate must withhold the next spawn until its window elapses") +} + +// TestLSPTool_CrashLoopPendingReportBlocksDirectToolCalls verifies that a +// tool call reaching the LSP handler while a crash-loop report is pending +// — ensureInitialized's lazy per-request path, which bypasses +// tools.StartableToolSet entirely — fails fast on the same error without +// spawning a new server, and without consuming the one-shot report: only +// the wrapper's own paced retry (Start, once its backoff window elapses) +// may do that. +func TestLSPTool_CrashLoopPendingReportBlocksDirectToolCalls(t *testing.T) { + t.Parallel() + + spawnLog := filepath.Join(t.TempDir(), "spawns.log") + env := []string{fakeLSPServerEnv + "=" + spawnLog} + + policy := lifecycle.Policy{ + Backoff: lifecycle.Backoff{Initial: 2 * time.Millisecond, Max: 5 * time.Millisecond, Multiplier: 2}, + CrashLoop: lifecycle.CrashLoop{Threshold: 3, Window: time.Minute}, + } + tool := New(os.Args[0], nil, env, t.TempDir(), policy) + t.Cleanup(func() { _ = tool.Stop(t.Context()) }) + + require.NoError(t, tool.Start(t.Context())) + + require.Eventually(t, func() bool { + return tool.State().State == lifecycle.StateFailed + }, 10*time.Second, 5*time.Millisecond, "supervisor did not detect the crash loop") + + spawnsAtLoop := countSpawns(t, spawnLog) + assert.Equal(t, 3, spawnsAtLoop) + + // ensureInitialized has a fast path keyed on the atomic `initialized` + // flag, which a raw crash (detected only in the background watcher) + // does not clear — only a fresh Connect or an explicit Close does. That + // pre-existing gap (independent of crash-loop pacing; it also affects + // the ordinary exhausted-restart give-up) means a tool call arriving + // immediately after this crash would still see the stale flag and skip + // the check below entirely. Clear it here to exercise the check as it + // would run once that flag correctly reflects the disconnect. + tool.handler.initialized.Store(false) + + // A tool call arriving now goes through ensureInitialized, not + // StartableToolSet.TryStart. It must see the same error and must not + // spawn a server on its own. + err := tool.handler.ensureInitialized(t.Context()) + require.ErrorIs(t, err, lifecycle.ErrCrashLooping) + assert.Equal(t, spawnsAtLoop, countSpawns(t, spawnLog), "a direct tool call must not spawn while a crash-loop report is pending") + + // The report must still be pending for the next caller: a direct call + // must not have consumed it. + require.ErrorIs(t, tool.handler.supervisor.PendingCrashLoopError(), lifecycle.ErrCrashLooping) + + // Only the "real" gated caller (standing in for StartableToolSet's + // paced Restart/Start once its window elapses) may consume it and + // reconnect for real: the very next Start still reports it once + // (ensureInitialized's peek above did not consume it), and the Start + // after that performs the genuine reconnect. + err = tool.Start(t.Context()) + require.ErrorIs(t, err, lifecycle.ErrCrashLooping) + assert.Equal(t, spawnsAtLoop, countSpawns(t, spawnLog), "the consuming Start must not itself spawn either") + + require.NoError(t, tool.Start(t.Context())) + assert.Equal(t, spawnsAtLoop+1, countSpawns(t, spawnLog)) + assert.NoError(t, tool.handler.supervisor.PendingCrashLoopError()) +} diff --git a/pkg/tools/lifecycle/errors.go b/pkg/tools/lifecycle/errors.go index 045ba8b7c0..d4de7e72a3 100644 --- a/pkg/tools/lifecycle/errors.go +++ b/pkg/tools/lifecycle/errors.go @@ -26,6 +26,17 @@ var ( // Restartable per policy. ErrServerCrashed = errors.New("server crashed") + // ErrCrashLooping means the supervisor observed Policy.CrashLoop's + // threshold of ErrServerCrashed disconnects within its window and gave + // up restarting (state -> Failed) instead of relaunching again right + // away. Wraps the triggering ErrServerCrashed, so errors.Is(err, + // ErrServerCrashed) still matches. Produced only by Supervisor itself + // (watch's detector, reported once by the next Start) — never by + // Classify — so a single, isolated crash (handled by the ordinary + // restart policy, never escalated) can't be mistaken for a sustained + // loop. + ErrCrashLooping = errors.New("server crash-looping") + // ErrInitTimeout means the initialize handshake did not complete // within the configured deadline. ErrInitTimeout = errors.New("initialize timed out") diff --git a/pkg/tools/lifecycle/supervisor.go b/pkg/tools/lifecycle/supervisor.go index 8a470cf56b..8b1a874112 100644 --- a/pkg/tools/lifecycle/supervisor.go +++ b/pkg/tools/lifecycle/supervisor.go @@ -68,6 +68,35 @@ const ( RestartAlways ) +// CrashLoop configures the supervisor's crash-loop detector: Threshold +// disconnects classified as ErrServerCrashed within Window stop the +// supervisor (state -> Failed, wrapped in ErrCrashLooping) instead of +// restarting again right away, so the caller's own backoff gate (e.g. +// StartableToolSet) paces the next attempt. Zero values default to +// Threshold=3, Window=1 minute. +// +// A single crash, or crashes spread out wider than Window apart, are +// ordinary restarts handled entirely by Restart/Backoff above and never +// trip this detector. +type CrashLoop struct { + Threshold int + Window time.Duration +} + +func (c CrashLoop) threshold() int { + if c.Threshold <= 0 { + return 3 + } + return c.Threshold +} + +func (c CrashLoop) window() time.Duration { + if c.Window <= 0 { + return time.Minute + } + return c.Window +} + // Backoff parameters for restart attempts. Zero values default to // 1s..32s exponential (matching historical MCP behaviour). type Backoff struct { @@ -120,6 +149,10 @@ type Policy struct { // individual tool calls; the Supervisor itself does not use it. CallTimeout time.Duration + // CrashLoop tunes the crash-loop detector (see CrashLoop's doc). Zero + // value uses the CrashLoop defaults. + CrashLoop CrashLoop + // OnDisconnect is called when the session ends, with Wait()'s result. // Useful for cache invalidation. OnDisconnect func(err error) @@ -188,6 +221,18 @@ type Supervisor struct { // shared MCP session. inflightConnect *pendingConnect + // crashTimes holds the timestamps of recent ErrServerCrashed disconnects + // (watch), pruned to policy.CrashLoop.window() on every record. Cleared + // whenever crashLoopErr is consumed or set, and on Stop. + crashTimes []time.Time + // crashLoopErr is a one-shot report: set by watch when the crash-loop + // threshold is reached, and returned by the very next Start instead of + // reconnecting, then cleared so the Start after that attempts a genuine + // reconnect. The one-shot handshake exists because only the caller (the + // StartableToolSet backoff gate) knows when enough time has passed to + // retry for real. + crashLoopErr error + // randFloat is the jitter source; tests may override. randFloat func() float64 } @@ -213,6 +258,20 @@ func (s *Supervisor) State() StateInfo { return s.tracker.Snapshot() } // requests (Ready or Degraded). func (s *Supervisor) IsReady() bool { return s.tracker.State().IsUsable() } +// PendingCrashLoopError returns the crash-loop report awaiting the next +// Start, without consuming it — a peek, unlike Start's one-shot read. +// Repeated calls keep returning the same error until the Start that +// actually consumes it runs. Callers that reach the supervisor outside +// their own backoff gate (e.g. a toolset's lazy per-request start) should +// check this before calling Start, so they fail fast on a known crash +// loop instead of being the one to consume — and reconnect on behalf of +// — a one-shot report the gate hasn't paced yet. +func (s *Supervisor) PendingCrashLoopError() error { + s.mu.Lock() + defer s.mu.Unlock() + return s.crashLoopErr +} + // MarkReadyForTesting forces the supervisor into StateReady without going // through Connect. Test-only backdoor; production code must not call this. func (s *Supervisor) MarkReadyForTesting() { s.tracker.Set(StateReady) } @@ -246,6 +305,15 @@ func (s *Supervisor) Start(ctx context.Context) error { s.mu.Unlock() return ErrNotStarted } + // A crash loop detected by watch reports itself here exactly once + // (see crashLoopErr's doc): the very next Start after this one attempts + // a genuine reconnect rather than short-circuiting again. + if err := s.crashLoopErr; err != nil { + s.crashLoopErr = nil + s.crashTimes = nil + s.mu.Unlock() + return err + } s.mu.Unlock() s.tracker.Set(StateStarting) @@ -416,6 +484,11 @@ func (s *Supervisor) Stop(ctx context.Context) error { watchDone := s.watchDone pending := s.inflightConnect s.inflightConnect = nil + // A deliberate Stop discards any pending crash-loop verdict and its + // history: a future Start begins a fresh window rather than resuming + // one that spans an intentional stop/start cycle. + s.crashLoopErr = nil + s.crashTimes = nil s.mu.Unlock() s.tracker.Set(StateStopped) @@ -541,6 +614,10 @@ func (s *Supervisor) watch(ctx context.Context) { forced := s.forceRestart s.forceRestart = false s.session = nil + // Only an actual crash (not a forced/deliberate close, not a clean + // exit) ever counts toward the loop: those are handled entirely by + // the ordinary restart policy below. + crashLooping := !forced && errors.Is(waitErr, ErrServerCrashed) && s.recordCrashLocked(time.Now()) s.mu.Unlock() s.tracker.Fail(StateRestarting, waitErr) @@ -550,6 +627,27 @@ func (s *Supervisor) watch(ctx context.Context) { cb(waitErr) } + if crashLooping { + err := wrap(ErrCrashLooping, waitErr) + s.mu.Lock() + if s.stopping { + // Stop won the race (landed between the unlock above and here) + // and already set StateStopped: let it own the terminal state + // rather than clobbering it back to Failed. + s.mu.Unlock() + return + } + s.crashLoopErr = err + s.mu.Unlock() + s.tracker.Fail(StateFailed, err) + log.Error("supervisor: crash-looping; giving up", "name", s.name, "threshold", s.policy.CrashLoop.threshold(), "window", s.policy.CrashLoop.window()) + if cb := s.policy.OnFailed; cb != nil { + cb(err) + } + s.signalDone() + return + } + if !s.shouldRestart(waitErr, forced) { s.tracker.Fail(StateFailed, waitErr) if cb := s.policy.OnFailed; cb != nil { @@ -569,6 +667,21 @@ func (s *Supervisor) watch(ctx context.Context) { } } +// recordCrashLocked appends a crash observed at now to the crash-loop +// window, pruning entries older than policy.CrashLoop.window(), and +// reports whether the loop threshold has been reached. s.mu must be held. +func (s *Supervisor) recordCrashLocked(now time.Time) bool { + cutoff := now.Add(-s.policy.CrashLoop.window()) + kept := s.crashTimes[:0] + for _, t := range s.crashTimes { + if t.After(cutoff) { + kept = append(kept, t) + } + } + s.crashTimes = append(kept, now) + return len(s.crashTimes) >= s.policy.CrashLoop.threshold() +} + // shouldRestart applies the supervisor's restart policy to decide whether // the watcher should reconnect after a Wait result. A forced reconnect // (RestartAndWait) bypasses the policy. diff --git a/pkg/tools/lifecycle/supervisor_test.go b/pkg/tools/lifecycle/supervisor_test.go index 4bcf5851ce..f296b4d61e 100644 --- a/pkg/tools/lifecycle/supervisor_test.go +++ b/pkg/tools/lifecycle/supervisor_test.go @@ -3,6 +3,7 @@ package lifecycle_test import ( "context" "errors" + "fmt" "sync" "sync/atomic" "testing" @@ -828,3 +829,408 @@ func TestSupervisor_StopConcurrent(t *testing.T) { assert.Check(t, is.Equal(s.State().State, lifecycle.StateStopped)) assert.Check(t, sess.waitDone.Load(), "a Stop returned before watcher's Wait() completed") } + +// crashErr wraps err (typically a plain "boom"-style message) in +// lifecycle.ErrServerCrashed, matching the shape lspSession.Wait produces +// for a real crashed process. +func crashErr(msg string) error { + return fmt.Errorf("%w: %s", lifecycle.ErrServerCrashed, msg) +} + +// TestSupervisor_CrashLoopStopsRestartingAndReportsOnStart verifies the +// full crash-loop lifecycle: the first two crashes (within the window) are +// ordinary restarts handled entirely by the normal restart policy: no +// pacing, state back to Ready each time. Only the third crash — reaching +// CrashLoop.Threshold — stops the supervisor from reconnecting again and +// reports ErrCrashLooping. That report is a one-shot: the Start that reads +// it does not itself reconnect, but the Start after that does. +func TestSupervisor_CrashLoopStopsRestartingAndReportsOnStart(t *testing.T) { + t.Parallel() + + sess1, sess2, sess3, sess4 := newFakeSession(), newFakeSession(), newFakeSession(), newFakeSession() + c := newScriptedConnector( + scriptStep{session: sess1}, + scriptStep{session: sess2}, + scriptStep{session: sess3}, + scriptStep{session: sess4}, + ) + + restarted := make(chan struct{}, 4) + failed := make(chan error, 1) + s := lifecycle.New("test", c, lifecycle.Policy{ + Backoff: fastBackoff, + CrashLoop: lifecycle.CrashLoop{Threshold: 3, Window: time.Minute}, + OnRestart: func(context.Context) { + select { + case restarted <- struct{}{}: + default: + } + }, + OnFailed: func(err error) { + select { + case failed <- err: + default: + } + }, + }) + + assert.NilError(t, s.Start(t.Context())) + + // Crash 1: an ordinary restart, not a loop yet. + sess1.fail(crashErr("boom")) + select { + case <-restarted: + case <-time.After(2 * time.Second): + t.Fatal("supervisor did not restart after the first crash") + } + assert.Check(t, is.Equal(s.State().State, lifecycle.StateReady), "a single crash must not trip the loop detector") + + // Crash 2: still just an ordinary restart. + sess2.fail(crashErr("boom again")) + select { + case <-restarted: + case <-time.After(2 * time.Second): + t.Fatal("supervisor did not restart after the second crash") + } + assert.Check(t, is.Equal(s.State().State, lifecycle.StateReady), "two crashes must not trip the loop detector") + + // Crash 3 reaches the threshold: give up, no further reconnect attempt. + sess3.fail(crashErr("boom a third time")) + + var loopErr error + select { + case loopErr = <-failed: + case <-time.After(2 * time.Second): + t.Fatal("supervisor did not report a crash loop") + } + assert.Check(t, errors.Is(loopErr, lifecycle.ErrCrashLooping)) + assert.Check(t, errors.Is(loopErr, lifecycle.ErrServerCrashed), "ErrCrashLooping must still match ErrServerCrashed") + assert.Check(t, is.Equal(s.State().State, lifecycle.StateFailed)) + assert.Check(t, is.Equal(c.Calls(), 3), "no reconnect attempt once the loop is detected") + + // The next Start reports the loop once, without connecting again. + err := s.Start(t.Context()) + assert.Check(t, errors.Is(err, lifecycle.ErrCrashLooping)) + assert.Check(t, is.Equal(c.Calls(), 3), "the one-shot report must not itself connect") + + // The Start after that attempts a genuine reconnect. + assert.NilError(t, s.Start(t.Context())) + assert.Check(t, is.Equal(c.Calls(), 4)) + assert.Check(t, is.Equal(s.State().State, lifecycle.StateReady)) + + assert.NilError(t, s.Stop(t.Context())) +} + +// TestSupervisor_CrashLoopIgnoresCleanDisconnects verifies that a clean +// exit (Wait returning nil) never counts toward the crash-loop threshold, +// even when there are more of them than the threshold and RestartAlways +// keeps reconnecting after every one. +func TestSupervisor_CrashLoopIgnoresCleanDisconnects(t *testing.T) { + t.Parallel() + + sessions := make([]*fakeSession, 5) + steps := make([]scriptStep, 5) + for i := range sessions { + sessions[i] = newFakeSession() + steps[i] = scriptStep{session: sessions[i]} + } + c := newScriptedConnector(steps...) + + restarted := make(chan struct{}, 10) + s := lifecycle.New("test", c, lifecycle.Policy{ + Restart: lifecycle.RestartAlways, + Backoff: fastBackoff, + CrashLoop: lifecycle.CrashLoop{Threshold: 3, Window: time.Minute}, + OnRestart: func(context.Context) { + select { + case restarted <- struct{}{}: + default: + } + }, + }) + + assert.NilError(t, s.Start(t.Context())) + for i := range 4 { + _ = sessions[i].Close(t.Context()) // clean disconnect, not a crash + select { + case <-restarted: + case <-time.After(2 * time.Second): + t.Fatalf("supervisor did not restart after clean close %d", i+1) + } + } + + assert.Check(t, is.Equal(s.State().State, lifecycle.StateReady), + "clean disconnects (more of them than the crash-loop threshold) must never trip the loop detector") + assert.Check(t, is.Equal(c.Calls(), 5)) + + assert.NilError(t, s.Stop(t.Context())) +} + +// TestSupervisor_CrashLoopIgnoresForcedRestart verifies that a deliberate +// RestartAndWait (e.g. /toolset-restart) never counts toward the +// crash-loop threshold, even repeated more often than the threshold. +func TestSupervisor_CrashLoopIgnoresForcedRestart(t *testing.T) { + t.Parallel() + + sessions := make([]*fakeSession, 5) + steps := make([]scriptStep, 5) + for i := range sessions { + sessions[i] = newFakeSession() + steps[i] = scriptStep{session: sessions[i]} + } + c := newScriptedConnector(steps...) + + s := lifecycle.New("test", c, lifecycle.Policy{ + Backoff: fastBackoff, + CrashLoop: lifecycle.CrashLoop{Threshold: 3, Window: time.Minute}, + }) + + assert.NilError(t, s.Start(t.Context())) + for i := range 4 { + sessions[i].waitParked(t) + assert.NilError(t, s.RestartAndWait(t.Context(), 2*time.Second)) + } + + assert.Check(t, is.Equal(s.State().State, lifecycle.StateReady), + "forced restarts (more of them than the crash-loop threshold) must never be counted as crashes") + assert.Check(t, is.Equal(c.Calls(), 5)) + + assert.NilError(t, s.Stop(t.Context())) +} + +// TestSupervisor_CrashLoopStopIsClean verifies that Stop terminates cleanly +// from a crash-loop Failed state — discarding the pending crash-loop +// report and history as part of teardown — without hanging or leaving the +// watcher goroutine behind. +func TestSupervisor_CrashLoopStopIsClean(t *testing.T) { + t.Parallel() + + sess1, sess2, sess3 := newFakeSession(), newFakeSession(), newFakeSession() + c := newScriptedConnector( + scriptStep{session: sess1}, + scriptStep{session: sess2}, + scriptStep{session: sess3}, + ) + + failed := make(chan error, 1) + s := lifecycle.New("test", c, lifecycle.Policy{ + Backoff: fastBackoff, + CrashLoop: lifecycle.CrashLoop{Threshold: 3, Window: time.Minute}, + OnFailed: func(err error) { + select { + case failed <- err: + default: + } + }, + }) + + assert.NilError(t, s.Start(t.Context())) + sess1.fail(crashErr("boom")) + sess2.fail(crashErr("boom again")) + sess3.fail(crashErr("boom a third time")) + + select { + case <-failed: + case <-time.After(2 * time.Second): + t.Fatal("supervisor did not report a crash loop") + } + assert.Check(t, is.Equal(s.State().State, lifecycle.StateFailed)) + + assert.NilError(t, s.Stop(t.Context())) + assert.Check(t, is.Equal(s.State().State, lifecycle.StateStopped)) + // A permanently-stopped supervisor never restarts, crash loop or not. + assert.Check(t, errors.Is(s.Start(t.Context()), lifecycle.ErrNotStarted)) +} + +// TestSupervisor_PendingCrashLoopErrorIsNonConsuming verifies that +// PendingCrashLoopError is a pure peek: repeated calls keep returning the +// same report without ever connecting, and without disturbing Start's own +// one-shot consume-then-reconnect contract. +func TestSupervisor_PendingCrashLoopErrorIsNonConsuming(t *testing.T) { + t.Parallel() + + sess1, sess2, sess3, sess4 := newFakeSession(), newFakeSession(), newFakeSession(), newFakeSession() + c := newScriptedConnector( + scriptStep{session: sess1}, + scriptStep{session: sess2}, + scriptStep{session: sess3}, + scriptStep{session: sess4}, + ) + + failed := make(chan error, 1) + s := lifecycle.New("test", c, lifecycle.Policy{ + Backoff: fastBackoff, + CrashLoop: lifecycle.CrashLoop{Threshold: 3, Window: time.Minute}, + OnFailed: func(err error) { + select { + case failed <- err: + default: + } + }, + }) + + assert.NilError(t, s.Start(t.Context())) + sess1.fail(crashErr("boom")) + sess2.fail(crashErr("boom again")) + sess3.fail(crashErr("boom a third time")) + + select { + case <-failed: + case <-time.After(2 * time.Second): + t.Fatal("supervisor did not report a crash loop") + } + + // Peeking repeatedly must not consume the report or trigger a connect. + for range 3 { + assert.Check(t, errors.Is(s.PendingCrashLoopError(), lifecycle.ErrCrashLooping)) + } + assert.Check(t, is.Equal(c.Calls(), 3), "peeking must never itself connect") + + // Start still reports it once, then reconnects for real on the next call. + err := s.Start(t.Context()) + assert.Check(t, errors.Is(err, lifecycle.ErrCrashLooping)) + assert.Check(t, s.PendingCrashLoopError() == nil, "Start must have consumed the report") + assert.Check(t, is.Equal(c.Calls(), 3)) + + assert.NilError(t, s.Start(t.Context())) + assert.Check(t, is.Equal(c.Calls(), 4)) + + assert.NilError(t, s.Stop(t.Context())) +} + +// TestSupervisor_CrashLoopWindowPrunesOldCrashes verifies that crashes +// spread out wider than CrashLoop.Window apart never accumulate toward +// the threshold: each one ages out of the window before the next occurs, +// so the supervisor keeps restarting normally instead of giving up. +func TestSupervisor_CrashLoopWindowPrunesOldCrashes(t *testing.T) { + t.Parallel() + + sessions := make([]*fakeSession, 5) + steps := make([]scriptStep, 5) + for i := range sessions { + sessions[i] = newFakeSession() + steps[i] = scriptStep{session: sessions[i]} + } + c := newScriptedConnector(steps...) + + restarted := make(chan struct{}, 10) + s := lifecycle.New("test", c, lifecycle.Policy{ + Backoff: fastBackoff, + CrashLoop: lifecycle.CrashLoop{Threshold: 3, Window: 20 * time.Millisecond}, + OnRestart: func(context.Context) { + select { + case restarted <- struct{}{}: + default: + } + }, + }) + + assert.NilError(t, s.Start(t.Context())) + for i := range 4 { + sessions[i].fail(crashErr("boom")) + select { + case <-restarted: + case <-time.After(2 * time.Second): + t.Fatalf("supervisor did not restart after crash %d", i+1) + } + // Real time must actually pass beyond Window so the next crash finds + // this one already pruned; there is nothing to synchronize on here + // other than wall-clock time itself. + time.Sleep(30 * time.Millisecond) //nolint:forbidigo // proving the crash-loop window ages entries out; no event to synchronize on + } + + assert.Check(t, is.Equal(s.State().State, lifecycle.StateReady), + "crashes spread wider than the window must never accumulate toward the threshold") + assert.Check(t, is.Equal(c.Calls(), 5)) + + assert.NilError(t, s.Stop(t.Context())) +} + +// TestSupervisor_CrashLoopStopWinningRaceReportsStopped verifies that a +// Stop landing between the crash-loop branch computing crashLooping +// (unlocked) and it re-locking to record crashLoopErr leaves the +// supervisor in StateStopped, not StateFailed: Stop, once it wins, +// owns the terminal state rather than having it clobbered back to Failed. +func TestSupervisor_CrashLoopStopWinningRaceReportsStopped(t *testing.T) { + t.Parallel() + + sess1, sess2, sess3 := newFakeSession(), newFakeSession(), newFakeSession() + c := newScriptedConnector( + scriptStep{session: sess1}, + scriptStep{session: sess2}, + scriptStep{session: sess3}, + ) + + disconnected := make(chan struct{}) + release := make(chan struct{}) + restarted := make(chan struct{}, 4) + var thirdCrash atomic.Bool + s := lifecycle.New("test", c, lifecycle.Policy{ + Backoff: fastBackoff, + CrashLoop: lifecycle.CrashLoop{Threshold: 3, Window: time.Minute}, + OnRestart: func(context.Context) { + select { + case restarted <- struct{}{}: + default: + } + }, + OnDisconnect: func(error) { + // OnDisconnect runs after crashLooping is already computed + // (true, for the third crash) but before crashLoopErr is + // recorded: exactly the race window under test. Park the + // watcher here so the test can land a concurrent Stop inside it. + if thirdCrash.Load() { + close(disconnected) + <-release + } + }, + }) + + assert.NilError(t, s.Start(t.Context())) + + // The first two crashes must actually land and restart (onto sess2, + // then sess3) before the third is fired, or thirdCrash could be set + // before the watcher has even processed the first one. + sess1.fail(crashErr("boom")) + select { + case <-restarted: + case <-time.After(2 * time.Second): + t.Fatal("supervisor did not restart after the first crash") + } + + sess2.fail(crashErr("boom again")) + select { + case <-restarted: + case <-time.After(2 * time.Second): + t.Fatal("supervisor did not restart after the second crash") + } + + // sess3 is now live; its crash is the third, tripping the loop. + thirdCrash.Store(true) + sess3.fail(crashErr("boom a third time")) + <-disconnected + + stopDone := make(chan error, 1) + go func() { stopDone <- s.Stop(t.Context()) }() + + // Wait for Stop to actually win the race and record StateStopped + // before releasing the parked crash-loop branch: Stop sets state + // synchronously, before it ever blocks waiting for the watcher to exit. + poll.WaitOn(t, func(poll.LogT) poll.Result { + if s.State().State == lifecycle.StateStopped { + return poll.Success() + } + return poll.Continue("supervisor state=%s", s.State().State) + }, poll.WithTimeout(2*time.Second), poll.WithDelay(5*time.Millisecond)) + close(release) + + select { + case err := <-stopDone: + assert.NilError(t, err) + case <-time.After(2 * time.Second): + t.Fatal("Stop did not return") + } + + assert.Check(t, is.Equal(s.State().State, lifecycle.StateStopped), + "Stop winning the race must leave the supervisor Stopped, not clobbered back to Failed") +} diff --git a/pkg/tools/startable_backoff.go b/pkg/tools/startable_backoff.go index d09b010278..34dc508f5a 100644 --- a/pkg/tools/startable_backoff.go +++ b/pkg/tools/startable_backoff.go @@ -6,6 +6,7 @@ import ( "time" "github.com/docker/docker-agent/pkg/modelerrors" + "github.com/docker/docker-agent/pkg/tools/lifecycle" ) const ( @@ -13,18 +14,33 @@ const ( startBackoffMax = 5 * time.Minute ) -// startBackoffRetryable reports whether err carries a retryable HTTP status -// that warrants pacing the next start attempt: 429, 408, 500, 502, 503, 504, -// or 529 (see modelerrors.isRetryableStatusCode). This is a fixed -// enumeration, not a full 5xx range — codes such as 501, 505, or the -// Cloudflare 520-527 family do NOT arm the gate. Only a *modelerrors.StatusError -// in the error chain arms the gate; plain network errors and the regex -// fallback in RetryableHTTPStatus are intentionally excluded so port numbers, -// PIDs, and chunk counters in plain error text cannot arm the gate. +// startBackoffRetryable reports whether err warrants pacing the next start +// attempt. Two independent categories arm the gate: // -// A StatusError wins even when context.DeadlineExceeded is also in the chain. +// - lifecycle.ErrCrashLooping: the supervisor itself already judged this a +// sustained crash loop (see lifecycle.Supervisor's CrashLoop policy) and +// is reporting it through Start instead of reconnecting immediately. +// Checked directly via errors.Is — no StatusError involved, because the +// supervisor already did the one-off-vs-loop judgment; the gate only +// needs to pace the retry. A bare lifecycle.ErrServerCrashed that hasn't +// (yet) escalated to ErrCrashLooping does NOT arm: see "deliberately +// excluded" below. +// - a retryable HTTP status carried by a *modelerrors.StatusError: 429, +// 408, 500, 502, 503, 504, or 529 (see modelerrors.isRetryableStatusCode). +// This is a fixed enumeration, not a full 5xx range — codes such as 501, +// 505, or the Cloudflare 520-527 family do NOT arm the gate. Only a +// *modelerrors.StatusError in the error chain arms this branch; plain +// network errors and the regex fallback in RetryableHTTPStatus are +// intentionally excluded so port numbers, PIDs, and chunk counters in +// plain error text cannot arm the gate. A StatusError wins even when +// context.DeadlineExceeded is also in the chain. // // Deliberately excluded from arming (these must never pace): +// - A bare lifecycle.ErrServerCrashed not (yet) escalated to +// ErrCrashLooping: a single crash is the supervisor's own restart +// policy's job (fast, unpaced reconnect), not this gate's — pacing it +// here too would double up on the supervisor's own backoff and delay a +// legitimate one-off recovery. // - lifecycle.ErrServerUnavailable: missing binary / process-not-found — fast-retry. // - lifecycle.ErrTransport: connection refused / no such host — fast-retry. // - lifecycle.ErrAuthRequired / ErrCapabilityMissing: permanent — fail promptly. @@ -32,12 +48,10 @@ const ( // supervisor's own reconnect policy without per-turn pacing. // - Plain error strings: excluded to avoid false positives on numeric // patterns in port numbers or counters. -// -// Note: lifecycle.ErrServerCrashed (a server that started then crashed) is -// NOT currently surfaced by supervisor.Start(); it flows only through the -// supervisor's internal watcher goroutine. LSP crash-loop pacing is therefore -// deferred until that sentinel is propagated through the start path. func startBackoffRetryable(err error) bool { + if errors.Is(err, lifecycle.ErrCrashLooping) { + return true + } var se *modelerrors.StatusError if !errors.As(err, &se) { return false diff --git a/pkg/tools/startable_backoff_test.go b/pkg/tools/startable_backoff_test.go index 530c48ac64..2e279ee7e6 100644 --- a/pkg/tools/startable_backoff_test.go +++ b/pkg/tools/startable_backoff_test.go @@ -815,3 +815,49 @@ func TestStartBackoffRetryable_4xxStatusDoesNotArm(t *testing.T) { assert.Check(t, is.Equal(inner.starts.Load(), int32(3)), "400 client errors must not arm the backoff gate") } + +// TestStartBackoffRetryable_ErrServerCrashed verifies that a bare +// lifecycle.ErrServerCrashed (a single crash, not escalated to a loop) does +// NOT arm the gate: it is the supervisor's own restart policy's job, not +// this gate's. +func TestStartBackoffRetryable_ErrServerCrashed(t *testing.T) { + inner := &startErrToolSet{} + inner.setErr(fmt.Errorf("%w: exit status 1", lifecycle.ErrServerCrashed)) + s := newThrottledStartable(inner) + + for range 3 { + _, err := s.TryStart(t.Context()) + assert.Check(t, err != nil) + } + assert.Check(t, is.Equal(inner.starts.Load(), int32(3)), + "a one-off server crash must not pace retries") +} + +// TestStartBackoffRetryable_ErrCrashLooping verifies that +// lifecycle.ErrCrashLooping — the supervisor's own verdict that a crash +// loop is underway — arms the gate exactly like a retryable HTTP status, +// via TryStart end-to-end. +func TestStartBackoffRetryable_ErrCrashLooping(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + inner := &startErrToolSet{} + inner.setErr(fmt.Errorf("%w: %w", lifecycle.ErrCrashLooping, lifecycle.ErrServerCrashed)) + s := newThrottledStartable(inner) + + // Attempt 1: underlying Start runs and fails. + _, err := s.TryStart(t.Context()) + assert.Check(t, err != nil) + assert.Check(t, is.Equal(inner.starts.Load(), int32(1))) + + // Immediate retry: gate must block it (still within the window). + _, err = s.TryStart(t.Context()) + assert.Check(t, err != nil) + assert.Check(t, is.Equal(inner.starts.Load(), int32(1)), + "a crash loop must arm the gate like a retryable HTTP status") + + // After the window expires the next TryStart runs the underlying attempt. + time.Sleep(tools.ExportedStartBackoffBase + time.Millisecond) //nolint:forbidigo // inside synctest bubble + _, err = s.TryStart(t.Context()) + assert.Check(t, err != nil) + assert.Check(t, is.Equal(inner.starts.Load(), int32(2))) + }) +}