From 22a0aafcc1cf09f41682ca097e712b349d9af347 Mon Sep 17 00:00:00 2001 From: ARUNAVO RAY Date: Fri, 11 Sep 2026 05:35:03 +0530 Subject: [PATCH] fix(recovery): stop resuming jobs that are still running and checkpoint every item (#423) Three things combined to make crash recovery restart live jobs: - isJobInterrupted treated any job older than two hours as interrupted before looking at its checkpoint, so a large job that heartbeated every two minutes was resumed while it was still running. The age rule is gone: a checkpoint is the job's proof of life, and only a missing or stale one (ten minutes) marks a job interrupted. - processWithRetry only checkpointed the item that tripped the interval, so a resume re-processed up to N-1 finished items. Every completed item is now recorded; the interval only throttles the published progress event. - updateMirrorJobProgress was a read-modify-write of completedItemIds with no lock, so two items finishing in the same tick could drop each other. Writes are serialized per job with the keyed lock from #419. A resumed job also records a fresh start time. Found while investigating #417. --- CHANGELOG.md | 5 +++ src/lib/helpers.ts | 53 ++++++++++++++++++----- src/lib/interrupted-job-detection.test.ts | 35 ++++++++++++--- src/lib/interrupted-job-detection.ts | 30 ++++++------- src/lib/utils/concurrency.test.ts | 43 ++++++++++++++++++ src/lib/utils/concurrency.ts | 32 ++++++++------ 6 files changed, 153 insertions(+), 45 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6ffceea..7a22807 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -42,6 +42,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Comprehensive fix report documentation ### Fixed +- Crash recovery no longer resumes a job that is still running + - A job older than two hours was treated as interrupted even while it checkpointed every two minutes, so recovery started a second pass over the same repositories alongside the original; the age rule is gone and only a missing or stale checkpoint marks a job interrupted + - Every completed item is now recorded in the job's checkpoint instead of one in every N, so a real resume skips exactly what was done; progress events are still throttled + - Concurrent items completing at the same time could overwrite each other's checkpoint; job progress writes are serialized per job + - A resumed job records a fresh start time - A per-organization Mirror Destination equal to the organization's own name is kept instead of being dropped (#416) - The editor treated a typed value matching the organization name as "reset to default" and saved no override, which only holds under the preserve strategy; under single-org the default is the destination organization, so the repositories went there - The organization card now shows the default for the configured strategy in the preview, the placeholder, the helper text and the reset button, and marks any stored destination as custom diff --git a/src/lib/helpers.ts b/src/lib/helpers.ts index 90f61b2..e8381e3 100644 --- a/src/lib/helpers.ts +++ b/src/lib/helpers.ts @@ -4,6 +4,7 @@ import { eq, and } from "drizzle-orm"; import { buildInterruptedJobsCondition } from "./interrupted-job-detection"; import { v4 as uuidv4 } from "uuid"; import { publishEvent } from "./events"; +import { withKeyedLock } from "./utils/keyed-mutex"; import { triggerJobNotification } from "./notification-service"; export async function createMirrorJob({ @@ -113,17 +114,18 @@ export async function createMirrorJob({ } /** - * Updates the progress of a mirror job + * Updates the progress of a mirror job. + * + * Progress is a read-modify-write of the completedItemIds list, and items + * finish concurrently, so the update runs under a per-job lock: two items + * completing in the same tick used to read the same list and one of them + * silently overwrote the other. A resume then re-processed the lost item. + * + * `silent` records the progress without publishing an event. Every completed + * item is checkpointed so a resume knows exactly what is done; the events the + * dashboard listens to are throttled by the caller instead. */ -export async function updateMirrorJobProgress({ - jobId, - completedItemId, - status, - message, - details, - inProgress, - isCompleted, -}: { +export async function updateMirrorJobProgress(params: { jobId: string; completedItemId?: string; status?: RepoStatus; @@ -131,7 +133,25 @@ export async function updateMirrorJobProgress({ details?: string; inProgress?: boolean; isCompleted?: boolean; + /** Record a new start time, as when recovery resumes the job. */ + startedAt?: Date; + /** Write the progress without publishing a mirror-status event. */ + silent?: boolean; }) { + return withKeyedLock(`mirror-job:${params.jobId}`, () => updateMirrorJobProgressLocked(params)); +} + +async function updateMirrorJobProgressLocked({ + jobId, + completedItemId, + status, + message, + details, + inProgress, + isCompleted, + startedAt, + silent = false, +}: Parameters[0]) { try { // Get the current job const [job] = await db @@ -157,6 +177,10 @@ export async function updateMirrorJobProgress({ } } + if (startedAt) { + updates.startedAt = startedAt; + } + // Update status if provided if (status) { updates.status = status; @@ -195,6 +219,10 @@ export async function updateMirrorJobProgress({ ...updates, }; + if (silent) { + return updatedJob; + } + // Create deduplication key for progress updates let deduplicationKey: string | undefined; if (completedItemId) { @@ -324,12 +352,15 @@ export async function resumeInterruptedJob(job: any) { return null; } - // Update the job to show it's being resumed + // Update the job to show it's being resumed. The start time moves to + // now so the row records this pass; the fresh checkpoint written here is + // what keeps a second recovery sweep from resuming the same job again. await updateMirrorJobProgress({ jobId: job.id, message: `Resuming job with ${remainingItemIds.length} remaining items`, details: `Job was interrupted and is being resumed. ${job.completedItemIds.length} of ${job.itemIds.length} items were already processed.`, inProgress: true, + startedAt: new Date(), }); return { diff --git a/src/lib/interrupted-job-detection.test.ts b/src/lib/interrupted-job-detection.test.ts index 77d1eb4..7ebcc95 100644 --- a/src/lib/interrupted-job-detection.test.ts +++ b/src/lib/interrupted-job-detection.test.ts @@ -18,7 +18,6 @@ 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, @@ -32,14 +31,15 @@ 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", () => { + test("keeps the 10 minute checkpoint window and has no stale age cutoff", () => { 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); + // The two hour "stale regardless of checkpoint" rule resumed live jobs + // and is gone; a checkpointing job is alive however old it is. + expect(Object.keys(cutoffs)).toEqual(["checkpointCutoff"]); }); test("heartbeat interval leaves room for a missed beat inside the checkpoint window", () => { @@ -78,9 +78,18 @@ describe("isJobInterrupted", () => { ).toBe(true); }); - test("a job running past the stale window is interrupted even with a fresh checkpoint", () => { + test("a job that keeps checkpointing stays live however long ago it started", () => { expect( isJobInterrupted({ inProgress: true, startedAt: ago(3 * HOUR), lastCheckpoint: ago(MINUTE) }, NOW) + ).toBe(false); + expect( + isJobInterrupted({ inProgress: true, startedAt: ago(30 * HOUR), lastCheckpoint: ago(2 * MINUTE) }, NOW) + ).toBe(false); + }); + + test("an old job whose checkpoint went stale is still interrupted", () => { + expect( + isJobInterrupted({ inProgress: true, startedAt: ago(3 * HOUR), lastCheckpoint: ago(11 * MINUTE) }, NOW) ).toBe(true); }); @@ -235,6 +244,22 @@ describe("wiring", () => { ).toBe(true); }); + test("resumeInterruptedJob records a fresh start time for the resumed pass", () => { + const resumeBody = HELPERS_SRC.slice(HELPERS_SRC.indexOf("export async function resumeInterruptedJob(")); + expect(/inProgress:\s*true,\s*startedAt:\s*new Date\(\)/.test(resumeBody)).toBe(true); + }); + + test("updateMirrorJobProgress serializes writes per job so concurrent items cannot drop each other", () => { + expect( + /withKeyedLock\(`mirror-job:\$\{params\.jobId\}`,\s*\(\)\s*=>\s*updateMirrorJobProgressLocked\(params\)\)/.test(HELPERS_SRC) + ).toBe(true); + }); + + test("processWithRetry checkpoints every completed item instead of the one that trips the interval", () => { + expect(/await onCheckpoint\(jobId,\s*itemId,\s*publish\)/.test(CONCURRENCY_SRC)).toBe(true); + expect(CONCURRENCY_SRC.includes("onCheckpoint(jobId, 'final')")).toBe(false); + }); + 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( diff --git a/src/lib/interrupted-job-detection.ts b/src/lib/interrupted-job-detection.ts index 62dbf2f..eaf2c1e 100644 --- a/src/lib/interrupted-job-detection.ts +++ b/src/lib/interrupted-job-detection.ts @@ -2,11 +2,20 @@ * 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: + * A job is interrupted when either of these holds: * - 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. + * no recorded start at all, as legacy rows may). + * + * A job that keeps checkpointing is alive, however long ago it started. + * Running jobs refresh their checkpoint on a heartbeat between items, so a + * missing or old checkpoint is the only evidence that the process behind a + * job is gone. There used to be a third rule, "started more than two hours + * ago, whatever the checkpoint says", which resumed large jobs while they + * were still running: the resumed pass then re-processed the same + * repositories alongside the original, which is one way release assets got + * duplicated (#417). A job that hangs without ever finishing an item is a + * different problem and is not what recovery is for. * * 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 @@ -24,9 +33,6 @@ 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 @@ -37,14 +43,11 @@ 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), }; } @@ -61,10 +64,9 @@ export interface JobLivenessFields { 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; + const { checkpointCutoff } = computeInterruptedJobCutoffs(now); + // A checkpoint is the job's own proof of life; its age decides on its own. if (job.lastCheckpoint) return job.lastCheckpoint < checkpointCutoff; // No checkpoint yet: give the job the same window it would get after a @@ -82,7 +84,7 @@ export function buildInterruptedJobsCondition( table: typeof mirrorJobs, now: Date = new Date(), ): SQL { - const { checkpointCutoff, staleCutoff } = computeInterruptedJobCutoffs(now); + const { checkpointCutoff } = computeInterruptedJobCutoffs(now); return and( eq(table.inProgress, true), @@ -95,8 +97,6 @@ export function buildInterruptedJobsCondition( ), // 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.test.ts b/src/lib/utils/concurrency.test.ts index 2196b67..db5af6d 100644 --- a/src/lib/utils/concurrency.test.ts +++ b/src/lib/utils/concurrency.test.ts @@ -165,3 +165,46 @@ describe("processWithRetry", () => { } }); }); + +describe("processWithRetry checkpointing", () => { + test("records every completed item and publishes on the interval", async () => { + const items = ["a", "b", "c", "d", "e"]; + const calls: Array<[string, string, boolean]> = []; + + await processWithRetry(items, async (item) => item.toUpperCase(), { + concurrencyLimit: 1, + jobId: "job-1", + getItemId: (item) => item, + checkpointInterval: 2, + onCheckpoint: async (jobId, itemId, publish) => { + calls.push([jobId, itemId, publish]); + }, + }); + + // One checkpoint per item, in completion order, and no "final" placeholder. + expect(calls.map(([, itemId]) => itemId)).toEqual(items); + expect(calls.every(([jobId]) => jobId === "job-1")).toBe(true); + // Progress events only every second item. + expect(calls.map(([, , publish]) => publish)).toEqual([false, true, false, true, false]); + }); + + test("does not checkpoint an item that failed for good", async () => { + const calls: string[] = []; + + // processInParallel logs a failed item and carries on, so the call resolves. + await processWithRetry(["ok", "bad"], async (item) => { + if (item === "bad") throw new Error("nope"); + return item; + }, { + concurrencyLimit: 1, + maxRetries: 0, + jobId: "job-2", + getItemId: (item) => item, + onCheckpoint: async (_jobId, itemId) => { + calls.push(itemId); + }, + }); + + expect(calls).toEqual(["ok"]); + }); +}); diff --git a/src/lib/utils/concurrency.ts b/src/lib/utils/concurrency.ts index 460d3e7..48a140c 100644 --- a/src/lib/utils/concurrency.ts +++ b/src/lib/utils/concurrency.ts @@ -93,8 +93,15 @@ export async function processWithRetry( onRetry?: (item: T, error: Error, attempt: number) => void; jobId?: string; // Optional job ID for checkpointing getItemId?: (item: T) => string; // Function to get a unique ID for each item - onCheckpoint?: (jobId: string, completedItemId: string) => Promise; // Callback for checkpointing - checkpointInterval?: number; // How many items to process before checkpointing + /** + * Called once for EVERY completed item, so a resume knows exactly what is + * done. `publish` is true every `checkpointInterval` items: the callback + * uses it to throttle progress events, not the checkpoint itself. It used + * to be called only for the item that tripped the interval, and a resume + * re-processed everything finished since the previous checkpoint. + */ + onCheckpoint?: (jobId: string, completedItemId: string, publish: boolean) => Promise; + checkpointInterval?: number; // How many completed items between published progress events } = {} ): Promise { const { @@ -130,16 +137,16 @@ export async function processWithRetry( const result = await processItem(item); - // Handle checkpointing if enabled + // Record every completed item; publish progress on the interval. if (jobId && getItemId && onCheckpoint) { const itemId = getItemId(item); itemsProcessedSinceLastCheckpoint++; - // Checkpoint based on the interval - if (itemsProcessedSinceLastCheckpoint >= checkpointInterval) { - await onCheckpoint(jobId, itemId); + const publish = itemsProcessedSinceLastCheckpoint >= checkpointInterval; + if (publish) { itemsProcessedSinceLastCheckpoint = 0; } + await onCheckpoint(jobId, itemId, publish); } return result; @@ -186,12 +193,6 @@ export async function processWithRetry( onProgress ); - // Final checkpoint if there are remaining items since the last checkpoint - if (jobId && getItemId && onCheckpoint && itemsProcessedSinceLastCheckpoint > 0) { - // We don't have a specific item ID for the final checkpoint, so we'll use a placeholder - await onCheckpoint(jobId, 'final'); - } - return results; } @@ -309,8 +310,10 @@ export async function processWithResilience( }); }, JOB_HEARTBEAT_INTERVAL_MS); - // Define the checkpoint function - const onCheckpoint = async (jobId: string, completedItemId: string) => { + // Define the checkpoint function. Every completed item is written to the + // job row so a resume skips exactly what is done; only every + // checkpointInterval-th write also publishes a progress event. + const onCheckpoint = async (jobId: string, completedItemId: string, publish: boolean = true) => { const itemName = items.find(item => getItemId(item) === completedItemId) ? getItemName(items.find(item => getItemId(item) === completedItemId)!) : 'unknown'; @@ -319,6 +322,7 @@ export async function processWithResilience( jobId, completedItemId, message: `Processed item: ${itemName}`, + silent: !publish, }); };