import type { TaskRunView } from "@umt/tasks"; import { createTempTree, type TempTree } from "@umt/testkit"; import { sql } from "kysely"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { enqueueSweep, enqueueTracked } from "../../src/modules/acquisition-organize/services/common.ts"; import { type QueueScrapeInput, queueScrape, runScrapeRecord } from "../../src/modules/acquisition-organize/services/scrape.ts"; import { createHarness, type Harness, pgAvailable, seedBase, seedEntity } from "./harness.ts"; const hasPg = await pgAvailable(); const runView = (runId: string, input: unknown, status: TaskRunView["status"] = "pending"): TaskRunView => ({ runId, taskName: "scrapeMedia", version: 2, status, attempts: 0, idempotencyKey: null, traceId: null, priority: 0, input, output: null, error: null, cancelRequestedAt: null, cancelReason: null, availableAt: null, createdAt: new Date(), startedAt: null, finishedAt: null, }); describe.skipIf(!hasPg)("入队关联与巡检恢复", () => { let h: Harness; let lib: TempTree; let rootId: string; let entityId: string; beforeAll(async () => { lib = await createTempTree({}, { sentinel: true }); h = await createHarness("enqueue_recovery"); ({ rootId } = await seedBase(h.db, { libraryPath: lib.root, cachePath: lib.root, qbUrl: "http://127.0.0.1:9" })); entityId = await seedEntity(h.db, { kind: "movie", title: "恢复测试", year: 2020 }); await sql`create function reject_association() returns trigger language plpgsql as $$ begin if new.engine_run_id is not null and new.engine_run_id is distinct from old.engine_run_id then raise exception 'injected association failure'; end if; return new; end $$`.execute(h.db); }); async function rejectAssociation(table: "task" | "scrape_record") { await sql .raw(`create trigger reject_association before update on ${table} for each row execute function reject_association()`) .execute(h.db); } async function allowAssociation() { for (const table of ["task", "scrape_record"]) await sql.raw(`drop trigger if exists reject_association on ${table}`).execute(h.db); } afterEach(async () => { vi.restoreAllMocks(); await allowAssociation(); }); afterAll(async () => { await h?.close(); await lib?.cleanup(); }); const task = (key: string) => ({ kind: "reconcileLibrary" as const, idempotencyKey: key, input: {} }); const scrape = (path: string, idempotencyKey?: string): QueueScrapeInput => ({ rootId: `lib:${rootId}`, rootPath: lib.root, libraryRootId: rootId, path, match: { mediaEntityId: entityId, provider: null, externalId: null, kind: "movie", title: "恢复测试", year: 2020, source: "chosen", }, trigger: "manual", mode: "missing", ...(idempotencyKey ? { idempotencyKey } : {}), }); it("引擎查询异常跳过本轮,下一轮仍查原任务", async () => { const t = task("query_error"); const q = await enqueueTracked(h.d, t); const enqueue = vi.spyOn(h.engine, "enqueue"); const get = vi .spyOn(h.engine, "get") .mockRejectedValueOnce(new Error("temporary engine outage")) .mockResolvedValue(runView(q.runId, {}, "running")); expect(await enqueueSweep(h.d, t, "next")).toBeNull(); expect(await enqueueSweep(h.d, t, "later")).toBeNull(); expect(enqueue).not.toHaveBeenCalled(); expect(get).toHaveBeenNthCalledWith(2, q.runId); expect( (await h.db.selectFrom("task").select("engine_run_id").where("id", "=", q.taskId).executeTakeFirstOrThrow()) .engine_run_id, ).toBe(q.runId); }); it.each(["missing", "completed"])("明确 %s 时允许补投", async (state) => { const t = task(`retry_${state}`); const q = await enqueueTracked(h.d, t); vi.spyOn(h.engine, "get").mockResolvedValue(state === "missing" ? null : runView(q.runId, {}, "completed")); const enqueue = vi.spyOn(h.engine, "enqueue"); expect(await enqueueSweep(h.d, t, "next")).not.toBeNull(); expect(enqueue).toHaveBeenCalledOnce(); expect(enqueue.mock.calls[0]?.[2].idempotencyKey).toBe(`${t.kind}:${t.idempotencyKey}:next`); }); it.each([false, true])("关联写回失败后巡检按原 taskId 找回已接受的 run (旧关联=%s)", async (oldAssociation) => { const t = task(`association_${oldAssociation}`); if (oldAssociation) await enqueueTracked(h.d, t); await rejectAssociation("task"); const q = await enqueueTracked(h.d, t, { engineKey: "new-slot" }); await h.db .insertInto("tasks_run_meta") .values({ run_id: q.runId, task_name: t.kind, task_version: 1, idempotency_key: "new-slot", trace_id: q.taskId }) .execute(); await allowAssociation(); vi.spyOn(h.engine, "get").mockResolvedValue(runView(q.runId, {}, "running")); const enqueue = vi.spyOn(h.engine, "enqueue"); expect(await enqueueSweep(h.d, t, "later")).toBeNull(); expect(enqueue).not.toHaveBeenCalled(); expect( (await h.db.selectFrom("task").select("engine_run_id").where("id", "=", q.taskId).executeTakeFirstOrThrow()) .engine_run_id, ).toBe(q.runId); }); it("刮削已接受而关联失败时保持 queued,重试复用记录与原幂等键", async () => { const input = scrape("association-retry"); await rejectAssociation("scrape_record"); const q = await queueScrape(h.d, input); expect( await h.db .selectFrom("scrape_record") .select(["status", "reason", "engine_run_id"]) .where("id", "=", q.recordId) .executeTakeFirstOrThrow(), ).toMatchObject({ status: "queued", reason: null, engine_run_id: null }); const key = h.engine.enqueued.at(-1)?.key; await allowAssociation(); const enqueue = vi.spyOn(h.engine, "enqueue").mockResolvedValue({ runId: q.taskId!, deduplicated: true }); vi.spyOn(h.engine, "get").mockResolvedValue(runView(q.taskId!, { recordId: q.recordId })); expect(await queueScrape(h.d, input)).toEqual({ recordId: q.recordId, taskId: q.taskId, deduplicated: true }); expect(enqueue.mock.calls[0]?.[2].idempotencyKey).toBe(key); expect((await h.db.selectFrom("scrape_record").select("id").where("path", "=", input.path).execute()).length).toBe( 1, ); }); it("刮削处理器开始时补关联,不会把已接受的工作误标为入队失败", async () => { await rejectAssociation("scrape_record"); const q = await queueScrape(h.d, scrape("worker-backfill")); await allowAssociation(); await h.db.updateTable("library_root").set({ management_mode: "register_only" }).where("id", "=", rootId).execute(); const r = await runScrapeRecord(h.d, q.recordId, undefined, q.taskId!); expect(r.reason).toBe("scrape_root_read_only"); expect( ( await h.db .selectFrom("scrape_record") .select("engine_run_id") .where("id", "=", q.recordId) .executeTakeFirstOrThrow() ).engine_run_id, ).toBe(q.taskId); }); it("确实入队失败的新刮削记录标失败", async () => { vi.spyOn(h.engine, "enqueue").mockRejectedValue(new Error("enqueue rejected")); const input = scrape("enqueue-failed"); await expect(queueScrape(h.d, input)).rejects.toThrow("enqueue rejected"); expect( await h.db .selectFrom("scrape_record") .select(["status", "reason"]) .where("path", "=", input.path) .executeTakeFirstOrThrow(), ).toMatchObject({ status: "failed", reason: "task_enqueue_failed" }); }); it("自动重放从原 run 输入找回未关联记录,只删除本次多建的记录", async () => { const input = scrape("auto-replay", "auto-stable-key"); await rejectAssociation("scrape_record"); const q = await queueScrape(h.d, input); await allowAssociation(); await h.db.updateTable("scrape_record").set({ status: "succeeded" }).where("id", "=", q.recordId).execute(); vi.spyOn(h.engine, "enqueue").mockResolvedValue({ runId: q.taskId!, deduplicated: true }); vi.spyOn(h.engine, "get").mockResolvedValue(runView(q.taskId!, { recordId: q.recordId }, "completed")); expect(await queueScrape(h.d, input)).toEqual({ recordId: q.recordId, taskId: q.taskId, deduplicated: true }); const recs = await h.db .selectFrom("scrape_record") .select(["id", "status", "engine_run_id"]) .where("path", "=", input.path) .execute(); expect(recs).toEqual([{ id: q.recordId, status: "succeeded", engine_run_id: q.taskId }]); }); });