145 lines
10 KiB
TypeScript
145 lines
10 KiB
TypeScript
import { randomUUID } from "node:crypto";
|
|
import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
|
|
import { tmpdir } from "node:os";
|
|
import { join, resolve } from "node:path";
|
|
|
|
import { afterEach, describe, expect, it } from "vitest";
|
|
|
|
import { CreditService } from "../../apps/api/src/credits.js";
|
|
import { GenerationSubmissionService, StaticGenerationModelCatalog } from "../../apps/api/src/generation-submission.js";
|
|
import { ManagedStorage } from "../../apps/api/src/managed-storage.js";
|
|
import { ProjectService } from "../../apps/api/src/projects.js";
|
|
import { RegistrationService } from "../../apps/api/src/registration.js";
|
|
import { MockResendAdapter } from "../../apps/api/src/resend-adapter.js";
|
|
import type { GenerationAdapter, GenerationAdapterResult } from "../../apps/worker/src/ai-adapter-contract.js";
|
|
import { GenerationProcessor } from "../../apps/worker/src/generation-processor.js";
|
|
|
|
const now = Date.parse("2026-08-02T17:00:00.000Z");
|
|
const modelId = "gemini-3-pro-image-preview";
|
|
const png = Buffer.concat([Buffer.from([137, 80, 78, 71, 13, 10, 26, 10]), Buffer.alloc(96, 0x43)]);
|
|
const roots: string[] = [];
|
|
const closeables: Array<{ close(): void }> = [];
|
|
|
|
function evidence(file: string, value: unknown) {
|
|
const root = process.env.DADA_EVIDENCE_DIR_WORKER_RECOVERY;
|
|
if (!root) return;
|
|
const directory = resolve(root, "TDD-WP3-WRK-001-recovery");
|
|
mkdirSync(directory, { recursive: true });
|
|
writeFileSync(resolve(directory, file), `${JSON.stringify(value, null, 2)}\n`);
|
|
}
|
|
|
|
async function harness(adapter: GenerationAdapter) {
|
|
const dataRoot = mkdtempSync(join(tmpdir(), "dada-wp3-04-worker-"));
|
|
roots.push(dataRoot);
|
|
mkdirSync(join(dataRoot, "db"), { recursive: true });
|
|
const databasePath = join(dataRoot, "db", "dada.sqlite3");
|
|
const registration = new RegistrationService({
|
|
challengePepper: Buffer.alloc(32, 0x41), clock: () => now, currentPrivacyNoticeVersion: "p0a-registration-notice-v1",
|
|
databasePath, invitePepper: Buffer.alloc(32, 0x42), resend: new MockResendAdapter(), sessionPepper: Buffer.alloc(32, 0x43),
|
|
});
|
|
const projects = new ProjectService({ clock: () => now, databasePath });
|
|
const credits = new CreditService({ clock: () => now, databasePath });
|
|
const storage = new ManagedStorage({ dataRoot, databasePath });
|
|
const submissions = new GenerationSubmissionService({
|
|
clock: () => now, credits,
|
|
models: new StaticGenerationModelCatalog([{
|
|
configSetVersion: 1, configVersion: 1, contractValidationStatus: "verified", creditCost: 1, enabled: true, modelId,
|
|
promptMaxLength: 1_000, referenceLimits: { maxFileBytes: 1_024, maxFiles: 2, maxTotalBytes: 2_048 },
|
|
runtimeAvailability: { availableForNewJobs: true, reason: null }, supportedRatios: ["3:4"],
|
|
}]),
|
|
storage,
|
|
});
|
|
closeables.push(submissions, storage, credits, projects, registration);
|
|
const userId = randomUUID();
|
|
registration.database.prepare("INSERT INTO users (user_id, normalized_email, role, status, counts_toward_stage_limit, registration_id, created_at) VALUES (?, ?, 'user', 'active', 1, ?, ?)")
|
|
.run(userId, `${userId}@example.invalid`, randomUUID(), now);
|
|
registration.database.prepare("INSERT INTO user_profiles (user_id, creator_name, social_id, private_content_notice_version, private_content_notice_acknowledged_at) VALUES (?, 'Worker User', '@worker', NULL, NULL)")
|
|
.run(userId);
|
|
registration.database.prepare("INSERT INTO credit_accounts (user_id, available_balance, reserved_balance, updated_at) VALUES (?, 10, 0, ?)")
|
|
.run(userId, now);
|
|
const submitted = await submissions.submit({
|
|
clientSubmissionId: randomUUID(), confirmedCreditCost: 1, existingReferenceAssetIds: [],
|
|
idempotencyKey: `generation-${randomUUID()}-${randomUUID()}`, mode: "new_project", modelConfigVersion: 1,
|
|
modelId, newReferences: [], prompt: "Worker 崩溃恢复", ratio: "3:4", userId,
|
|
});
|
|
const processor = new GenerationProcessor({ adapter, clock: () => now + 1_000, dataRoot, databasePath, workerId: "worker-first" });
|
|
closeables.push(processor);
|
|
return { adapter, dataRoot, databasePath, generationId: submitted.task.generationId, processor, registration, submissions, userId };
|
|
}
|
|
|
|
function completed(): GenerationAdapterResult {
|
|
return { outputs: [{ bytes: png, mimeType: "image/png", pixelHeight: 1440, pixelWidth: 1080 }], status: "completed" };
|
|
}
|
|
|
|
afterEach(() => {
|
|
for (const value of closeables.splice(0).reverse()) value.close();
|
|
for (const root of roots.splice(0)) rmSync(root, { force: true, recursive: true });
|
|
});
|
|
|
|
describe("TDD-WP3-WRK-001-recovery", () => {
|
|
it("renews a lease, persists an async reference, and polls after restart without resubmitting", async () => {
|
|
let starts = 0;
|
|
let polls = 0;
|
|
const adapter: GenerationAdapter = {
|
|
async start() { starts += 1; return { status: "pending", upstreamJobReference: "upstream:async-1" }; },
|
|
async poll() { polls += 1; return completed(); },
|
|
};
|
|
const fixture = await harness(adapter);
|
|
expect(() => new GenerationProcessor({ adapter, clock: () => now + 1_000, dataRoot: fixture.dataRoot, databasePath: fixture.databasePath, workerId: "worker-second" }))
|
|
.toThrow("worker_instance_lock_conflict");
|
|
const pending = await fixture.processor.processNext();
|
|
expect(pending).toMatchObject({ generationId: fixture.generationId, status: "running" });
|
|
expect(fixture.registration.database.prepare("SELECT upstream_job_reference, lease_owner FROM generation_jobs WHERE generation_id = ?").get(fixture.generationId))
|
|
.toEqual({ lease_owner: "worker-first", upstream_job_reference: "upstream:async-1" });
|
|
const heartbeat = fixture.processor.heartbeat(fixture.generationId);
|
|
expect(heartbeat).toMatchObject({ heartbeat_at: now + 1_000, lease_expires_at: now + 31_000 });
|
|
fixture.registration.database.prepare("UPDATE generation_jobs SET lease_expires_at = ? WHERE generation_id = ?").run(now - 1, fixture.generationId);
|
|
fixture.processor.close();
|
|
closeables.splice(closeables.indexOf(fixture.processor), 1);
|
|
const restarted = new GenerationProcessor({ adapter, clock: () => now + 2_000, dataRoot: fixture.dataRoot, databasePath: fixture.databasePath, workerId: "worker-restart" });
|
|
closeables.push(restarted);
|
|
expect(fixture.registration.database.prepare("SELECT lease_owner, upstream_job_reference FROM generation_jobs WHERE generation_id = ?").get(fixture.generationId))
|
|
.toEqual({ lease_owner: "worker-restart", upstream_job_reference: "upstream:async-1" });
|
|
const result = await restarted.processGeneration(fixture.generationId);
|
|
expect(result).toMatchObject({ status: "succeeded" });
|
|
expect({ starts, polls }).toEqual({ polls: 1, starts: 1 });
|
|
evidence("worker-events.json", { starts, polls, recovery_action: "poll", upstream_job_reference: "upstream:async-1" });
|
|
evidence("db-diff.json", { lease_owner_after_recovery: "worker-restart", final_status: result.status, upstream_outcome_known: true, upstream_cost_reconciliation: "not_required" });
|
|
evidence("external-calls.json", { poll_calls: 1, start_calls: 1, duplicate_submit_calls: 0 });
|
|
});
|
|
|
|
it("fails an unknown result without a safe reference and releases exactly once for manual reconciliation", async () => {
|
|
const adapter: GenerationAdapter = { async start() { throw new Error("provider_timeout_without_reference"); } };
|
|
const fixture = await harness(adapter);
|
|
const first = await fixture.processor.processNext();
|
|
const replay = await fixture.processor.processGeneration(fixture.generationId);
|
|
expect(first).toMatchObject({ category: "unknown_retryable", status: "failed" });
|
|
expect(replay).toEqual(first);
|
|
expect(fixture.registration.database.prepare("SELECT upstream_job_reference, upstream_outcome_known, upstream_cost_reconciliation, lease_owner FROM generation_jobs WHERE generation_id = ?").get(fixture.generationId))
|
|
.toEqual({ lease_owner: null, upstream_job_reference: null, upstream_outcome_known: 0, upstream_cost_reconciliation: "pending_manual_review" });
|
|
expect(fixture.registration.database.prepare("SELECT COUNT(*) AS count FROM credit_ledger WHERE reference_id = ? AND entry_type = 'generation_release'").get(fixture.generationId)).toEqual({ count: 1 });
|
|
evidence("worker-events.json", { starts: 1, polls: 0, duplicate_submit_calls: 0, unknown_result: true });
|
|
evidence("db-diff.json", { final_status: "failed", category: "unknown_retryable", upstream_outcome_known: false, upstream_cost_reconciliation: "pending_manual_review", release_count: 1 });
|
|
evidence("external-calls.json", { start_calls: 1, poll_calls: 0, duplicate_submit_calls: 0 });
|
|
});
|
|
|
|
it("blocks new submissions while worker is degraded and reopens the gate when ready", async () => {
|
|
const fixture = await harness({ async start() { return completed(); } });
|
|
fixture.processor.setDegraded("lease_recovery_failed");
|
|
const degradedUserId = randomUUID();
|
|
fixture.registration.database.prepare("INSERT INTO users (user_id, normalized_email, role, status, counts_toward_stage_limit, registration_id, created_at) VALUES (?, ?, 'user', 'active', 1, ?, ?)")
|
|
.run(degradedUserId, `${degradedUserId}@example.invalid`, randomUUID(), now);
|
|
fixture.registration.database.prepare("INSERT INTO user_profiles (user_id, creator_name, social_id, private_content_notice_version, private_content_notice_acknowledged_at) VALUES (?, 'Degraded User', '@degraded', NULL, NULL)")
|
|
.run(degradedUserId);
|
|
fixture.registration.database.prepare("INSERT INTO credit_accounts (user_id, available_balance, reserved_balance, updated_at) VALUES (?, 10, 0, ?)")
|
|
.run(degradedUserId, now);
|
|
await expect(fixture.submissions.submit({
|
|
clientSubmissionId: randomUUID(), confirmedCreditCost: 1, existingReferenceAssetIds: [],
|
|
idempotencyKey: `generation-${randomUUID()}-${randomUUID()}`, mode: "new_project", modelConfigVersion: 1,
|
|
modelId, newReferences: [], prompt: "blocked while degraded", ratio: "3:4", userId: degradedUserId,
|
|
})).rejects.toMatchObject({ code: "generation_blocked", errorCategory: "gateway_contract_invalid" });
|
|
expect(fixture.processor.setReady()).toMatchObject({ status: "ready" });
|
|
expect(fixture.registration.database.prepare("SELECT status FROM worker_runtime_state WHERE singleton = 1").get()).toEqual({ status: "ready" });
|
|
});
|
|
});
|