From cc80265bb2664ddee04492c7a1652c53493b0639 Mon Sep 17 00:00:00 2001 From: Wenxin Duan Date: Tue, 15 Sep 2026 01:14:25 +0800 Subject: [PATCH] fix: recover matching durable image video submission receipts --- packages/harness/src/image-video/service.ts | 20 ++++++++-- packages/harness/tests/image-video.test.ts | 42 +++++++++++++++++++++ 2 files changed, 59 insertions(+), 3 deletions(-) diff --git a/packages/harness/src/image-video/service.ts b/packages/harness/src/image-video/service.ts index 5a9c4df..8ae4b76 100644 --- a/packages/harness/src/image-video/service.ts +++ b/packages/harness/src/image-video/service.ts @@ -6,6 +6,7 @@ import { HarnessError } from "../errors/harness-error.js"; import { ImageVideoDraftSchema, recordSchema, taskIdSchema, validationSchema, type ImageVideoDraft, type ImageVideoPublicTask, type ImageVideoRecord, type ImageVideoRemote, type ImageVideoValidation } from "./contracts.js"; const locks = new Map>(); +const submissionReceiptSchema = recordSchema.pick({ version: true, id: true, revision: true, requestHash: true, jobId: true }).required().strict(); export const sha256 = (bytes: Buffer | string) => createHash("sha256").update(bytes).digest("hex"); const fail = (code: string, message: string): never => { throw new HarnessError(code, message); }; export type ImageVideoOptions = { root: string; attachments: Pick; remote: ImageVideoRemote; collect: (url: string, directory: string, duration: number) => Promise }; @@ -57,8 +58,8 @@ export class ImageVideoService { async submit(sessionId: string, id: string): Promise { return this.lock(id, async () => { const r = await this.read(id); this.owner(r, sessionId); - if (r.jobId) return this.public(r); if (r.state === "unknown_submission") return fail("I2V_UNKNOWN_SUBMISSION", "Submission outcome unknown; no safe retry or new billing request is allowed"); + if (r.jobId) return this.public(r); if (r.state !== "awaiting_submission" || !r.assetId) return fail("I2V_NOT_READY", "Reference import must be ready before generation approval"); await this.options.remote.preflight(); await this.reference(r); r.state = "submitting"; await this.save(r); @@ -76,7 +77,7 @@ export class ImageVideoService { async refresh(id: string): Promise { return this.lock(id, async () => { const r = await this.read(id); - if (r.state === "ready" || r.state === "failed" || Date.now() - (r.lastPollAt ?? 0) < 30_000) return this.public(r); + if (r.state === "unknown_submission" || r.state === "ready" || r.state === "failed" || Date.now() - (r.lastPollAt ?? 0) < 30_000) return this.public(r); if (!r.jobId && !(r.upload && ["importing", "unknown_upload"].includes(r.state))) return this.public(r); r.lastPollAt = Date.now(); await this.save(r); try { @@ -118,6 +119,15 @@ export class ImageVideoService { } private async reference(r: ImageVideoRecord) { await this.regular(join(this.root, r.id, "reference")); const bytes = await readFile(join(this.root, r.id, "reference")); if (sha256(bytes) !== r.referenceHash) fail("I2V_MEDIA_INVALID", "Reference changed after draft"); return bytes; } private async regular(path: string) { const info = await lstat(path); if (!info.isFile() || info.isSymbolicLink()) fail("I2V_RECORD_INVALID", "Unsafe registered file"); } + private async matchingSubmissionReceipt(r: ImageVideoRecord): Promise { + const path = join(this.root, r.id, "submission-receipt.json"); + try { + await this.regular(path); + const receipt = submissionReceiptSchema.parse(JSON.parse(await readFile(path, "utf8"))); + if (receipt.id !== r.id || receipt.requestHash !== r.requestHash || receipt.revision !== r.revision || !receipt.jobId.trim() || (r.jobId !== undefined && r.jobId !== receipt.jobId)) return undefined; + return receipt.jobId; + } catch { return undefined; } // Preserve absent, malformed and conflicting evidence for manual reconciliation. + } private async read(id: string): Promise { if (!taskIdSchema.safeParse(id).success) return fail("I2V_TASK_NOT_FOUND", "Task not found"); const path = join(this.root, id, "record.json"); @@ -126,7 +136,11 @@ export class ImageVideoService { try { await this.regular(path); const r = recordSchema.parse(JSON.parse(await readFile(path, "utf8"))); if (r.id !== id || r.requestHash !== sha256(JSON.stringify(r.draft))) throw new Error("Binding mismatch"); - if (r.state === "submitting") { r.state = "unknown_submission"; await this.save(r); } + if (r.state === "submitting") { + const jobId = await this.matchingSubmissionReceipt(r); + if (jobId !== undefined) { r.jobId = jobId; r.state = "queued"; delete r.lastPollAt; delete r.error; await this.save(r); } + else { r.state = "unknown_submission"; await this.save(r); } + } if (["initializing_upload", "uploading", "finalizing"].includes(r.state)) { r.state = "unknown_upload"; await this.save(r); } return r; } catch { return fail("I2V_RECORD_INVALID", "Task record is invalid; manual recovery required"); } diff --git a/packages/harness/tests/image-video.test.ts b/packages/harness/tests/image-video.test.ts index eb603e2..60c3950 100644 --- a/packages/harness/tests/image-video.test.ts +++ b/packages/harness/tests/image-video.test.ts @@ -95,4 +95,46 @@ describe("durable image video tasks", () => { const receipt = JSON.parse(await readFile(join(f.options.root, task.id, "submission-receipt.json"), "utf8")) as { jobId: string }; expect(receipt.jobId).toBe("late_job"); }); + it("recovers a matching persisted submit receipt after restart and polls that job without resubmitting", async () => { + const f = await fixture(); const task = await f.service.create("ses_owner", f.draft); + await f.service.upload("ses_owner", task.id); await f.service.refresh(task.id); + const file = join(f.options.root, task.id, "record.json"); const record = JSON.parse(await readFile(file, "utf8")) as ImageVideoRecord; + record.state = "submitting"; record.revision++; + await writeFile(file, JSON.stringify(record)); + await writeFile(join(f.options.root, task.id, "submission-receipt.json"), JSON.stringify({ version: 1, id: task.id, requestHash: record.requestHash, revision: record.revision, jobId: "accepted_before_crash" })); + f.calls.length = 0; + f.remote.status = (r) => { f.calls.push(`status:${r.jobId ?? "missing"}`); return Promise.resolve({ state: "running" }); }; + const restored = new harness.ImageVideoService(f.options); + expect(await restored.view(task.id)).toMatchObject({ state: "queued", jobId: "accepted_before_crash" }); + expect(JSON.parse(await readFile(file, "utf8")) as ImageVideoRecord).toMatchObject({ state: "queued", jobId: "accepted_before_crash", revision: record.revision + 1 }); + expect(await restored.refresh(task.id)).toMatchObject({ state: "running", jobId: "accepted_before_crash" }); + await new harness.ImageVideoService(f.options).submit("ses_owner", task.id); + expect(f.calls).toEqual(["status:accepted_before_crash"]); + }); + it.each(["task", "hash", "stale", "future", "malformed", "empty_job", "version", "extra_field", "conflicting_job"] as const)("quarantines a %s receipt without polling it or submitting again", async (kind) => { + const f = await fixture(); const task = await f.service.create("ses_owner", f.draft); + await f.service.upload("ses_owner", task.id); await f.service.refresh(task.id); + const file = join(f.options.root, task.id, "record.json"); const record = JSON.parse(await readFile(file, "utf8")) as ImageVideoRecord; + record.state = "submitting"; record.revision++; + delete record.lastPollAt; + if (kind === "conflicting_job") record.jobId = "different_record_job"; + const receipt: Record = { version: 1, id: task.id, requestHash: record.requestHash, revision: record.revision, jobId: "untrusted_receipt_job" }; + if (kind === "task") receipt.id = "i2v_11111111-1111-1111-1111-111111111111"; + if (kind === "hash") receipt.requestHash = "a".repeat(64); + if (kind === "stale") receipt.revision = record.revision - 1; + if (kind === "future") receipt.revision = record.revision + 1; + if (kind === "empty_job") receipt.jobId = ""; + if (kind === "version") receipt.version = 2; + if (kind === "extra_field") receipt.unexpected = "conflicting format"; + const serialized = kind === "malformed" ? "{broken-json" : JSON.stringify(receipt); + await writeFile(file, JSON.stringify(record)); + const receiptPath = join(f.options.root, task.id, "submission-receipt.json"); await writeFile(receiptPath, serialized); + f.calls.length = 0; + const restored = new harness.ImageVideoService(f.options); + expect((await restored.view(task.id)).state).toBe("unknown_submission"); + expect((await restored.refresh(task.id)).state).toBe("unknown_submission"); + await expect(restored.submit("ses_owner", task.id)).rejects.toMatchObject({ code: "I2V_UNKNOWN_SUBMISSION" }); + expect(f.calls).toEqual([]); + expect(await readFile(receiptPath, "utf8")).toBe(serialized); + }); });