From 3a42945fccbaedea2714b0a684df14cc92fb697a Mon Sep 17 00:00:00 2001 From: Jyong Date: Wed, 29 Jul 2026 11:13:19 -0400 Subject: [PATCH] Refresh retry base revision and preserve immutable candidates --- .../document-compilation-attempt-job.test.ts | 50 ++++++++++++++++- .../src/document-compilation-attempt-job.ts | 44 ++++++++++----- ...ent-compilation-attempt-repository.test.ts | 54 +++++++++++++++++++ ...document-compilation-attempt-repository.ts | 45 ++++++++++++++++ 4 files changed, 178 insertions(+), 15 deletions(-) diff --git a/knowledge-fs/packages/api/src/document-compilation-attempt-job.test.ts b/knowledge-fs/packages/api/src/document-compilation-attempt-job.test.ts index be734c2cae6..98982ed8bd7 100644 --- a/knowledge-fs/packages/api/src/document-compilation-attempt-job.test.ts +++ b/knowledge-fs/packages/api/src/document-compilation-attempt-job.test.ts @@ -1,7 +1,10 @@ import { describe, expect, it, vi } from "vitest"; import { createDurableDocumentCompilationJobStateMachine } from "./document-compilation-attempt-job"; -import { createInMemoryDocumentCompilationAttemptRepository } from "./document-compilation-attempt-repository"; +import { + DocumentCompilationAttemptHeadConflictError, + createInMemoryDocumentCompilationAttemptRepository, +} from "./document-compilation-attempt-repository"; const attemptId = "11111111-1111-4111-8111-111111111111"; const outboxId = "22222222-2222-4222-8222-222222222222"; @@ -196,6 +199,7 @@ describe("durable document compilation job control plane", () => { it("cancels before dispatch and reactivates a user-canceled attempt on retry", async () => { let now = "2026-07-13T10:00:00.000Z"; + let publicationHeadRevision = 0; const attempts = createInMemoryDocumentCompilationAttemptRepository(); const jobs = createDurableDocumentCompilationJobStateMachine({ attempts, @@ -204,7 +208,7 @@ describe("durable document compilation job control plane", () => { generatePublicationGenerationId: () => generationId, maxExecutionAttempts: 5, now: () => now, - resolveBaseHeadRevision: async () => 0, + resolveBaseHeadRevision: async () => publicationHeadRevision, }); const started = await jobs.start({ documentAssetId: assetId, @@ -220,12 +224,54 @@ describe("durable document compilation job control plane", () => { stage: "canceled", }); now = "2026-07-13T10:02:00.000Z"; + publicationHeadRevision = 1; await expect(jobs.retry?.(started.id)).resolves.toMatchObject({ + baseHeadRevision: 1, runState: "dispatch_pending", stage: "queued", }); }); + it("refreshes the retry base when the publication head changes during admission", async () => { + let now = "2026-07-13T10:00:00.000Z"; + let publicationHeadRevision = 0; + const repository = createInMemoryDocumentCompilationAttemptRepository(); + const retryTerminal = vi.fn(async (input) => { + if (retryTerminal.mock.calls.length === 1) { + publicationHeadRevision = 2; + throw new DocumentCompilationAttemptHeadConflictError( + input.baseHeadRevision ?? -1, + publicationHeadRevision, + ); + } + return repository.retryTerminal(input); + }); + const jobs = createDurableDocumentCompilationJobStateMachine({ + attempts: { ...repository, retryTerminal }, + generateAttemptId: () => attemptId, + generateOutboxId: () => outboxId, + generatePublicationGenerationId: () => generationId, + maxExecutionAttempts: 5, + now: () => now, + resolveBaseHeadRevision: async () => publicationHeadRevision, + }); + const started = await jobs.start({ + documentAssetId: assetId, + knowledgeSpaceId: spaceId, + tenantId: "tenant-1", + version: 1, + }); + await jobs.cancel(started.id, "user request"); + + now = "2026-07-13T10:01:00.000Z"; + publicationHeadRevision = 1; + await expect(jobs.retry?.(started.id)).resolves.toMatchObject({ + baseHeadRevision: 2, + runState: "dispatch_pending", + }); + expect(retryTerminal.mock.calls.map(([input]) => input.baseHeadRevision)).toEqual([1, 2]); + }); + it("rechecks deletion admission and binds a fresh caller permission before retry", async () => { let deletionActive = false; let now = "2026-07-13T10:00:00.000Z"; diff --git a/knowledge-fs/packages/api/src/document-compilation-attempt-job.ts b/knowledge-fs/packages/api/src/document-compilation-attempt-job.ts index 773ddee0dac..8a952b57353 100644 --- a/knowledge-fs/packages/api/src/document-compilation-attempt-job.ts +++ b/knowledge-fs/packages/api/src/document-compilation-attempt-job.ts @@ -118,20 +118,38 @@ export function createDurableDocumentCompilationJobStateMachine({ knowledgeSpaceId: current.knowledgeSpaceId, tenantId: current.tenantId, }); - const retried = await attempts.retryTerminal({ - attemptId: current.id, - expectedRowVersion: current.rowVersion, - now: now(), - ...permissionBinding, - }); - if (!retried) { - throw new Error("Document compilation attempt cannot be retried"); + for (let retry = 0; retry < maxHeadConflictRetries; retry += 1) { + const baseHeadRevision = + current.candidatePublicationId || current.candidateFingerprint + ? current.baseHeadRevision + : await resolveBaseHeadRevision(current); + validateNonnegativeInteger(baseHeadRevision, "baseHeadRevision"); + try { + const retried = await attempts.retryTerminal({ + attemptId: current.id, + baseHeadRevision, + expectedRowVersion: current.rowVersion, + now: now(), + ...permissionBinding, + }); + if (!retried) { + throw new Error("Document compilation attempt cannot be retried"); + } + recordDurableTaskOperationalMetric(metrics, { + lifecycle: "retry", + taskKind: "document_compilation", + }); + return attemptToCompilationJob(retried); + } catch (error) { + if ( + !(error instanceof DocumentCompilationAttemptHeadConflictError) || + retry === maxHeadConflictRetries - 1 + ) { + throw error; + } + } } - recordDurableTaskOperationalMetric(metrics, { - lifecycle: "retry", - taskKind: "document_compilation", - }); - return attemptToCompilationJob(retried); + throw new Error("Document compilation retry could not snapshot the publication head"); }, start: async (input) => { const normalized = normalizeStartInput(input); diff --git a/knowledge-fs/packages/api/src/document-compilation-attempt-repository.test.ts b/knowledge-fs/packages/api/src/document-compilation-attempt-repository.test.ts index 326e758a843..502797e9fef 100644 --- a/knowledge-fs/packages/api/src/document-compilation-attempt-repository.test.ts +++ b/knowledge-fs/packages/api/src/document-compilation-attempt-repository.test.ts @@ -541,6 +541,55 @@ describe("in-memory document compilation attempt repository", () => { ).resolves.toBeNull(); }); + it("never rebases a terminal attempt after its immutable candidate is bound", async () => { + const repository = createInMemoryDocumentCompilationAttemptRepository(); + await repository.start(startInput()); + await dispatch(repository, lockToken, "queue-1"); + const running = await repository.claim({ + attemptId, + expectedRowVersion: 1, + leaseExpiresAt: "2026-07-13T12:02:00.000Z", + leaseToken, + now: "2026-07-13T12:00:02.000Z", + queueJobId: "queue-1", + workerId: "worker-1", + }); + const candidatePublicationId = "018f0d60-7a49-7cc2-9c1b-5b36f18f3901"; + const candidateFingerprint = `projection-set-sha256:${"e".repeat(64)}`; + const bound = await repository.advance({ + attemptId, + candidateFingerprint, + candidatePublicationId, + checkpoint: "queued", + expectedRowVersion: running?.rowVersion ?? -1, + leaseToken, + now: "2026-07-13T12:00:03.000Z", + }); + const failed = await repository.fail({ + attemptId, + errorCode: "PUBLISH_FAILED", + errorMessage: "publication failed", + expectedRowVersion: bound?.rowVersion ?? -1, + leaseToken, + now: "2026-07-13T12:00:04.000Z", + }); + + await expect( + repository.retryTerminal({ + attemptId, + baseHeadRevision: 3, + expectedRowVersion: failed?.rowVersion ?? -1, + now: "2026-07-13T12:01:00.000Z", + }), + ).rejects.toThrow("cannot rebase an immutable candidate"); + await expect(repository.get(attemptId)).resolves.toMatchObject({ + baseHeadRevision: 2, + candidateFingerprint, + candidatePublicationId, + runState: "failed", + }); + }); + it("CAS-fails exhausted active work and releases its active slot", async () => { const repository = createInMemoryDocumentCompilationAttemptRepository(); await repository.start(startInput({ maxExecutionAttempts: 1 })); @@ -1346,6 +1395,9 @@ describe("database document compilation attempt repository", () => { if (input.tableName === "document_assets") { return result([activeDocumentAssetRow()], 0); } + if (input.tableName === "projection_set_publication_heads") { + return result([{ head_revision: 3 }], 0); + } if (input.tableName === "logical_documents" && input.operation === "select") { return input.sql.includes(" JOIN ") ? result([{ id: logicalDocumentId }], 0) @@ -1398,6 +1450,7 @@ describe("database document compilation attempt repository", () => { await expect( repository.retryTerminal({ attemptId, + baseHeadRevision: 3, expectedRowVersion: 4, now: "2026-07-13T12:06:00.000Z", permissionSnapshot: { @@ -1408,6 +1461,7 @@ describe("database document compilation attempt repository", () => { requestedBySubjectId: "current-editor", }), ).resolves.toMatchObject({ + baseHeadRevision: 3, permissionSnapshot: { accessChannel: "interactive", id: freshSnapshotId, diff --git a/knowledge-fs/packages/api/src/document-compilation-attempt-repository.ts b/knowledge-fs/packages/api/src/document-compilation-attempt-repository.ts index f0c6c851ac6..c6ec4d98fdc 100644 --- a/knowledge-fs/packages/api/src/document-compilation-attempt-repository.ts +++ b/knowledge-fs/packages/api/src/document-compilation-attempt-repository.ts @@ -252,6 +252,12 @@ export interface SupersedeDocumentCompilationAttemptInput { export interface RetryTerminalDocumentCompilationAttemptInput { readonly attemptId: string; + /** + * Fresh publication head captured by the control plane. It rebases only attempts that have not + * bound an immutable candidate; bound candidates retain their original base for idempotent + * recovery. + */ + readonly baseHeadRevision?: number | undefined; readonly capabilityGrantId?: string | undefined; readonly availableAt?: string | undefined; readonly expectedRowVersion: number; @@ -736,6 +742,7 @@ export function createInMemoryDocumentCompilationAttemptRepository( if (active) { return null; } + const baseHeadRevision = terminalRetryBaseHeadRevision(current, input); const event = requiredMemoryOutbox(outbox, current.id); const now = canonicalDateTime(input.now, "now"); const nextEvent = parseOutboxEvent({ @@ -755,6 +762,7 @@ export function createInMemoryDocumentCompilationAttemptRepository( const nextAttempt = parseAttempt({ ...current, activeSlot: 1, + baseHeadRevision, completedAt: undefined, executionAttempts: 0, externalJobId: undefined, @@ -1335,6 +1343,20 @@ export function createDatabaseDocumentCompilationAttemptRepository({ if (active) { return null; } + const baseHeadRevision = terminalRetryBaseHeadRevision(current, input); + if (input.baseHeadRevision !== undefined && !hasBoundCompilationCandidate(current)) { + const actualHeadRevision = await databaseCurrentHeadRevision( + database, + transaction, + current, + ); + if (actualHeadRevision !== baseHeadRevision) { + throw new DocumentCompilationAttemptHeadConflictError( + baseHeadRevision, + actualHeadRevision, + ); + } + } await requireDatabaseCompilationControlResources( database, transaction, @@ -1369,6 +1391,7 @@ export function createDatabaseDocumentCompilationAttemptRepository({ { ...current, activeSlot: 1, + baseHeadRevision, completedAt: undefined, executionAttempts: 0, externalJobId: undefined, @@ -1623,6 +1646,28 @@ function parseRetryPermissionBinding( }; } +function terminalRetryBaseHeadRevision( + current: DocumentCompilationAttempt, + input: Pick, +): number { + const requested = + input.baseHeadRevision === undefined + ? current.baseHeadRevision + : nonnegativeInteger(input.baseHeadRevision, "baseHeadRevision"); + if (hasBoundCompilationCandidate(current) && requested !== current.baseHeadRevision) { + throw new DocumentCompilationAttemptTransitionError( + "Document compilation retry cannot rebase an immutable candidate", + ); + } + return requested; +} + +function hasBoundCompilationCandidate( + attempt: Pick, +): boolean { + return attempt.candidateFingerprint !== undefined || attempt.candidatePublicationId !== undefined; +} + function parseStartInput( input: StartDocumentCompilationAttemptInput, ): ParsedStartDocumentCompilationAttemptInput {