From f767a0a2b3e79bd7713f3d97b60de2acfff196d7 Mon Sep 17 00:00:00 2001 From: ARUNAVO RAY Date: Wed, 2 Sep 2026 11:59:37 +0530 Subject: [PATCH] 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 | 4 +- src/lib/helpers.ts | 49 +++-- src/lib/interrupted-job-detection.test.ts | 254 ++++++++++++++++++++++ src/lib/interrupted-job-detection.ts | 102 +++++++++ src/lib/utils/concurrency.ts | 16 +- src/middleware.ts | 58 +++-- 6 files changed, 443 insertions(+), 40 deletions(-) create mode 100644 src/lib/interrupted-job-detection.test.ts create mode 100644 src/lib/interrupted-job-detection.ts diff --git a/docs/RECOVERY_IMPROVEMENTS.md b/docs/RECOVERY_IMPROVEMENTS.md index 80348c7..d961c17 100644 --- a/docs/RECOVERY_IMPROVEMENTS.md +++ b/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 diff --git a/src/lib/helpers.ts b/src/lib/helpers.ts index 4ae1eb8..90f61b2 100644 --- a/src/lib/helpers.ts +++ b/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 { + 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. diff --git a/src/lib/interrupted-job-detection.test.ts b/src/lib/interrupted-job-detection.test.ts new file mode 100644 index 0000000..77d1eb4 --- /dev/null +++ b/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); + }); +}); diff --git a/src/lib/interrupted-job-detection.ts b/src/lib/interrupted-job-detection.ts new file mode 100644 index 0000000..62dbf2f --- /dev/null +++ b/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), + ), + )!; +} diff --git a/src/lib/utils/concurrency.ts b/src/lib/utils/concurrency.ts index 595e4cc..460d3e7 100644 --- a/src/lib/utils/concurrency.ts +++ b/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( } = 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( // 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( unregisterActiveJob(jobId); throw error; + } finally { + clearInterval(heartbeat); } } diff --git a/src/middleware.ts b/src/middleware.ts index 9bd681c..5218ab4 100644 --- a/src/middleware.ts +++ b/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((_, 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 | undefined; + const timeout = new Promise((_, 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; + } } }