407 lines
16 KiB
JavaScript
407 lines
16 KiB
JavaScript
// Local disposable PostgreSQL only. Install embedded-postgres into .qa/video-task-db first.
|
|
const fs = require("fs"),
|
|
path = require("path"),
|
|
assert = require("node:assert/strict");
|
|
(async () => {
|
|
const { default: EmbeddedPostgres } = await import(
|
|
path.resolve(".qa/video-task-db/node_modules/embedded-postgres/dist/index.js")
|
|
);
|
|
const pg = new EmbeddedPostgres({
|
|
databaseDir: path.resolve(".qa/video-task-db/data"),
|
|
port: 55439,
|
|
user: "canvas_qa",
|
|
password: "local-disposable-only",
|
|
persistent: false,
|
|
onLog: (message) => console.log(message),
|
|
onError: (message) => console.error(message),
|
|
});
|
|
let client;
|
|
try {
|
|
await pg.initialise();
|
|
await pg.start();
|
|
client = pg.getPgClient();
|
|
await client.connect();
|
|
process.env.POSTGRES_URL =
|
|
"postgres://canvas_qa:local-disposable-only@127.0.0.1:55439/postgres";
|
|
await client.query(
|
|
'CREATE TABLE "User" (id uuid PRIMARY KEY); CREATE TABLE "CanvasProject" (id uuid PRIMARY KEY, "userId" uuid);',
|
|
);
|
|
await client.query(
|
|
fs.readFileSync("lib/db/migrations/0091_add_canvas_capability_tasks.sql", "utf8"),
|
|
);
|
|
const esbuild = require(
|
|
path.resolve(
|
|
"node_modules/.pnpm",
|
|
fs.readdirSync("node_modules/.pnpm").find((name) => name.startsWith("esbuild@0.25.")),
|
|
"node_modules/esbuild",
|
|
),
|
|
);
|
|
await esbuild.build({
|
|
entryPoints: ["lib/video-canvas/task-store.ts"],
|
|
outfile: ".qa/video-task-db/store.cjs",
|
|
bundle: true,
|
|
platform: "node",
|
|
format: "cjs",
|
|
packages: "external",
|
|
alias: { "@": process.cwd() },
|
|
plugins: [
|
|
{
|
|
name: "server-only",
|
|
setup(build) {
|
|
build.onResolve({ filter: /^server-only$/ }, () => ({
|
|
path: "empty",
|
|
namespace: "qa",
|
|
}));
|
|
build.onLoad({ filter: /.*/, namespace: "qa" }, () => ({ contents: "" }));
|
|
},
|
|
},
|
|
],
|
|
});
|
|
const store = require(path.resolve(".qa/video-task-db/store.cjs"));
|
|
const user = crypto.randomUUID(),
|
|
other = crypto.randomUUID(),
|
|
project = crypto.randomUUID();
|
|
await client.query('INSERT INTO "User" VALUES ($1),($2)', [user, other]);
|
|
await client.query('INSERT INTO "CanvasProject" VALUES ($1,$2)', [project, user]);
|
|
assert.equal(await store.ownsTaskProject(user, project), true);
|
|
assert.equal(await store.ownsTaskProject(other, project), false);
|
|
const request = {
|
|
requestId: crypto.randomUUID(),
|
|
projectId: project,
|
|
kind: "video",
|
|
params: { mode: "text-to-video", prompt: "QA" },
|
|
node: {
|
|
id: "node-a",
|
|
title: "QA",
|
|
kind: "media",
|
|
inputType: "video",
|
|
x: 0,
|
|
y: 0,
|
|
width: 300,
|
|
height: 200,
|
|
},
|
|
edges: [],
|
|
};
|
|
const copies = await Promise.all(
|
|
Array.from({ length: 10 }, () => store.createCanvasGenerationTask(user, request)),
|
|
);
|
|
assert.equal(new Set(copies.map((task) => task.id)).size, 1);
|
|
const task = copies[0];
|
|
await assert.rejects(
|
|
store.createCanvasGenerationTask(user, {
|
|
...request,
|
|
params: { ...request.params, prompt: "different" },
|
|
}),
|
|
);
|
|
const claims = await Promise.all(
|
|
Array.from({ length: 10 }, () => store.claimCanvasGenerationTask(user, project)),
|
|
);
|
|
assert.equal(claims.filter(Boolean).length, 1);
|
|
assert.equal((await store.cancelCanvasGenerationTask(user, project, task.id)).ok, false);
|
|
assert.equal((await store.cancelCanvasGenerationTask(other, project, task.id)).task, null);
|
|
const pending = await store.createCanvasGenerationTask(user, {
|
|
...request,
|
|
requestId: crypto.randomUUID(),
|
|
node: { ...request.node, id: "node-b" },
|
|
});
|
|
assert.equal(await store.claimCanvasGenerationTask(user, project), null);
|
|
assert.equal((await store.cancelCanvasGenerationTask(user, project, pending.id)).ok, true);
|
|
await store.finishCanvasGenerationTask(task.id, { url: "https://example.com/qa.mp4" });
|
|
assert.equal(
|
|
(await store.listCanvasGenerationTasks(user, project)).find((t) => t.id === task.id).result
|
|
.url,
|
|
"https://example.com/qa.mp4",
|
|
);
|
|
assert.equal((await store.listCanvasGenerationTasks(other, project)).length, 0);
|
|
assert.equal(await store.hideCanvasGenerationNode(user, project, task.id, "node-a"), true);
|
|
assert.deepEqual(
|
|
(await store.listCanvasGenerationTasks(user, project)).find((t) => t.id === task.id).request
|
|
.hiddenNodeIds,
|
|
["node-a"],
|
|
);
|
|
const stale = await store.createCanvasGenerationTask(user, {
|
|
...request,
|
|
requestId: crypto.randomUUID(),
|
|
node: { ...request.node, id: "node-c" },
|
|
});
|
|
await store.claimCanvasGenerationTask(user, project);
|
|
await client.query(
|
|
"UPDATE \"CanvasCapabilityTask\" SET \"updatedAt\"=(now() at time zone 'UTC')-interval '3 minutes' WHERE id=$1",
|
|
[stale.id],
|
|
);
|
|
assert.equal(
|
|
(await store.listCanvasGenerationTasks(user, project)).find((t) => t.id === stale.id).status,
|
|
"interrupted",
|
|
);
|
|
assert.equal(await store.claimCanvasGenerationTask(user, project), null);
|
|
await store.checkpointCanvasGenerationTask(stale.id, {
|
|
upstreamTask: { taskId: "ignored-after-interruption" },
|
|
});
|
|
assert.equal(
|
|
(await store.listCanvasGenerationTasks(user, project)).find((t) => t.id === stale.id).result,
|
|
null,
|
|
);
|
|
await store.finishCanvasGenerationTask(stale.id, { url: "https://example.com/late.mp4" });
|
|
assert.equal(
|
|
(await store.listCanvasGenerationTasks(user, project)).find((t) => t.id === stale.id).status,
|
|
"succeeded",
|
|
);
|
|
const upstream = await store.createCanvasGenerationTask(user, {
|
|
...request,
|
|
requestId: crypto.randomUUID(),
|
|
node: { ...request.node, id: "node-upstream" },
|
|
});
|
|
await store.claimCanvasGenerationTask(user, project);
|
|
await store.checkpointCanvasGenerationTask(upstream.id, {
|
|
upstreamTask: { operation: "enhance", taskId: "provider-123" },
|
|
});
|
|
await store.interruptCanvasGenerationTask(upstream.id, "查询连接中断");
|
|
const recovered = (await store.listCanvasGenerationTasks(user, project)).find(
|
|
(t) => t.id === upstream.id,
|
|
);
|
|
assert.equal(recovered.status, "interrupted");
|
|
assert.equal(recovered.result.upstreamTask.taskId, "provider-123");
|
|
assert.equal(await store.claimCanvasGenerationTask(user, project), null);
|
|
// Reclaim one checkpoint under concurrency and reject all stale worker writes.
|
|
const enhanceRequest = {
|
|
...request,
|
|
requestId: crypto.randomUUID(),
|
|
kind: "image-operation",
|
|
node: { ...request.node, id: "node-recoverable" },
|
|
params: { operation: "enhance", imageUrl: "https://example.com/source.png" },
|
|
};
|
|
const resumable = await store.createCanvasGenerationTask(user, enhanceRequest);
|
|
const firstLease = await store.claimCanvasGenerationTask(user, project);
|
|
await store.checkpointCanvasGenerationTask(
|
|
resumable.id,
|
|
{
|
|
upstreamTask: {
|
|
operation: "enhance",
|
|
taskId: "provider-resume",
|
|
sourceImageUrl: enhanceRequest.params.imageUrl,
|
|
},
|
|
},
|
|
firstLease.attempts,
|
|
);
|
|
await store.interruptCanvasGenerationTask(resumable.id, "query lost", firstLease.attempts);
|
|
assert.equal(
|
|
(
|
|
await store.createCanvasGenerationTask(user, {
|
|
...enhanceRequest,
|
|
requestId: crypto.randomUUID(),
|
|
})
|
|
).id,
|
|
resumable.id,
|
|
);
|
|
assert.equal(await store.claimCanvasGenerationTask(user, project), null);
|
|
await client.query(
|
|
"UPDATE \"CanvasCapabilityTask\" SET \"updatedAt\"=(now() at time zone 'UTC')-interval '6 minutes' WHERE id=$1",
|
|
[resumable.id],
|
|
);
|
|
const resumed = (
|
|
await Promise.all(
|
|
Array.from({ length: 10 }, () => store.claimCanvasGenerationTask(user, project)),
|
|
)
|
|
).filter(Boolean);
|
|
assert.equal(resumed.length, 1);
|
|
assert.equal(resumed[0].attempts, 2);
|
|
assert.equal(await store.ownsCanvasTaskLease(resumable.id, 1), false);
|
|
assert.equal(await store.ownsCanvasTaskLease(resumable.id, 2), true);
|
|
await store.finishCanvasGenerationTask(
|
|
resumable.id,
|
|
{ images: [{ originalUrl: "stale" }] },
|
|
undefined,
|
|
1,
|
|
);
|
|
await store.interruptCanvasGenerationTask(resumable.id, "stale failure", 1);
|
|
await store.checkpointCanvasGenerationTask(
|
|
resumable.id,
|
|
{ upstreamTask: { taskId: "wrong" } },
|
|
1,
|
|
);
|
|
const fenced = (await store.listCanvasGenerationTasks(user, project)).find(
|
|
(t) => t.id === resumable.id,
|
|
);
|
|
assert.equal(fenced.status, "running");
|
|
assert.equal(fenced.result.upstreamTask.taskId, "provider-resume");
|
|
await store.finishCanvasGenerationTask(
|
|
resumable.id,
|
|
{ images: [{ originalUrl: "https://example.com/recovered.png" }] },
|
|
undefined,
|
|
2,
|
|
);
|
|
|
|
// Video checkpoints use the same durable lease and original provider identity.
|
|
const videoRequest = {
|
|
...request,
|
|
requestId: crypto.randomUUID(),
|
|
kind: "video",
|
|
node: { ...request.node, id: "video-recoverable" },
|
|
params: { prompt: "QA only", preferredModelRef: "video-model" },
|
|
};
|
|
const videoTask = await store.createCanvasGenerationTask(user, videoRequest);
|
|
const videoLease = await store.claimCanvasGenerationTask(user, project);
|
|
const videoCheckpoint = {
|
|
operation: "video",
|
|
provider: "aliyun",
|
|
taskId: "original-video",
|
|
modelRef: "video-model",
|
|
modelId: "wan-qa",
|
|
providerId: "provider-qa",
|
|
baseUrl: "https://provider.invalid",
|
|
};
|
|
await store.checkpointCanvasGenerationTask(
|
|
videoTask.id,
|
|
{ upstreamTask: videoCheckpoint },
|
|
videoLease.attempts,
|
|
);
|
|
await store.interruptCanvasGenerationTask(
|
|
videoTask.id,
|
|
"query disconnected",
|
|
videoLease.attempts,
|
|
);
|
|
assert.equal(
|
|
(
|
|
await store.createCanvasGenerationTask(user, {
|
|
...videoRequest,
|
|
requestId: crypto.randomUUID(),
|
|
})
|
|
).id,
|
|
videoTask.id,
|
|
);
|
|
await client.query(
|
|
`UPDATE "CanvasCapabilityTask" SET "updatedAt"=(now() at time zone 'UTC')-interval '6 minutes' WHERE id=$1`,
|
|
[videoTask.id],
|
|
);
|
|
const videoReclaim = (
|
|
await Promise.all(
|
|
Array.from({ length: 10 }, () => store.claimCanvasGenerationTask(user, project)),
|
|
)
|
|
).filter(Boolean);
|
|
assert.equal(videoReclaim.length, 1);
|
|
assert.deepEqual(videoReclaim[0].result.upstreamTask, videoCheckpoint);
|
|
await store.finishCanvasGenerationTask(
|
|
videoTask.id,
|
|
{ videoUrl: "https://output.invalid/video.mp4" },
|
|
undefined,
|
|
videoReclaim[0].attempts,
|
|
);
|
|
console.log(
|
|
"PASS: video checkpoint recovery, ten concurrent reclaimers, original upstream identity and duplicate prevention",
|
|
);
|
|
|
|
// A separate worker process exits after its first GET. Restart must read the persisted ID.
|
|
const callsFile = path.resolve(".qa/video-task-db/provider-calls.jsonl");
|
|
fs.writeFileSync(callsFile, "");
|
|
const unexpected = `throw Error("unexpected provider path")`;
|
|
const stubs = {
|
|
"./smart-edit/execute": `export const executeSmartEdit=async()=>{${unexpected}};`,
|
|
"./audio-generation": `export const executeCanvasAudio=async()=>{${unexpected}};`,
|
|
"@/lib/ai/tools/generate-video": `export const executeDirectVideoGeneration=async()=>{${unexpected}};`,
|
|
"@/lib/canvas/generate-image-response": `export const executeCanvasImageResponse=async()=>{${unexpected}};`,
|
|
"./workflow-analysis-response": `export const executeWorkflowAnalysisResponse=async()=>{${unexpected}};`,
|
|
"@/lib/canvas/operations/extend": `export const executeExtendPOST=async()=>{${unexpected}};`,
|
|
"@/lib/canvas/operations/rmbg": `export const executeRemoveBackgroundPOST=async()=>{${unexpected}};`,
|
|
"@/lib/canvas/operations/angle-edit": `export const executeAngleEditPOST=async()=>{${unexpected}};`,
|
|
"@/lib/canvas/operations/enhance": `
|
|
import fs from 'node:fs';
|
|
const log=(method,id)=>fs.appendFileSync(${JSON.stringify(callsFile)},JSON.stringify({method,id})+'\\n');
|
|
export const executeEnhancePOST=async(req,user,onCreated)=>{log('POST','durable-upstream');await onCreated?.('durable-upstream');return Response.json({processing:true,taskId:'durable-upstream'},{status:202});};
|
|
export const executeEnhanceGET=async req=>{const id=new URL(req.url).searchParams.get('taskId');log('GET',id);if(process.env.QA_CRASH==='1')process.exit(17);return Response.json({success:true,original_url:'https://example.com/process-recovered.png'});};`,
|
|
"server-only": "",
|
|
};
|
|
await esbuild.build({
|
|
entryPoints: ["lib/video-canvas/task-worker.ts"],
|
|
outfile: ".qa/video-task-db/worker.cjs",
|
|
bundle: true,
|
|
platform: "node",
|
|
format: "cjs",
|
|
packages: "external",
|
|
alias: { "@": process.cwd() },
|
|
plugins: [
|
|
{
|
|
name: "mock-provider-only",
|
|
setup(build) {
|
|
build.onResolve({ filter: /.*/ }, (args) =>
|
|
stubs[args.path] !== undefined ? { path: args.path, namespace: "qa" } : undefined,
|
|
);
|
|
build.onLoad({ filter: /.*/, namespace: "qa" }, (args) => ({
|
|
contents: stubs[args.path],
|
|
loader: "js",
|
|
}));
|
|
},
|
|
},
|
|
],
|
|
});
|
|
const restartTask = await store.createCanvasGenerationTask(user, {
|
|
...enhanceRequest,
|
|
requestId: crypto.randomUUID(),
|
|
node: { ...enhanceRequest.node, id: "node-process-restart" },
|
|
});
|
|
const childCode = `require(${JSON.stringify(path.resolve(".qa/video-task-db/worker.cjs"))}).runCanvasGenerationQueue(${JSON.stringify(user)},${JSON.stringify(project)}).then(()=>process.exit(0),e=>{console.error(e);process.exit(1)});`;
|
|
const runChild = (crash) =>
|
|
new Promise((resolve, reject) => {
|
|
require("node:child_process").execFile(
|
|
process.execPath,
|
|
["-e", childCode],
|
|
{ env: { ...process.env, QA_CRASH: crash ? "1" : "0" }, timeout: 30000 },
|
|
(error, stdout, stderr) => {
|
|
if (error && !(crash && error.code === 17)) reject(Error(stderr || String(error)));
|
|
else resolve();
|
|
},
|
|
);
|
|
});
|
|
await runChild(true);
|
|
const crashed = (await store.listCanvasGenerationTasks(user, project)).find(
|
|
(t) => t.id === restartTask.id,
|
|
);
|
|
assert.equal(crashed.status, "running");
|
|
assert.equal(crashed.result.upstreamTask.taskId, "durable-upstream");
|
|
await client.query(
|
|
"UPDATE \"CanvasCapabilityTask\" SET \"updatedAt\"=(now() at time zone 'UTC')-interval '6 minutes' WHERE id=$1",
|
|
[restartTask.id],
|
|
);
|
|
assert.equal(
|
|
(await store.listCanvasGenerationTasks(user, project)).find((t) => t.id === restartTask.id)
|
|
.status,
|
|
"interrupted",
|
|
);
|
|
await client.query(
|
|
"UPDATE \"CanvasCapabilityTask\" SET \"updatedAt\"=(now() at time zone 'UTC')-interval '6 minutes' WHERE id=$1",
|
|
[restartTask.id],
|
|
);
|
|
await runChild(false);
|
|
const final = (await store.listCanvasGenerationTasks(user, project)).find(
|
|
(t) => t.id === restartTask.id,
|
|
);
|
|
assert.equal(final.status, "succeeded");
|
|
assert.equal(final.attempts, 2);
|
|
assert.equal(final.result.images[0].originalUrl, "https://example.com/process-recovered.png");
|
|
const providerCalls = fs
|
|
.readFileSync(callsFile, "utf8")
|
|
.trim()
|
|
.split("\n")
|
|
.map((line) => JSON.parse(line));
|
|
assert.deepEqual(
|
|
providerCalls.map((call) => call.method),
|
|
["POST", "GET", "GET"],
|
|
);
|
|
assert(providerCalls.every((call) => call.id === "durable-upstream"));
|
|
console.log(
|
|
"PASS: real PostgreSQL checkpoint recovery, ten concurrent reclaimers, retry cooldown, stale-worker fencing, duplicate prevention, abrupt worker process restart with POST/GET/GET only.",
|
|
);
|
|
console.log(
|
|
"PASS: real PostgreSQL ownership, idempotency (10 submits), exclusive claiming (10 workers), serial queue, cancellation CAS, durable results, hidden nodes, interrupted task without resubmission, late-result recovery.",
|
|
);
|
|
} finally {
|
|
if (client) await client.end();
|
|
await pg.stop();
|
|
}
|
|
})().then(
|
|
() => process.exit(0),
|
|
(error) => {
|
|
console.error(error);
|
|
process.exit(1);
|
|
},
|
|
);
|