opensource/dify
2
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-09-05 08:48:10 +08:00
Code Issues Packages Projects Releases Wiki Activity

Refresh retry base revision and preserve immutable candidates

Browse Source
This commit is contained in:
Jyong 2026-07-29 11:13:19 -04:00
parent b8a2f991c1
commit 3a42945fcc
4 changed files with 178 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
knowledge-fs/packages/api/src/document-compilation-attempt-job.test.ts
View File

@ -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";

44
knowledge-fs/packages/api/src/document-compilation-attempt-job.ts
View File

@ -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);

54
knowledge-fs/packages/api/src/document-compilation-attempt-repository.test.ts
View File

@ -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,

45
knowledge-fs/packages/api/src/document-compilation-attempt-repository.ts
View File

@ -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<RetryTerminalDocumentCompilationAttemptInput, "baseHeadRevision">,
): 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<DocumentCompilationAttempt, "candidateFingerprint" | "candidatePublicationId">,
): boolean {
return attempt.candidateFingerprint !== undefined || attempt.candidatePublicationId !== undefined;
}
function parseStartInput(
input: StartDocumentCompilationAttemptInput,
): ParsedStartDocumentCompilationAttemptInput {