Back off checkpoint capture after repeated failures

apps/server/src/checkpointing/CaptureBackoff.test.ts

@@ -0,0 +1,191 @@
+import { describe, expect, it } from "vite-plus/test";
+
+import { cooldownForFailureCount, makeCaptureBackoff } from "./CaptureBackoff.ts";
+
+const MINUTE = 60_000;
+const CWD = "/repo/workspace";
+
+describe("cooldownForFailureCount", () => {
+  it("tolerates transient failures before opening a cooldown", () => {
+    expect(cooldownForFailureCount(1)).toBe(0);
+    expect(cooldownForFailureCount(2)).toBe(0);
+  });
+
+  it("backs off further the longer capture keeps failing", () => {
+    expect(cooldownForFailureCount(3)).toBe(5 * MINUTE);
+    expect(cooldownForFailureCount(4)).toBe(10 * MINUTE);
+    expect(cooldownForFailureCount(5)).toBe(20 * MINUTE);
+  });
+
+  it("caps the cooldown so a workspace is retried eventually", () => {
+    expect(cooldownForFailureCount(20)).toBe(60 * MINUTE);
+    expect(cooldownForFailureCount(500)).toBe(60 * MINUTE);
+  });
+});
+
+describe("makeCaptureBackoff", () => {
+  it("does not skip before the failure threshold is reached", () => {
+    const backoff = makeCaptureBackoff<string>();
+
+    backoff.recordFailure(CWD, 0, "timeout");
+    backoff.recordFailure(CWD, 1_000, "timeout");
+
+    expect(backoff.beginAttempt(CWD, 2_000).skip).toBe(false);
+  });
+
+  it("skips and replays the recorded failure once capture keeps failing", () => {
+    const backoff = makeCaptureBackoff<string>();
+
+    for (const at of [0, 1_000, 2_000]) {
+      backoff.recordFailure(CWD, at, "git add timed out");
+    }
+
+    const decision = backoff.beginAttempt(CWD, 3_000);
+    expect(decision.skip).toBe(true);
+    expect(decision.lastError).toBe("git add timed out");
+    expect(decision.remainingMs).toBeGreaterThan(0);
+  });
+
+  it("retries again once the cooldown elapses", () => {
+    const backoff = makeCaptureBackoff<string>();
+
+    for (const at of [0, 0, 0]) {
+      backoff.recordFailure(CWD, at, "timeout");
+    }
+
+    expect(backoff.beginAttempt(CWD, 5 * MINUTE - 1).skip).toBe(true);
+    expect(backoff.beginAttempt(CWD, 5 * MINUTE).skip).toBe(false);
+  });
+
+  it("clears the record after a capture succeeds", () => {
+    const backoff = makeCaptureBackoff<string>();
+
+    for (const at of [0, 0, 0]) {
+      backoff.recordFailure(CWD, at, "timeout");
+    }
+    expect(backoff.beginAttempt(CWD, 1_000).skip).toBe(true);
+
+    backoff.recordSuccess(CWD);
+
+    expect(backoff.beginAttempt(CWD, 1_000).skip).toBe(false);
+    expect(backoff.trackedWorkspaceCount).toBe(0);
+  });
+
+  it("tracks each workspace independently", () => {
+    const backoff = makeCaptureBackoff<string>();
+    const healthy = "/repo/other";
+
+    for (const at of [0, 0, 0]) {
+      backoff.recordFailure(CWD, at, "timeout");
+    }
+
+    expect(backoff.beginAttempt(CWD, 1_000).skip).toBe(true);
+    expect(backoff.beginAttempt(healthy, 1_000).skip).toBe(false);
+  });
+
+  it("bounds how many workspaces it retains", () => {
+    const backoff = makeCaptureBackoff<string>();
+
+    for (let index = 0; index < 400; index += 1) {
+      backoff.recordFailure(`/repo/workspace-${index}`, index, "timeout");
+    }
+
+    expect(backoff.trackedWorkspaceCount).toBe(256);
+    // The oldest entries are the ones dropped: an evicted workspace starts
+    // counting again from one, a retained one continues.
+    expect(backoff.recordFailure("/repo/workspace-0", 1_000, "timeout").consecutiveFailures).toBe(
+      1,
+    );
+    expect(backoff.recordFailure("/repo/workspace-399", 1_000, "timeout").consecutiveFailures).toBe(
+      2,
+    );
+  });
+
+  it("keeps a repeatedly failing workspace alive through eviction pressure", () => {
+    const backoff = makeCaptureBackoff<string>();
+
+    // A workspace in active use fails on every turn while many one-off
+    // workspaces churn past it.
+    for (let index = 0; index < 400; index += 1) {
+      backoff.recordFailure(CWD, index, "timeout");
+      backoff.recordFailure(`/repo/other-${index}`, index, "timeout");
+    }
+
+    expect(backoff.beginAttempt(CWD, 400).skip).toBe(true);
+  });
+
+  it("keeps a workspace that is only being skipped safe from eviction", () => {
+    const backoff = makeCaptureBackoff<string>();
+
+    for (const at of [0, 0, 0]) {
+      backoff.recordFailure(CWD, at, "timeout");
+    }
+
+    // A skipped workspace never calls recordFailure again, so only the skip
+    // itself can keep it ahead of churn from unrelated workspaces.
+    for (let index = 0; index < 400; index += 1) {
+      expect(backoff.beginAttempt(CWD, 1_000).skip).toBe(true);
+      backoff.recordFailure(`/repo/other-${index}`, index, "timeout");
+    }
+
+    expect(backoff.beginAttempt(CWD, 1_000).skip).toBe(true);
+  });
+
+  it("does not reserve for a workspace that has not reached the threshold", () => {
+    const backoff = makeCaptureBackoff<string>();
+
+    backoff.recordFailure(CWD, 0, "timeout");
+
+    // One transient failure must not suppress a capture running alongside it.
+    expect(backoff.beginAttempt(CWD, 1_000).skip).toBe(false);
+    expect(backoff.beginAttempt(CWD, 1_000).skip).toBe(false);
+    expect(backoff.beginAttempt(CWD, 2_000).skip).toBe(false);
+  });
+
+  it("releases only one caller when the cooldown expires", () => {
+    const backoff = makeCaptureBackoff<string>();
+
+    for (const at of [0, 0, 0]) {
+      backoff.recordFailure(CWD, at, "timeout");
+    }
+
+    // Threads sharing a workspace can complete turns together, and only the
+    // first past the cooldown should pay for the capture.
+    const released = [
+      backoff.beginAttempt(CWD, 5 * MINUTE),
+      backoff.beginAttempt(CWD, 5 * MINUTE),
+      backoff.beginAttempt(CWD, 5 * MINUTE),
+    ].filter((decision) => !decision.skip);
+
+    expect(released).toHaveLength(1);
+  });
+
+  it("lets the reservation lapse when an attempt never reports back", () => {
+    const backoff = makeCaptureBackoff<string>();
+
+    for (const at of [0, 0, 0]) {
+      backoff.recordFailure(CWD, at, "timeout");
+    }
+    backoff.beginAttempt(CWD, 5 * MINUTE);
+
+    // An interrupted capture reports neither success nor failure, so the
+    // reservation must expire on its own rather than wedge the workspace.
+    expect(backoff.beginAttempt(CWD, 5 * MINUTE + 30_000).skip).toBe(true);
+    expect(backoff.beginAttempt(CWD, 6 * MINUTE + 1).skip).toBe(false);
+  });
+
+  it("keeps extending the cooldown while failures continue", () => {
+    const backoff = makeCaptureBackoff<string>();
+
+    for (const at of [0, 0, 0]) {
+      backoff.recordFailure(CWD, at, "timeout");
+    }
+    const firstRemaining = backoff.beginAttempt(CWD, 0).remainingMs;
+
+    // The next attempt after the cooldown fails again, so the wait grows.
+    backoff.recordFailure(CWD, 5 * MINUTE, "timeout");
+    const secondRemaining = backoff.beginAttempt(CWD, 5 * MINUTE).remainingMs;
+
+    expect(secondRemaining).toBeGreaterThan(firstRemaining);
+  });
+});

apps/server/src/checkpointing/CaptureBackoff.ts

@@ -0,0 +1,146 @@
+/**
+ * Consecutive-failure backoff for workspace checkpoint capture.
+ *
+ * Capture runs a full-tree `git add -A` under a temporary index after every
+ * completed turn. On a repository large enough to exceed the VCS process
+ * timeout, that capture can never succeed, so the unguarded retry pegs a CPU
+ * core for as long as any thread is in use and litters `.git/objects/pack`
+ * with `tmp_pack_*` files from each killed process.
+ *
+ * After a few consecutive failures for a workspace, capture is skipped for a
+ * growing cooldown instead of being retried every turn. Any success clears
+ * the record, so a transient failure (a lock held by a concurrent git
+ * command) costs nothing.
+ *
+ * @module CaptureBackoff
+ */
+
+/** Failures tolerated before a workspace enters cooldown. */
+const FAILURE_THRESHOLD = 3;
+const BASE_COOLDOWN_MS = 5 * 60_000;
+const MAX_COOLDOWN_MS = 60 * 60_000;
+
+/**
+ * Only a workspace that later succeeds clears its own record, so a workspace
+ * that fails once and is then abandoned would otherwise be retained for the
+ * lifetime of the server. Bound the tracked set and evict least-recently
+ * touched entries; dropping one only costs a workspace its failure history.
+ */
+const MAX_TRACKED_WORKSPACES = 256;
+
+/**
+ * How long the first caller past an expired cooldown holds the retry to
+ * itself. Threads sharing a workspace complete turns independently, so
+ * without this they would all be released at once and launch the same
+ * expensive capture together. Comfortably longer than the VCS process
+ * timeout that kills a stuck capture, and it lapses on its own, so an
+ * attempt that never reports back cannot wedge the workspace.
+ */
+const ATTEMPT_RESERVATION_MS = 60_000;
+
+export interface CaptureBackoffDecision<E> {
+  readonly skip: boolean;
+  /** Milliseconds left in the cooldown, for logging. Zero when not skipping. */
+  readonly remainingMs: number;
+  /**
+   * The failure that opened the cooldown. Replayed instead of inventing a new
+   * error, so callers keep seeing the real reason capture is unavailable and
+   * the error channel is unchanged.
+   */
+  readonly lastError: E | null;
+}
+
+export function cooldownForFailureCount(consecutiveFailures: number): number {
+  if (consecutiveFailures < FAILURE_THRESHOLD) return 0;
+  const doublings = consecutiveFailures - FAILURE_THRESHOLD;
+  // Clamp the exponent before shifting so a long-lived workspace cannot
+  // overflow into a negative or infinite cooldown.
+  const scale = 2 ** Math.min(doublings, 10);
+  return Math.min(BASE_COOLDOWN_MS * scale, MAX_COOLDOWN_MS);
+}
+
+export interface CaptureFailureOutcome {
+  readonly consecutiveFailures: number;
+  /** Zero while the workspace is still under the failure threshold. */
+  readonly cooldownMs: number;
+}
+
+interface WorkspaceRecord<E> {
+  consecutiveFailures: number;
+  skipUntilMs: number;
+  lastError: E;
+}
+
+/**
+ * Tracks capture health per workspace. Callers ask whether to skip, then
+ * report the outcome of any capture they actually ran.
+ */
+export function makeCaptureBackoff<E>() {
+  const recordByCwd = new Map<string, WorkspaceRecord<E>>();
+
+  /** Move a record to the most-recent position so eviction sees real usage. */
+  const touch = (cwd: string, record: WorkspaceRecord<E>) => {
+    recordByCwd.delete(cwd);
+    recordByCwd.set(cwd, record);
+  };
+
+  return {
+    /**
+     * Decide whether this caller should run a capture. Mutating: a caller
+     * released past an expired cooldown reserves the attempt, and any read
+     * refreshes eviction recency so a workspace being actively skipped is not
+     * evicted by churn from unrelated workspaces.
+     */
+    beginAttempt(cwd: string, nowMs: number): CaptureBackoffDecision<E> {
+      const record = recordByCwd.get(cwd);
+      if (!record) {
+        return { skip: false, remainingMs: 0, lastError: null };
+      }
+
+      touch(cwd, record);
+      // Below the threshold a workspace has no cooldown at all, and its zero
+      // deadline must not read as one that just expired: reserving there
+      // would let a single transient failure suppress a concurrent capture.
+      if (record.consecutiveFailures < FAILURE_THRESHOLD) {
+        return { skip: false, remainingMs: 0, lastError: null };
+      }
+
+      if (nowMs < record.skipUntilMs) {
+        return {
+          skip: true,
+          remainingMs: record.skipUntilMs - nowMs,
+          lastError: record.lastError,
+        };
+      }
+
+      record.skipUntilMs = nowMs + ATTEMPT_RESERVATION_MS;
+      return { skip: false, remainingMs: 0, lastError: null };
+    },
+
+    recordSuccess(cwd: string): void {
+      recordByCwd.delete(cwd);
+    },
+
+    recordFailure(cwd: string, nowMs: number, error: E): CaptureFailureOutcome {
+      const consecutiveFailures = (recordByCwd.get(cwd)?.consecutiveFailures ?? 0) + 1;
+      const cooldownMs = cooldownForFailureCount(consecutiveFailures);
+      touch(cwd, {
+        consecutiveFailures,
+        skipUntilMs: cooldownMs === 0 ? 0 : nowMs + cooldownMs,
+        lastError: error,
+      });
+
+      while (recordByCwd.size > MAX_TRACKED_WORKSPACES) {
+        const oldestCwd = recordByCwd.keys().next().value;
+        if (oldestCwd === undefined) break;
+        recordByCwd.delete(oldestCwd);
+      }
+
+      return { consecutiveFailures, cooldownMs };
+    },
+
+    get trackedWorkspaceCount(): number {
+      return recordByCwd.size;
+    },
+  };
+}