fix: stop treating live mirror jobs as interrupted (#372) (#373)

A job that had just started, with in_progress=1 and no checkpoint yet,
matched findInterruptedJobs immediately. Since #297 the middleware runs
that check on every request, so a live job was "resumed" by recovery
while the original process was still working on the same repositories.
The request-level 15 second timeout then released the in-flight latch
without cancelling recovery, and later requests logged misleading
"already in progress" and "completed with some issues" lines.

- Move the liveness rule into interrupted-job-detection.ts, as both a
  predicate and the SQL condition used by findInterruptedJobs. A job
  with no checkpoint is only interrupted once it is older than the 10
  minute checkpoint window (or has no recorded start at all).
- Stamp an initial checkpoint on in-progress jobs at creation.
- Refresh the checkpoint every 2 minutes from processWithResilience so a
  single long item cannot make a live job look interrupted. The refresh
  only touches rows still in progress.
- Keep the middleware recovery latch held until the recovery promise
  settles, not until the request stops waiting, and clear the timeout
  timer so it no longer leaves a dangling rejection.

Tests cover the predicate, the SQL against an in-memory SQLite database,
and the wiring into helpers, concurrency, and middleware.

Claude-Session: https://claude.ai/code/session_01Tp9pmi65a8k5jLMQFLf4JX

docs/RECOVERY_IMPROVEMENTS.md

@@ -63,9 +63,11 @@ The middleware now serves as a fallback recovery mechanism:
 
 #### Improvements:
 - **Proper Drizzle ORM syntax** for all database queries
-- **Enhanced interrupted job detection** with multiple criteria:
+- **Enhanced interrupted job detection** with multiple criteria (see `src/lib/interrupted-job-detection.ts`):
   - Jobs with no recent checkpoint (10+ minutes)
+  - Jobs that never wrote a checkpoint and started 10+ minutes ago (a job that just started is live, not interrupted)
   - Jobs running too long (2+ hours)
+- **Heartbeat during processing**: in-progress jobs start with a checkpoint and refresh it every 2 minutes, so one long item cannot make a live job look interrupted
 - **Detailed logging** of found interrupted jobs
 - **Better error handling** for database operations
 

src/lib/helpers.ts

@@ -1,6 +1,7 @@
 import type { RepoStatus } from "@/types/Repository";
 import { db, mirrorJobs } from "./db";
-import { eq, and, or, lt, isNull } from "drizzle-orm";
+import { eq, and } from "drizzle-orm";
+import { buildInterruptedJobsCondition } from "./interrupted-job-detection";
 import { v4 as uuidv4 } from "uuid";
 import { publishEvent } from "./events";
 import { triggerJobNotification } from "./notification-service";
@@ -63,7 +64,10 @@ export async function createMirrorJob({
     inProgress: inProgress !== undefined ? inProgress : false,
     startedAt: inProgress ? currentTimestamp : undefined,
     completedAt: undefined,
-    lastCheckpoint: undefined,
+    // An in-progress job starts with a checkpoint at its creation time.
+    // Without one, findInterruptedJobs used to treat the job as
+    // interrupted before its first item finished (issue #372).
+    lastCheckpoint: inProgress ? currentTimestamp : undefined,
   };
 
   try {
@@ -215,6 +219,22 @@ export async function updateMirrorJobProgress({
   }
 }
 
+/**
+ * Refreshes a job's checkpoint without recording progress. Called on a
+ * timer while a job is processing so that one long-running item (a large
+ * repository migration) cannot outlive the interrupted-job cutoff and get
+ * "resumed" by recovery while it is still live (issue #372).
+ *
+ * Only touches rows that are still in progress, so a heartbeat that fires
+ * after the job completed or failed changes nothing.
+ */
+export async function touchMirrorJobCheckpoint(jobId: string): Promise<void> {
+  await db
+    .update(mirrorJobs)
+    .set({ lastCheckpoint: new Date() })
+    .where(and(eq(mirrorJobs.id, jobId), eq(mirrorJobs.inProgress, true)));
+}
+
 /**
  * Finds interrupted jobs that need to be resumed with enhanced criteria.
  *
@@ -232,28 +252,15 @@ export async function findInterruptedJobs(
 ) {
   const { logFound = false } = options;
   try {
-    // Find jobs that are marked as in-progress but haven't been updated recently
-    const cutoffTime = new Date();
-    cutoffTime.setMinutes(cutoffTime.getMinutes() - 10); // Consider jobs inactive after 10 minutes without updates
-
-    // Also check for jobs that have been running for too long (over 2 hours)
-    const staleCutoffTime = new Date();
-    staleCutoffTime.setHours(staleCutoffTime.getHours() - 2);
-
+    // The liveness rule (checkpoint age, stale start, and the grace period
+    // for jobs that have not written a checkpoint yet) lives in
+    // interrupted-job-detection.ts so it can be tested against a real
+    // database. See issue #372 for why a null checkpoint alone is not
+    // enough to call a job interrupted.
     const interruptedJobs = await db
       .select()
       .from(mirrorJobs)
-      .where(
-        and(
-          eq(mirrorJobs.inProgress, true),
-          or(
-            // Jobs with no recent checkpoint
-            or(isNull(mirrorJobs.lastCheckpoint), lt(mirrorJobs.lastCheckpoint, cutoffTime)),
-            // Jobs that started too long ago (likely stale)
-            lt(mirrorJobs.startedAt, staleCutoffTime)
-          )
-        )
-      );
+      .where(buildInterruptedJobsCondition(mirrorJobs));
 
     // Log details about found jobs for debugging — opt-in to avoid
     // spamming the log when called from periodic passive checks.

src/lib/interrupted-job-detection.test.ts

@@ -0,0 +1,254 @@
+/**
+ * Tests for interrupted-job detection (issue #372).
+ *
+ * A job that had just started, with `inProgress=true` and no checkpoint yet,
+ * matched `findInterruptedJobs` immediately, so request-time recovery
+ * "resumed" jobs that were still running. The rule now lives in
+ * interrupted-job-detection.ts, both as a predicate and as the SQL condition
+ * used by findInterruptedJobs. The SQL is exercised against a real in-memory
+ * SQLite database. The wiring into helpers.ts, concurrency.ts, and
+ * middleware.ts is asserted by reading the source, following the convention
+ * in orchestrator-resume-after-startup.test.ts.
+ */
+import { describe, test, expect, beforeAll, afterAll } from "bun:test";
+import { readFileSync } from "node:fs";
+import { join } from "node:path";
+import { Database } from "bun:sqlite";
+import { drizzle } from "drizzle-orm/bun-sqlite";
+import { mirrorJobs } from "@/lib/db/schema";
+import {
+  INTERRUPTED_CHECKPOINT_AGE_MS,
+  STALE_JOB_AGE_MS,
+  JOB_HEARTBEAT_INTERVAL_MS,
+  computeInterruptedJobCutoffs,
+  isJobInterrupted,
+  buildInterruptedJobsCondition,
+  type JobLivenessFields,
+} from "./interrupted-job-detection";
+
+const MINUTE = 60 * 1000;
+const HOUR = 60 * MINUTE;
+const NOW = new Date("2026-09-02T12:00:00Z");
+const ago = (ms: number) => new Date(NOW.getTime() - ms);
+
+describe("computeInterruptedJobCutoffs", () => {
+  test("keeps the 10 minute checkpoint window and 2 hour stale window", () => {
+    expect(INTERRUPTED_CHECKPOINT_AGE_MS).toBe(10 * MINUTE);
+    expect(STALE_JOB_AGE_MS).toBe(2 * HOUR);
+
+    const cutoffs = computeInterruptedJobCutoffs(NOW);
+
+    expect(cutoffs.checkpointCutoff.getTime()).toBe(NOW.getTime() - 10 * MINUTE);
+    expect(cutoffs.staleCutoff.getTime()).toBe(NOW.getTime() - 2 * HOUR);
+  });
+
+  test("heartbeat interval leaves room for a missed beat inside the checkpoint window", () => {
+    expect(JOB_HEARTBEAT_INTERVAL_MS * 2).toBeLessThanOrEqual(INTERRUPTED_CHECKPOINT_AGE_MS);
+  });
+});
+
+describe("isJobInterrupted", () => {
+  test("a job that just started with no checkpoint is live (regression for #372)", () => {
+    expect(
+      isJobInterrupted({ inProgress: true, startedAt: ago(MINUTE), lastCheckpoint: null }, NOW)
+    ).toBe(false);
+  });
+
+  test("a job with no checkpoint is interrupted once it is older than the checkpoint window", () => {
+    expect(
+      isJobInterrupted({ inProgress: true, startedAt: ago(11 * MINUTE), lastCheckpoint: null }, NOW)
+    ).toBe(true);
+  });
+
+  test("a legacy row with neither checkpoint nor start time is interrupted", () => {
+    expect(
+      isJobInterrupted({ inProgress: true, startedAt: null, lastCheckpoint: null }, NOW)
+    ).toBe(true);
+  });
+
+  test("a fresh checkpoint keeps a job live", () => {
+    expect(
+      isJobInterrupted({ inProgress: true, startedAt: ago(HOUR), lastCheckpoint: ago(2 * MINUTE) }, NOW)
+    ).toBe(false);
+  });
+
+  test("a checkpoint older than the window marks the job interrupted", () => {
+    expect(
+      isJobInterrupted({ inProgress: true, startedAt: ago(HOUR), lastCheckpoint: ago(11 * MINUTE) }, NOW)
+    ).toBe(true);
+  });
+
+  test("a job running past the stale window is interrupted even with a fresh checkpoint", () => {
+    expect(
+      isJobInterrupted({ inProgress: true, startedAt: ago(3 * HOUR), lastCheckpoint: ago(MINUTE) }, NOW)
+    ).toBe(true);
+  });
+
+  test("jobs that are not in progress are never interrupted", () => {
+    expect(
+      isJobInterrupted({ inProgress: false, startedAt: ago(3 * HOUR), lastCheckpoint: null }, NOW)
+    ).toBe(false);
+  });
+});
+
+describe("buildInterruptedJobsCondition", () => {
+  const sqlite = new Database(":memory:");
+  const db = drizzle(sqlite);
+
+  beforeAll(() => {
+    // Only the columns the condition and the projection touch. The schema's
+    // other columns are irrelevant here and would need seeding otherwise.
+    sqlite.run(
+      "CREATE TABLE mirror_jobs (id TEXT PRIMARY KEY, in_progress INTEGER NOT NULL DEFAULT 0, started_at INTEGER, last_checkpoint INTEGER)"
+    );
+  });
+
+  afterAll(() => {
+    sqlite.close();
+  });
+
+  type Row = JobLivenessFields & { id: string };
+
+  // Drizzle stores `timestamp` mode integers as unix seconds.
+  const seconds = (value: Date | null | undefined) =>
+    value ? Math.floor(value.getTime() / 1000) : null;
+
+  function seed(rows: Row[]) {
+    sqlite.run("DELETE FROM mirror_jobs");
+    const insert = sqlite.prepare(
+      "INSERT INTO mirror_jobs (id, in_progress, started_at, last_checkpoint) VALUES (?, ?, ?, ?)"
+    );
+    for (const row of rows) {
+      insert.run(row.id, row.inProgress ? 1 : 0, seconds(row.startedAt), seconds(row.lastCheckpoint));
+    }
+  }
+
+  async function findInterruptedIds() {
+    const rows = await db
+      .select({ id: mirrorJobs.id })
+      .from(mirrorJobs)
+      .where(buildInterruptedJobsCondition(mirrorJobs, NOW));
+    return rows.map((row) => row.id).sort();
+  }
+
+  test("does not return a job that just started and has no checkpoint yet", async () => {
+    seed([
+      { id: "just-started", inProgress: true, startedAt: ago(MINUTE), lastCheckpoint: null },
+      { id: "first-item-running", inProgress: true, startedAt: ago(9 * MINUTE), lastCheckpoint: null },
+    ]);
+
+    expect(await findInterruptedIds()).toEqual([]);
+  });
+
+  test("returns jobs whose missing checkpoint is overdue", async () => {
+    seed([
+      { id: "overdue", inProgress: true, startedAt: ago(11 * MINUTE), lastCheckpoint: null },
+      { id: "no-start", inProgress: true, startedAt: null, lastCheckpoint: null },
+    ]);
+
+    expect(await findInterruptedIds()).toEqual(["no-start", "overdue"]);
+  });
+
+  test("agrees with isJobInterrupted across the whole grid of cases", async () => {
+    const starts: Array<[string, Date | null]> = [
+      ["s-none", null],
+      ["s-1m", ago(MINUTE)],
+      ["s-11m", ago(11 * MINUTE)],
+      ["s-3h", ago(3 * HOUR)],
+    ];
+    const checkpoints: Array<[string, Date | null]> = [
+      ["c-none", null],
+      ["c-2m", ago(2 * MINUTE)],
+      ["c-11m", ago(11 * MINUTE)],
+    ];
+
+    const rows: Row[] = [];
+    for (const inProgress of [true, false]) {
+      for (const [startLabel, startedAt] of starts) {
+        for (const [checkpointLabel, lastCheckpoint] of checkpoints) {
+          rows.push({
+            id: `${inProgress ? "live" : "done"}/${startLabel}/${checkpointLabel}`,
+            inProgress,
+            startedAt,
+            lastCheckpoint,
+          });
+        }
+      }
+    }
+    seed(rows);
+
+    const expected = rows
+      .filter((row) => isJobInterrupted(row, NOW))
+      .map((row) => row.id)
+      .sort();
+
+    expect(expected.length).toBeGreaterThan(0);
+    expect(expected.length).toBeLessThan(rows.length);
+    expect(await findInterruptedIds()).toEqual(expected);
+  });
+});
+
+describe("wiring", () => {
+  const read = (relative: string) => readFileSync(join(import.meta.dir, relative), "utf8");
+  const HELPERS_SRC = read("helpers.ts");
+  const CONCURRENCY_SRC = read("utils/concurrency.ts");
+  const MIDDLEWARE_SRC = read("../middleware.ts");
+
+  test("createMirrorJob writes an initial checkpoint for in-progress jobs", () => {
+    expect(
+      /lastCheckpoint:\s*inProgress\s*\?\s*currentTimestamp\s*:\s*undefined/.test(HELPERS_SRC),
+      "an in-progress job must start with a checkpoint so a null checkpoint means a legacy row, not a new job"
+    ).toBe(true);
+  });
+
+  test("findInterruptedJobs uses the shared condition instead of a bare null-checkpoint match", () => {
+    expect(
+      /\.where\(buildInterruptedJobsCondition\(mirrorJobs\)\)/.test(HELPERS_SRC),
+      "findInterruptedJobs must query through buildInterruptedJobsCondition"
+    ).toBe(true);
+    expect(
+      /or\(\s*isNull\(mirrorJobs\.lastCheckpoint\)/.test(HELPERS_SRC),
+      "a null checkpoint on its own must no longer count as interrupted"
+    ).toBe(false);
+  });
+
+  test("touchMirrorJobCheckpoint only refreshes rows that are still in progress", () => {
+    expect(/export async function touchMirrorJobCheckpoint\(/.test(HELPERS_SRC)).toBe(true);
+    expect(
+      /set\(\{\s*lastCheckpoint:\s*new Date\(\)\s*\}\)\s*\.where\(and\(eq\(mirrorJobs\.id,\s*jobId\),\s*eq\(mirrorJobs\.inProgress,\s*true\)\)\)/.test(
+        HELPERS_SRC
+      ),
+      "the heartbeat must not revive a job that already completed or failed"
+    ).toBe(true);
+  });
+
+  test("processWithResilience heartbeats while items run and stops when the job ends", () => {
+    expect(
+      /setInterval\([\s\S]*?touchMirrorJobCheckpoint\(jobId\)[\s\S]*?JOB_HEARTBEAT_INTERVAL_MS\)/.test(
+        CONCURRENCY_SRC
+      ),
+      "processWithResilience must refresh the checkpoint on a timer"
+    ).toBe(true);
+    expect(
+      /finally\s*\{\s*clearInterval\(heartbeat\);\s*\}/.test(CONCURRENCY_SRC),
+      "the heartbeat must be cleared on every exit path"
+    ).toBe(true);
+  });
+
+  test("middleware keeps the recovery latch until the recovery promise settles", () => {
+    expect(
+      /recoveryPromise\s*\.catch\(\(\)\s*=>\s*false\)\s*\.finally\(\(\)\s*=>\s*\{\s*recoveryInFlight\s*=\s*false;\s*\}\)/.test(
+        MIDDLEWARE_SRC
+      ),
+      "the latch must be released by the recovery promise, not by the request that started it"
+    ).toBe(true);
+    expect(
+      /if\s*\(\s*!latchFollowsRecovery\s*\)\s*\{\s*recoveryInFlight\s*=\s*false;\s*\}/.test(MIDDLEWARE_SRC),
+      "the request-level finally may only release the latch when recovery never started"
+    ).toBe(true);
+    expect(
+      /clearTimeout\(timeoutHandle\)/.test(MIDDLEWARE_SRC),
+      "the bounded wait must not leave a dangling rejection timer"
+    ).toBe(true);
+  });
+});

src/lib/interrupted-job-detection.ts

@@ -0,0 +1,102 @@
+/**
+ * Decides which in-progress mirror jobs count as interrupted and should be
+ * resumed by recovery (issue #372).
+ *
+ * A job is interrupted when any of these hold:
+ *   - it has a checkpoint older than the checkpoint window, or
+ *   - it has no checkpoint and is older than the checkpoint window (or has
+ *     no recorded start at all, as legacy rows may), or
+ *   - it started longer ago than the stale window, whatever its checkpoint.
+ *
+ * The grace period for jobs without a checkpoint is the point of this
+ * module. Checkpoints are written when an item completes, so a job that has
+ * just started, or is on a single long item, has none yet. Treating "no
+ * checkpoint" as "interrupted" made request-time recovery resume jobs that
+ * were still running.
+ *
+ * The rule exists twice on purpose: `isJobInterrupted` is the readable,
+ * directly testable form, and `buildInterruptedJobsCondition` is the SQL
+ * used by `findInterruptedJobs`. The test suite checks they agree.
+ */
+import { and, eq, isNull, lt, or, type SQL } from "drizzle-orm";
+import type { mirrorJobs } from "@/lib/db/schema";
+
+/** A job with no checkpoint newer than this is considered inactive. */
+export const INTERRUPTED_CHECKPOINT_AGE_MS = 10 * 60 * 1000;
+
+/** A job that started longer ago than this is considered stale regardless. */
+export const STALE_JOB_AGE_MS = 2 * 60 * 60 * 1000;
+
+/**
+ * How often a running job refreshes its checkpoint between item
+ * completions. Kept well inside the checkpoint window so one missed beat
+ * does not make a live job look interrupted.
+ */
+export const JOB_HEARTBEAT_INTERVAL_MS = 2 * 60 * 1000;
+
+export interface InterruptedJobCutoffs {
+  /** Checkpoints (or starts, for jobs without one) before this are too old. */
+  checkpointCutoff: Date;
+  /** Starts before this mark the job stale no matter what. */
+  staleCutoff: Date;
+}
+
+export function computeInterruptedJobCutoffs(now: Date = new Date()): InterruptedJobCutoffs {
+  return {
+    checkpointCutoff: new Date(now.getTime() - INTERRUPTED_CHECKPOINT_AGE_MS),
+    staleCutoff: new Date(now.getTime() - STALE_JOB_AGE_MS),
+  };
+}
+
+export interface JobLivenessFields {
+  inProgress: boolean | null | undefined;
+  startedAt: Date | null | undefined;
+  lastCheckpoint: Date | null | undefined;
+}
+
+/**
+ * Pure form of the interrupted-job rule. Must stay equivalent to
+ * `buildInterruptedJobsCondition`.
+ */
+export function isJobInterrupted(job: JobLivenessFields, now: Date = new Date()): boolean {
+  if (!job.inProgress) return false;
+
+  const { checkpointCutoff, staleCutoff } = computeInterruptedJobCutoffs(now);
+
+  if (job.startedAt && job.startedAt < staleCutoff) return true;
+
+  if (job.lastCheckpoint) return job.lastCheckpoint < checkpointCutoff;
+
+  // No checkpoint yet: give the job the same window it would get after a
+  // checkpoint. A row with no start time either has nothing to prove it
+  // is alive, so it is treated as interrupted.
+  return !job.startedAt || job.startedAt < checkpointCutoff;
+}
+
+/**
+ * SQL form of the interrupted-job rule for `findInterruptedJobs`. Takes the
+ * table so tests can run it against an in-memory database without the
+ * application database module.
+ */
+export function buildInterruptedJobsCondition(
+  table: typeof mirrorJobs,
+  now: Date = new Date(),
+): SQL {
+  const { checkpointCutoff, staleCutoff } = computeInterruptedJobCutoffs(now);
+
+  return and(
+    eq(table.inProgress, true),
+    or(
+      // No checkpoint yet, and old enough that one should exist by now
+      // (or no recorded start to judge by).
+      and(
+        isNull(table.lastCheckpoint),
+        or(isNull(table.startedAt), lt(table.startedAt, checkpointCutoff)),
+      ),
+      // Checkpoint present but too old.
+      lt(table.lastCheckpoint, checkpointCutoff),
+      // Running far too long, whatever the checkpoint says.
+      lt(table.startedAt, staleCutoff),
+    ),
+  )!;
+}

src/lib/utils/concurrency.ts

@@ -1,3 +1,5 @@
+import { JOB_HEARTBEAT_INTERVAL_MS } from "@/lib/interrupted-job-detection";
+
 /**
  * Utility for processing items in parallel with concurrency control
  *
@@ -225,7 +227,7 @@ export async function processWithResilience<T, R>(
   } = options;
 
   // Import helpers for job management and shutdown handling
-  const { createMirrorJob, updateMirrorJobProgress } = await import('@/lib/helpers');
+  const { createMirrorJob, updateMirrorJobProgress, touchMirrorJobCheckpoint } = await import('@/lib/helpers');
 
   // Import shutdown manager (with fallback for testing)
   let registerActiveJob: (jobId: string) => void = () => {};
@@ -297,6 +299,16 @@ export async function processWithResilience<T, R>(
   // Register the job with the shutdown manager
   registerActiveJob(jobId);
 
+  // Heartbeat while items are processing. Checkpoints are only written when
+  // an item completes, so a single long item would otherwise let the job
+  // cross the interrupted-job cutoff and be "resumed" by recovery while it
+  // is still running (issue #372).
+  const heartbeat = setInterval(() => {
+    touchMirrorJobCheckpoint(jobId).catch((error: unknown) => {
+      console.warn(`Failed to refresh checkpoint for job ${jobId}:`, error);
+    });
+  }, JOB_HEARTBEAT_INTERVAL_MS);
+
   // Define the checkpoint function
   const onCheckpoint = async (jobId: string, completedItemId: string) => {
     const itemName = items.find(item => getItemId(item) === completedItemId)
@@ -361,5 +373,7 @@ export async function processWithResilience<T, R>(
     unregisterActiveJob(jobId);
 
     throw error;
+  } finally {
+    clearInterval(heartbeat);
   }
 }

src/middleware.ts

@@ -27,6 +27,8 @@ let recoveryInitialized = false;
 // initializeRecovery() 5-minute throttle inside recovery.ts, which is
 // keyed on `lastRecoveryAttempt`). This prevents one middleware
 // invocation from triggering recovery while another is in flight.
+// Once a recovery run has started, the latch is held until that run
+// settles, even if the request that started it stops waiting.
 let recoveryInFlight = false;
 let cleanupServiceStarted = false;
 let schedulerServiceStarted = false;
@@ -128,14 +130,20 @@ export const onRequest = defineMiddleware(async (context, next) => {
   // The previous implementation used a once-per-process gate, so
   // any mid-runtime interruption (a sync that started after boot,
   // crashed mid-flight, and never got back to the resume path)
-  // would sit at `in_progress=true` forever — the periodic detector
+  // would sit at `in_progress=true` forever: the periodic detector
   // kept finding it, but the resumer never re-fired. This block
   // now re-evaluates on every request, gated by `recoveryInFlight`
   // (per-process) plus the 5-minute throttle inside
   // `initializeRecovery()` (which prevents thrashing if a resume
   // cycle keeps failing).
+  //
+  // A request only waits a bounded time for recovery, but the latch
+  // stays held until the recovery promise itself settles. Releasing
+  // it on the request timeout let later requests start overlapping
+  // attempts against the same jobs (issue #372).
   if (!recoveryInFlight) {
     recoveryInFlight = true;
+    let latchFollowsRecovery = false;
 
     try {
       // Check if recovery is actually needed before attempting
@@ -149,22 +157,34 @@ export const onRequest = defineMiddleware(async (context, next) => {
         }
         console.log('Attempting recovery from middleware...');
 
-        // Run recovery with a shorter timeout since this is during request handling
-        const recoveryResult = await Promise.race([
-          initializeRecovery({
-            skipIfRecentAttempt: true,
-            maxRetries: 2,
-            retryDelay: 3000,
-          }),
-          new Promise<boolean>((_, reject) => {
-            setTimeout(() => reject(new Error('Middleware recovery timeout')), 15000);
-          })
-        ]);
+        const recoveryPromise = initializeRecovery({
+          skipIfRecentAttempt: true,
+          maxRetries: 2,
+          retryDelay: 3000,
+        });
+        latchFollowsRecovery = true;
+        recoveryPromise
+          .catch(() => false)
+          .finally(() => {
+            recoveryInFlight = false;
+          });
 
-        if (recoveryResult) {
-          console.log('✅ Middleware recovery completed successfully');
-        } else {
-          console.log('⚠️  Middleware recovery completed with some issues');
+        // Bound how long this request waits; the recovery keeps running.
+        let timeoutHandle: ReturnType<typeof setTimeout> | undefined;
+        const timeout = new Promise<boolean>((_, reject) => {
+          timeoutHandle = setTimeout(() => reject(new Error('Middleware recovery timeout')), 15000);
+        });
+
+        try {
+          const recoveryResult = await Promise.race([recoveryPromise, timeout]);
+
+          if (recoveryResult) {
+            console.log('✅ Middleware recovery completed successfully');
+          } else {
+            console.log('⚠️  Middleware recovery completed with some issues');
+          }
+        } finally {
+          clearTimeout(timeoutHandle);
         }
       } else if (!recoveryInitialized) {
         // Only log this on the first request; otherwise we'd spam
@@ -183,7 +203,11 @@ export const onRequest = defineMiddleware(async (context, next) => {
 
       recoveryInitialized = true;
     } finally {
-      recoveryInFlight = false;
+      // Once recovery has started, the latch is released when it settles
+      // (see above), not when this request stops waiting.
+      if (!latchFollowsRecovery) {
+        recoveryInFlight = false;
+      }
     }
   }