From 70935f4c173e53c3d1daf7b10a7480ce34aecb3a Mon Sep 17 00:00:00 2001 From: root Date: Sat, 3 Oct 2026 20:59:06 +0800 Subject: [PATCH 1/2] =?UTF-8?q?=E6=95=B4=E7=90=86=EF=BC=9A=E8=A2=AB?= =?UTF-8?q?=E6=8E=92=E9=99=A4=E7=9A=84=E5=BE=85=E7=A1=AE=E8=AE=A4=E8=A7=86?= =?UTF-8?q?=E9=A2=91=E4=B8=8D=E7=AE=97=E6=88=90=E5=8A=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 季集待确认 / 没有季目录 / 命名失败而被排除的视频(excluded:episode_unresolved / no_target_dir / naming_failed)让整理记录从 succeeded 降为 partial;样例、过小、已在库等预期跳过不受影响。 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei --- packages/mapping/src/organize.ts | 36 +++++++++++++++++++++------ packages/mapping/test/convert.test.ts | 10 ++++++++ 2 files changed, 39 insertions(+), 7 deletions(-) diff --git a/packages/mapping/src/organize.ts b/packages/mapping/src/organize.ts index b4232de0..8a3a99e8 100644 --- a/packages/mapping/src/organize.ts +++ b/packages/mapping/src/organize.ts @@ -142,16 +142,38 @@ export function organizePlanToFileResults( return [...planned, ...excluded]; } -/** 逐文件状态 → 整理记录状态(全部完成后调用;仍有 queued/running 时为 running) */ -export function organizationStatusFromFiles(statuses: readonly FileResultStatus[]): OrganizationStatus { +/** + * 被排除但"本该入库"的视频:季集没定下来、没有对应季目录、命名缺字段。文件留在原处没人管, + * 不能算整理成功(Move 成功后会移除下载任务)。样例、过小、已在库等预期跳过不在此列。 + */ +export const UNRESOLVED_EXCLUSIONS = ["episode_unresolved", "no_target_dir", "naming_failed"] as const; +const UNRESOLVED_SKIP = new Set(UNRESOLVED_EXCLUSIONS.map((r) => `excluded:${r}`)); + +/** 文件结果的 skip_reason 是否属于"本该入库却被排除" */ +export const isUnresolvedExclusion = (skipReason: string | null | undefined): boolean => + !!skipReason && UNRESOLVED_SKIP.has(skipReason); + +/** + * 逐文件状态 → 整理记录状态(全部完成后调用;仍有 queued/running 时为 running)。 + * unresolved:被排除的待确认视频数(isUnresolvedExclusion),有则成功降为 partial。 + */ +export function organizationStatusFromFiles( + statuses: readonly FileResultStatus[], + opts: { unresolved?: number } = {}, +): OrganizationStatus { if (statuses.some((s) => s === "queued" || s === "running")) return "running"; const done = statuses.filter((s) => s !== "skipped"); - // 全部跳过(或没有文件)视为成功:没有需要做的事 - if (done.length === 0) return "succeeded"; const ok = done.filter((s) => s === "succeeded").length; - if (ok === done.length) return "succeeded"; - if (done.every((s) => s === "cancelled")) return "cancelled"; - return ok > 0 ? "partial" : "failed"; + // 全部跳过(或没有文件)视为成功:没有需要做的事 + const base: OrganizationStatus = + done.length === 0 || ok === done.length + ? "succeeded" + : done.every((s) => s === "cancelled") + ? "cancelled" + : ok > 0 + ? "partial" + : "failed"; + return base === "succeeded" && (opts.unresolved ?? 0) > 0 ? "partial" : base; } /** 整理计划 → organize.plan 能力输出(planId / payloadHash 由保存计划的一方给出) */ diff --git a/packages/mapping/test/convert.test.ts b/packages/mapping/test/convert.test.ts index 4b988fd3..52fce4f5 100644 --- a/packages/mapping/test/convert.test.ts +++ b/packages/mapping/test/convert.test.ts @@ -177,6 +177,16 @@ describe("整理计划 → 整理记录", () => { expect(M.organizationStatusFromFiles(["succeeded", "running"])).toBe("running"); expect(M.organizationStatusFromFiles(["cancelled"])).toBe("cancelled"); expect(M.organizationStatusFromFiles([])).toBe("succeeded"); + // 被排除的待确认视频(季集 / 季目录 / 命名)让成功降为 partial;预期跳过不影响 + expect(M.organizationStatusFromFiles(["succeeded", "skipped"], { unresolved: 1 })).toBe("partial"); + expect(M.organizationStatusFromFiles(["succeeded", "failed"], { unresolved: 1 })).toBe("partial"); + expect(M.organizationStatusFromFiles(["failed", "skipped"], { unresolved: 1 })).toBe("failed"); + expect(M.isUnresolvedExclusion("excluded:episode_unresolved")).toBe(true); + expect(M.isUnresolvedExclusion("excluded:no_target_dir")).toBe(true); + expect(M.isUnresolvedExclusion("excluded:naming_failed")).toBe(true); + expect(M.isUnresolvedExclusion("excluded:too_small")).toBe(false); + expect(M.isUnresolvedExclusion("already_in_library")).toBe(false); + expect(M.isUnresolvedExclusion(null)).toBe(false); }); }); From 0744be5c68830919cb736e66ddaca742b3b40111 Mon Sep 17 00:00:00 2001 From: root Date: Sat, 3 Oct 2026 20:59:06 +0800 Subject: [PATCH 2/2] =?UTF-8?q?=E6=95=B4=E7=90=86=EF=BC=9A=E5=8D=A1?= =?UTF-8?q?=E4=BD=8F=E8=AE=B0=E5=BD=95=E5=B7=A1=E6=A3=80=E9=87=8D=E6=8A=95?= =?UTF-8?q?=E3=80=81=E5=BA=93=E6=A0=B9=E7=A6=BB=E7=BA=BF=E6=9A=82=E5=81=9C?= =?UTF-8?q?=E3=80=81=E6=8E=92=E9=99=A4=E8=A7=86=E9=A2=91=E5=BC=80=E5=BE=85?= =?UTF-8?q?=E5=A4=84=E7=90=86=E3=80=81Move=20=E5=90=8E=E8=A1=A5=E7=A7=BB?= =?UTF-8?q?=E9=99=A4=20qB=20=E4=BB=BB=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 巡检(挂在"补提交与补入库"):running 且无活跃 verifyAndImport 运行超过 30 分钟的记录以 record: 重新入队(迁移 0069 记 requeue_count / requeued_at),重投 3 次仍卡住 → 未完成文件记失败、记录收尾、 开 organize_stalled 待处理 - executeRecord 库根离线 / 建目标目录失败不再抛错耗尽重试:记录保持 running 写原因、开待处理 (library_root_offline:<根> 复用规划期同一事项),任务按等待收尾;离线记录每 5 分钟重投、不计上限, 恢复后清原因并关闭离线事项 - 排除的待确认视频统一开 organize_excluded 待处理,记录原因 excluded_unresolved - Move 后移除 qB 任务抽成 removeMovedDownload:重放已结束记录与巡检都会对 move 成功但未移除的下载任务再试; 修正"下次同步重试"的误导日志 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei --- .../modules/acquisition-organize/handlers.ts | 8 +- .../src/modules/acquisition-organize/index.ts | 13 +- .../services/organize-exec.ts | 86 ++++- .../services/organize-finalize.ts | 82 +++-- .../services/organize-records.ts | 165 ++++++++- .../organize-recovery.test.ts | 326 ++++++++++++++++++ packages/contracts/src/reasons.ts | 3 + packages/contracts/test/db-sync.test.ts | 3 +- .../db/migrations/0069_organize_requeue.ts | 23 ++ packages/db/src/types.ts | 2 + 10 files changed, 674 insertions(+), 37 deletions(-) create mode 100644 apps/server/test/acquisition-organize/organize-recovery.test.ts create mode 100644 packages/db/migrations/0069_organize_requeue.ts diff --git a/apps/server/src/modules/acquisition-organize/handlers.ts b/apps/server/src/modules/acquisition-organize/handlers.ts index 625e7fd3..dd5b4705 100644 --- a/apps/server/src/modules/acquisition-organize/handlers.ts +++ b/apps/server/src/modules/acquisition-organize/handlers.ts @@ -165,7 +165,13 @@ export function registerAoTasks(registry: TaskRegistry, d: AoDeps) { ...(await executeRecord(d, recordId, { runId: t.runId, abortSignal: t.abortSignal })), })); }, - (r) => (r.status === "failed" ? "failed" : r.status === "deferred" ? "waiting" : "succeeded"), + // 执行返回 running = 库根离线 / 建目录失败而暂停(记录写明原因并开了待处理,由整理巡检重新入队) + (r) => + r.status === "failed" + ? "failed" + : r.status === "deferred" || r.status === "running" + ? "waiting" + : "succeeded", ); }, }), diff --git a/apps/server/src/modules/acquisition-organize/index.ts b/apps/server/src/modules/acquisition-organize/index.ts index 97f29087..507f2183 100644 --- a/apps/server/src/modules/acquisition-organize/index.ts +++ b/apps/server/src/modules/acquisition-organize/index.ts @@ -16,6 +16,7 @@ import { registerOrganizationRoutes } from "./routes/organization.ts"; import { registerRuleRoutes } from "./routes/rules.ts"; import { type AoDeps, automationBlock, enqueueSweep, enqueueTracked, timeSlot } from "./services/common.ts"; import { SWEEP_SLOT_MS, sweepReservedPlans, syncDownloads } from "./services/downloads.ts"; +import { recoverStalledRecords } from "./services/organize-records.ts"; import { sweepDueSlots, sweepMetadataRefresh, sweepSubscriptionSearch } from "./services/subscriptions.ts"; import { organizeDecisionHook } from "./services/takeover.ts"; import { loadVolumes, usedPercent, volumeThresholds } from "./services/volumes.ts"; @@ -39,7 +40,7 @@ const CLEANUP_SLOT_MS = 5 * 60_000; * 后台只做编排,start() 把各项周期登记到定时任务注册表(schedules 模块的调度器驱动,不同任务并行、同一任务不重入): * downloads.sync —— qB 同步 + 缓存阈值暂停/恢复(syncDownloads),只有轻量 HTTP 与 DB 写; * cache.threshold —— 超硬阈值入队 cleanupDownloadCache; - * downloads.orchestrate —— 巡检补投 submitDownload / verifyAndImport; + * downloads.orchestrate —— 巡检补投 submitDownload / verifyAndImport;整理记录卡在 running 的重投、Move 后补移除 qB 任务; * slots.reevaluate / subscriptions.search / subscriptions.metadata —— Slot 到期重评、订阅周期搜索、总集数刷新; * library.reconcile —— 按周期入队 reconcileLibrary。 * 复制/删种/全量 stat 这些重活都进任务引擎(租约 + attemptToken),不在调度里 await。 @@ -149,14 +150,20 @@ export function createAcquisitionOrganizeModule(overrides: AoOverrides = {}): Se key: "downloads.orchestrate", defaults: { name: "补提交与补入库", - description: "补交没提交成功的下载，补整理下载完但没入库的任务", + description: "补交没提交成功的下载，补整理下载完但没入库的任务，重新排队卡住的整理", group: "download", schedule: every(planMs), }, run: guarded(d, async (d) => { const submitted = await sweepReservedPlans(d); const imports = await sweepPendingImports(d); - return `补提交 ${submitted}，补入库 ${imports}`; + const rec = await recoverStalledRecords(d); + return ( + `补提交 ${submitted}，补入库 ${imports}` + + (rec.requeued ? `，重排整理 ${rec.requeued}` : "") + + (rec.failed ? `，整理停止 ${rec.failed}` : "") + + (rec.removed ? `，移除下载任务 ${rec.removed}` : "") + ); }), }); registerScheduledJob(ctx, { diff --git a/apps/server/src/modules/acquisition-organize/services/organize-exec.ts b/apps/server/src/modules/acquisition-organize/services/organize-exec.ts index a3a08b16..38cc6677 100644 --- a/apps/server/src/modules/acquisition-organize/services/organize-exec.ts +++ b/apps/server/src/modules/acquisition-organize/services/organize-exec.ts @@ -13,11 +13,13 @@ import { import type { OrganizationRecord } from "@umt/contracts"; import { sql } from "@umt/db"; import { withAttemptToken } from "@umt/tasks"; +import { closeDecisions } from "../../sources-identity/service/common.ts"; import { errMessage, jsonb } from "../util.ts"; import { type AoDeps, upsertDecision } from "./common.ts"; // 入库:冻结整理计划(organization_record + 文件明细)→ 逐文件可恢复执行(操作账本 fencing)→ // 资产/位置/路径绑定同事务登记 → Slot 推进 → Move 后移除 qB 任务记录(不删数据)。 -import { finalizeRecord } from "./organize-finalize.ts"; +import { finalizeRecord, removeMovedDownload } from "./organize-finalize.ts"; +import { emitRecord } from "./organize-plan.ts"; import type { FrozenPlan } from "./organize-types.ts"; import { cleanupReplacement, settleSamePathReplacement, stagedReplacementBackup } from "./replacement.ts"; import { scrapeAfterImport } from "./scrape.ts"; @@ -89,6 +91,17 @@ export async function executeRecord( enabled: frozen.scrapeAfterImport, }); } + // Move 成功但移除 qB 任务没成功(qB 暂时连不上 / 收尾后进程退出):重放时补做 + if (rec.action === "move" && rec.status === "succeeded" && rec.download_task_id) { + const dt = await db + .selectFrom("download_task") + .select(["id", "instance_id", "torrent_hash"]) + .where("id", "=", rec.download_task_id) + .where("removed_at", "is", null) + .where("managed", "=", true) + .executeTakeFirst(); + if (dt) await removeMovedDownload(d, dt); + } return { recordId, status: rec.status, @@ -99,11 +112,51 @@ export async function executeRecord( const frozen = rec.plan as unknown as FrozenPlan; const fsr = fsRuntime(d); // 根检查也走执行子进程:NFS/FUSE 挂死 → offline(timeout),按根离线重试,绝不当成文件缺失 + // 离线 / 建目录失败不抛错(抛错会耗尽任务重试、记录卡在 running):记录保持 running 并写明原因、开待处理, + // 由整理巡检(recoverStalledRecords)在库根恢复后重新入队 const health = await fsr.proxy.checkRoot(frozen.rootPath, frozen.sentinel); if (health.kind === "offline") { - throw new Error(`媒体库目录离线(${health.reason})，稍后重试`); + return holdRecord(d, rec, `${ROOT_OFFLINE}:${health.reason}`, { + kind: "other", + severity: "critical", + title: "媒体库目录离线，整理已暂停", + summary: `${frozen.rootPath} 不可访问或未挂载（${health.reason}），恢复后自动继续整理。文件不会因此被判为缺失`, + subjectType: "library_root", + subjectId: frozen.libraryRootId, + dedupeKey: `library_root_offline:${frozen.libraryRootId}`, + evidence: { reason: health.reason, recordId }, + }); + } + if (rec.action !== "in_place") { + try { + for (const dir of frozen.targetDirs) await mkdir(dir, { recursive: true }); + } catch (err) { + const code = (err as { code?: string } | null)?.code ?? "EIO"; + return holdRecord(d, rec, `${TARGET_DIR_UNAVAILABLE}:${code}`, { + kind: "other", + title: "整理目标目录建不出来", + summary: `${errMessage(err)}；检查目录权限或挂载后会自动重试`, + subjectType: "organization_record", + subjectId: recordId, + dedupeKey: `organize_target_dir:${recordId}`, + evidence: { code, targetDirs: frozen.targetDirs }, + }); + } + } + // 之前因库根离线 / 建目录失败暂停过:清掉等待原因,库根离线事项随之关闭 + if (isHoldReason(rec.reason)) { + await db.updateTable("organization_record").set({ reason: null }).where("id", "=", recordId).execute(); + } + if (rec.reason?.startsWith(ROOT_OFFLINE)) { + await closeDecisions( + d.ctx, + db, + `library_root_offline:${frozen.libraryRootId}`, + "resolved", + { key: "root_online", note: "媒体库目录已恢复", organizationRecordId: recordId }, + "system", + ); } - if (rec.action !== "in_place") for (const dir of frozen.targetDirs) await mkdir(dir, { recursive: true }); // 写前顺带清本目录残留 partial(对照 MP:只清本目录、>24h;失败吞掉)。copy 路径在 copyWithVerify 内还会再清一次 if (rec.action !== "in_place" && fsr.partialMaxAgeMs > 0) { for (const dir of frozen.targetDirs) { @@ -392,6 +445,33 @@ export async function executeRecord( return finalizeRecord(d, recordId); } +/** 记录暂停的原因前缀:库根离线 / 建目标目录失败(整理巡检据此判断要不要计入重投上限) */ +export const ROOT_OFFLINE = "library_root_offline"; +export const TARGET_DIR_UNAVAILABLE = "target_dir_unavailable"; +export const isHoldReason = (reason: string | null | undefined): boolean => + !!reason && (reason.startsWith(`${ROOT_OFFLINE}:`) || reason.startsWith(`${TARGET_DIR_UNAVAILABLE}:`)); + +/** 暂停执行:记录保持 running、写原因并开待处理;返回 running 让任务按"等待"收尾,不耗尽重试 */ +async function holdRecord( + d: AoDeps, + rec: { id: string; file_succeeded: number; file_failed: number }, + reason: string, + decision: Parameters[2], +): Promise { + await d.ctx.db.transaction().execute(async (trx) => { + await trx + .updateTable("organization_record") + .set({ reason }) + .where("id", "=", rec.id) + .where("status", "in", ["running", "queued"]) + .execute(); + await upsertDecision(d, trx, decision); + await emitRecord(d, trx, rec.id); + }); + d.ctx.logger.warn({ recordId: rec.id, reason }, "整理暂停，等待巡检重试"); + return { recordId: rec.id, status: "running", succeeded: rec.file_succeeded, failed: rec.file_failed }; +} + /** * replace 的旧文件(copy 已同卷移到 <根>/.umt-trash/replaced-*)改放成文件管理回收站的布局 * <根real>/.umt-trash///<原相对路径> 并写 trash_item(与 file-manager/trash.ts moveToTrash 同构, diff --git a/apps/server/src/modules/acquisition-organize/services/organize-finalize.ts b/apps/server/src/modules/acquisition-organize/services/organize-finalize.ts index 5a1a71b6..ebf781cf 100644 --- a/apps/server/src/modules/acquisition-organize/services/organize-finalize.ts +++ b/apps/server/src/modules/acquisition-organize/services/organize-finalize.ts @@ -4,7 +4,7 @@ import { statSafe } from "@umt/adapters"; import type { DownloadTaskRow, OrganizationRecordRow } from "@umt/db"; import { sql } from "@umt/db"; import { findOtherVersions } from "@umt/domain"; -import { organizationStatusFromFiles } from "@umt/mapping"; +import { isUnresolvedExclusion, organizationStatusFromFiles } from "@umt/mapping"; import { errMessage, jsonb } from "../util.ts"; import { type AoDeps, type Exec, loadInstances, upsertDecision } from "./common.ts"; import { applySlotEvent } from "./downloads.ts"; @@ -274,7 +274,9 @@ export async function finalizeRecord(d: AoDeps, recordId: string): Promise f.status); - const status = organizationStatusFromFiles(statuses); + // 季集待确认 / 没有季目录 / 命名失败而被排除的视频留在原处:记录按 partial,Move 不移除下载任务 + const unresolved = files.filter((f) => f.status === "skipped" && isUnresolvedExclusion(f.skip_reason)); + const status = organizationStatusFromFiles(statuses, { unresolved: unresolved.length }); const counts = { file_succeeded: statuses.filter((s) => s === "succeeded").length, file_failed: statuses.filter((s) => s === "failed").length, @@ -290,10 +292,31 @@ export async function finalizeRecord(d: AoDeps, recordId: string): Promise 0 && counts.file_failed === 0 && !rec.reason + ? { reason: "excluded_unresolved" } + : {}), }) .where("id", "=", recordId) .execute(); + if (unresolved.length > 0) { + const shown = unresolved + .slice(0, 5) + .map((f) => `${f.source_path.split("/").pop()}(${f.skip_reason?.replace(/^excluded:/, "")})`); + await upsertDecision(d, trx, { + kind: "other", + title: `有 ${unresolved.length} 个视频没有整理`, + summary: `季集、季目录或命名没能确定，文件留在原处${ + rec.action === "move" ? "，下载任务保留" : "" + }：${shown.join("、")}${unresolved.length > shown.length ? " 等" : ""}`, + subjectType: "organization_record", + subjectId: recordId, + dedupeKey: `organize_excluded:${recordId}`, + evidence: { + downloadTaskId: rec.download_task_id, + files: unresolved.map((f) => ({ path: f.source_path, reason: f.skip_reason })), + }, + }); + } // 下载任务 / 计划 / Slot const move = rec.action === "move" && status === "succeeded"; @@ -449,30 +472,7 @@ export async function finalizeRecord(d: AoDeps, recordId: string): Promise, +): Promise { + const topo = (await loadInstances(d.ctx.db, { enabledOnly: false, ids: [dt.instance_id] }))[0]; + if (!topo) return false; + try { + const client = await d.ports.downloader(topo.inst); + await client.delete([dt.torrent_hash], { deleteFiles: false }); + await d.ctx.db + .updateTable("download_task") + .set({ removed_at: d.ports.now(), managed: false }) + .where("id", "=", dt.id) + .execute(); + return true; + } catch (err) { + d.ctx.logger.warn( + { downloadTaskId: dt.id, err: { message: errMessage(err) } }, + "Move 后移除 qB 任务失败，整理巡检会重试", + ); + return false; + } +} + /** 关闭下载任务的"整理冲突 / 做种义务阻止移动 / 没识别出作品"事项(resolution.key=manual_organize,记整理记录 id) */ export async function closeOrganizeDecisions(d: AoDeps, trx: Exec, downloadTaskId: string, rec: OrganizationRecordRow) { const by = rec.actor_id ? `${rec.actor_type}:${rec.actor_id}` : rec.actor_type; diff --git a/apps/server/src/modules/acquisition-organize/services/organize-records.ts b/apps/server/src/modules/acquisition-organize/services/organize-records.ts index d89bfd7f..4f9658f8 100644 --- a/apps/server/src/modules/acquisition-organize/services/organize-records.ts +++ b/apps/server/src/modules/acquisition-organize/services/organize-records.ts @@ -1,6 +1,10 @@ import { reasonSentence } from "@umt/contracts"; -import type { OrganizationRecordRow } from "@umt/db"; -import type { Exec } from "./common.ts"; +import { type OrganizationRecordRow, sql } from "@umt/db"; +import { errMessage, jsonb } from "../util.ts"; +import { type AoDeps, type Exec, enqueueSweep, timeSlot, upsertDecision } from "./common.ts"; +import { SWEEP_SLOT_MS } from "./downloads.ts"; +import { isHoldReason, ROOT_OFFLINE } from "./organize-exec.ts"; +import { finalizeRecord, removeMovedDownload } from "./organize-finalize.ts"; // 入库:冻结整理计划(organization_record + 文件明细)→ 逐文件可恢复执行(操作账本 fencing)→ // 资产/位置/路径绑定同事务登记 → Slot 推进 → Move 后移除 qB 任务记录(不删数据)。 @@ -190,3 +194,160 @@ export async function loadRecordsPublic(exec: Exec, rows: OrganizationRecordRow[ }; }); } + +// ── 整理巡检:卡在 running 的记录重新入队;Move 成功但下载任务没移除的补移除 ── + +/** 记录 running 且没有活跃任务运行超过这么久,视为卡住 */ +export const STALLED_RECORD_MS = 30 * 60_000; +/** 库根离线暂停的记录:每隔这么久重投一次查库根(不计入重投上限) */ +export const ROOT_WAIT_RECHECK_MS = 5 * 60_000; +/** 巡检重投上限;用完仍卡住 → 未完成的文件记失败、记录收尾、开待处理 */ +export const MAX_RECORD_REQUEUES = 3; +/** Move 后移除下载任务的补做窗口:收尾后稍等(避开收尾自己的移除),太久的不再管 */ +const MOVE_REMOVE_GRACE_MS = 2 * 60_000; +const MOVE_REMOVE_WINDOW_MS = 7 * 24 * 3600_000; + +/** 引擎里排队 / 执行中的 verifyAndImport 涉及的记录 id 与下载任务 id */ +async function activeImportSubjects(d: AoDeps): Promise> { + const ids = new Set(); + for (const status of ["pending", "running"] as const) { + for (const run of await d.ctx.engine.list({ taskName: "verifyAndImport", status, limit: 500 })) { + const input = (run.input ?? {}) as { recordId?: unknown; downloadTaskId?: unknown }; + if (typeof input.recordId === "string") ids.add(input.recordId); + if (typeof input.downloadTaskId === "string") ids.add(input.downloadTaskId); + } + } + return ids; +} + +/** + * 巡检整理记录(挂在"补提交与补入库"定时任务上): + * - running / queued、没有活跃的 verifyAndImport 运行、超过 STALLED_RECORD_MS 没动静(任务重试耗尽、 + * planImport 提交后入队前进程退出)→ 以 record: 重新入队(executeRecord 可重入:已完成文件跳过, + * 账本未出结果的先对账);重投超过上限 → 未完成文件记失败、记录收尾并开待处理; + * - 库根离线暂停的记录每 ROOT_WAIT_RECHECK_MS 重投一次,不计上限(库根恢复即继续); + * - Move 成功但下载任务未移除(移除 qB 任务失败)→ 再试移除。 + */ +export async function recoverStalledRecords(d: AoDeps): Promise<{ requeued: number; failed: number; removed: number }> { + const db = d.ctx.db; + const now = d.ports.now(); + const out = { requeued: 0, failed: 0, removed: 0 }; + const rows = await db + .selectFrom("organization_record") + .select(["id", "download_task_id", "reason", "requeue_count", "requeued_at", "started_at", "created_at"]) + .where("status", "in", ["running", "queued"]) + .where( + sql`coalesce(requeued_at, started_at, created_at)`, + "<", + new Date(now.getTime() - Math.min(STALLED_RECORD_MS, ROOT_WAIT_RECHECK_MS)), + ) + .orderBy("created_at") + .limit(20) + .execute(); + const due = rows.filter((r) => { + const since = (r.requeued_at ?? r.started_at ?? r.created_at).getTime(); + const wait = r.reason?.startsWith(`${ROOT_OFFLINE}:`) ? ROOT_WAIT_RECHECK_MS : STALLED_RECORD_MS; + return now.getTime() - since >= wait; + }); + const active = due.length ? await activeImportSubjects(d) : new Set(); + const slot = timeSlot(now, SWEEP_SLOT_MS); + for (const r of due) { + if (active.has(r.id) || (r.download_task_id && active.has(r.download_task_id))) continue; + const rootWait = r.reason?.startsWith(`${ROOT_OFFLINE}:`) ?? false; + try { + if (!rootWait && r.requeue_count >= MAX_RECORD_REQUEUES) { + await failStalledRecord(d, r.id, r.requeue_count); + out.failed++; + continue; + } + const q = await enqueueSweep( + d, + { + kind: "verifyAndImport", + idempotencyKey: `record:${r.id}`, + input: { recordId: r.id }, + subjectType: "organization_record", + subjectId: r.id, + }, + slot, + ); + if (!q) continue; + await db + .updateTable("organization_record") + .set({ requeued_at: now, ...(rootWait ? {} : { requeue_count: r.requeue_count + 1 }) }) + .where("id", "=", r.id) + .execute(); + out.requeued++; + } catch (err) { + d.ctx.logger.warn({ recordId: r.id, err: { message: errMessage(err) } }, "整理记录巡检处理失败"); + } + } + + // Move 成功、下载任务仍纳管未移除:补移除 qB 任务(同一下载任务只看最新一条记录) + const moved = await db + .selectFrom("organization_record as r") + .innerJoin("download_task as dt", "dt.id", "r.download_task_id") + .select(["dt.id", "dt.instance_id", "dt.torrent_hash"]) + .where("r.action", "=", "move") + .where("r.status", "=", "succeeded") + .where("r.finished_at", "<", new Date(now.getTime() - MOVE_REMOVE_GRACE_MS)) + .where("r.finished_at", ">", new Date(now.getTime() - MOVE_REMOVE_WINDOW_MS)) + .where("dt.removed_at", "is", null) + .where("dt.managed", "=", true) + .where((eb) => + eb.not( + eb.exists( + eb + .selectFrom("organization_record as r2") + .select("r2.id") + .whereRef("r2.download_task_id", "=", "r.download_task_id") + .whereRef("r2.created_at", ">", "r.created_at"), + ), + ), + ) + .limit(20) + .execute(); + for (const dt of moved) if (await removeMovedDownload(d, dt)) out.removed++; + return out; +} + +/** 重投用完仍卡住:未完成文件记失败,按已完成的文件收尾(登记资产、推进下载任务 / Slot),开待处理 */ +async function failStalledRecord(d: AoDeps, recordId: string, requeues: number): Promise { + const db = d.ctx.db; + await db.transaction().execute(async (trx) => { + const rec = await trx + .selectFrom("organization_record") + .select(["id", "status", "reason", "download_task_id"]) + .where("id", "=", recordId) + .forUpdate() + .executeTakeFirst(); + if (!rec || (rec.status !== "running" && rec.status !== "queued")) return; + await trx + .updateTable("organization_file_result") + .set({ + status: "failed", + error: jsonb({ code: "STALLED", message: "整理多次重试仍未完成" }), + finished_at: d.ports.now(), + }) + .where("organization_record_id", "=", recordId) + .where("status", "in", ["queued", "running"]) + .execute(); + await trx + .updateTable("organization_record") + .set({ reason: "organize_stalled" }) + .where("id", "=", recordId) + .execute(); + await upsertDecision(d, trx, { + kind: "other", + title: "整理一直没完成，已停止", + summary: `重新排队 ${requeues} 次仍没有完成${ + isHoldReason(rec.reason) ? `（${rec.reason}）` : "" + }；已完成的文件照常登记，其余请检查后重新整理`, + subjectType: "organization_record", + subjectId: recordId, + dedupeKey: `organize_stalled:${recordId}`, + evidence: { requeues, previousReason: rec.reason, downloadTaskId: rec.download_task_id }, + }); + }); + await finalizeRecord(d, recordId); +} diff --git a/apps/server/test/acquisition-organize/organize-recovery.test.ts b/apps/server/test/acquisition-organize/organize-recovery.test.ts new file mode 100644 index 00000000..606810ef --- /dev/null +++ b/apps/server/test/acquisition-organize/organize-recovery.test.ts @@ -0,0 +1,326 @@ +import { existsSync } from "node:fs"; +import { rm, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { type EpisodeResolution, parseRelease, resolveEpisodes } from "@umt/domain"; +import { toParsedColumns } from "@umt/mapping"; +import type { TaskRunView } from "@umt/tasks"; +import { createMockQb, createTempTree, type MockQb, makeTorrent, type TempTree } from "@umt/testkit"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { syncDownloads } from "../../src/modules/acquisition-organize/services/downloads.ts"; +import { executeRecord, planImport } from "../../src/modules/acquisition-organize/services/organize.ts"; +import { + MAX_RECORD_REQUEUES, + recoverStalledRecords, +} from "../../src/modules/acquisition-organize/services/organize-records.ts"; +import { canonicalParsed } from "../../src/modules/sources-identity/service/episode-resolution.ts"; +import { createHarness, type Harness, pgAvailable, seedBase, seedEntity } from "./harness.ts"; + +// 整理卡死的回收(docs/plan/system-review.md T1-2):running 记录巡检重投 / 重投上限转失败; +// 库根离线不耗尽重试、恢复后推进;被排除的待确认视频让记录 partial(Move 不删 qB 任务); +// Move 后移除 qB 任务失败可重试。 + +const hasPg = await pgAvailable(); +const SIZE = 300_000; +const MIN = 60_000; + +// 已完结:S1 = 12、S2 = 13(`- 25` → S02E13,`- 40` 超出总集数 → 待确认) +const TABLE = { + seasons: [ + { season: 1, total: 12 }, + { season: 2, total: 13 }, + ], + latestSeason: 2, + ended: true, +}; +const ani = (ep: number) => `[ANi] Recovery Show - ${ep} [1080P].mkv`; + +describe.skipIf(!hasPg)("整理卡死回收(真实 PG + mock qB + 临时目录)", () => { + let h: Harness; + let qb: MockQb; + let lib: TempTree; + let cache: TempTree; + let base: Awaited>; + let n = 0; + + beforeAll(async () => { + qb = await createMockQb({ addVisibleAfterPolls: 0 }); + lib = await createTempTree({}, { sentinel: true, prefix: "umt-reclib-" }); + cache = await createTempTree( + { + "stall/Stall.Show.S01E01.1080p.WEB-DL.x264-GRP.mkv": SIZE, + "cap/Cap.Show.S01E01.1080p.WEB-DL.x264-GRP.mkv": SIZE, + "offline/Offline.Show.S01E01.1080p.WEB-DL.x264-GRP.mkv": SIZE, + }, + { prefix: "umt-reccache-" }, + ); + h = await createHarness("recovery"); + base = await seedBase(h.db, { libraryPath: lib.root, cachePath: cache.root, qbUrl: qb.url }); + }, 60_000); + + afterAll(async () => { + await h?.close(); + await qb?.close(); + for (const t of [lib, cache]) await t?.cleanup(); + }); + + const record = (id: string) => + h.db.selectFrom("organization_record").selectAll().where("id", "=", id).executeTakeFirstOrThrow(); + const ago = (ms: number) => new Date(Date.now() - ms); + const decision = (dedupeKey: string) => + h.db.selectFrom("decision").selectAll().where("dedupe_key", "=", dedupeKey).executeTakeFirst(); + + /** 手动按路径整理(Copy),只提交记录、不入队:模拟 planImport 提交后入队前进程退出 */ + async function manualRecord(dir: string, title: string) { + const entityId = await seedEntity(h.db, { kind: "series", title, year: 2024 }); + const r = await planImport(h.d, { + sourcePaths: [cache.path(dir)], + mediaEntityId: entityId, + action: "copy", + trigger: "manual", + actorType: "user", + }); + expect(r.status, JSON.stringify(r)).toBe("running"); + return r.recordId!; + } + + /** qB 里已完成的下载任务(接管进来,文件落盘),关联到资源 */ + async function completedTask(releaseId: string, name: string, files: string[]) { + n++; + const hash = n.toString(16).padStart(40, "c"); + qb.seed({ hash, name, tags: ["umt"], save_path: cache.root }); + await qb.complete(hash, { files: files.map((f) => ({ path: `${name}/${f}`, size: SIZE })), writeFiles: true }); + await syncDownloads(h.d); + const dt = await h.db + .selectFrom("download_task") + .select("id") + .where("torrent_hash", "=", hash) + .executeTakeFirstOrThrow(); + await h.db.updateTable("download_task").set({ source_release_id: releaseId }).where("id", "=", dt.id).execute(); + return { dtId: dt.id, hash }; + } + + /** 已识别资源(规范季集按季表判定) */ + async function release(entityId: string, title: string) { + const raw = parseRelease(title); + const resolution: EpisodeResolution = resolveEpisodes({ parsed: raw, kind: "series", table: TABLE, rules: [] }); + const cols = toParsedColumns(canonicalParsed(raw, resolution)); + const t = makeTorrent({ name: title, length: SIZE, pieceSeed: Math.floor(Math.random() * 1e6) }); + return ( + await h.db + .insertInto("source_release") + .values({ + site_id: base.siteId, + site_release_id: `${Math.floor(Math.random() * 1e9)}`, + title, + size_bytes: SIZE, + infohash_v1: t.hash, + media_entity_id: entityId, + resolution_status: "resolved", + parsed: JSON.stringify(cols.parsed), + parser_version: cols.parserVersion, + episode_resolution: JSON.stringify(resolution), + raw: JSON.stringify({ _site: { hr: false } }), + }) + .returning("id") + .executeTakeFirstOrThrow() + ).id; + } + + async function series(title: string) { + const id = await seedEntity(h.db, { kind: "series", title, year: 2023 }); + await h.db + .updateTable("media_entity") + .set({ metadata: JSON.stringify({ airStatus: { value: "Ended", source: "tmdb" } }) }) + .where("id", "=", id) + .execute(); + for (const s of TABLE.seasons) { + await h.db + .insertInto("content_unit") + .values({ + media_entity_id: id, + kind: "season", + season_number: s.season, + episode_total: s.total, + status: "confirmed", + source: "tmdb", + }) + .execute(); + } + return id; + } + + it("记录卡在 running 且没有活跃运行:巡检以 record: 重投,执行后完成", async () => { + const id = await manualRecord("stall", "卡住的剧"); + // 刚提交:还没到判定时间 + expect((await recoverStalledRecords(h.d)).requeued).toBe(0); + + await h.db + .updateTable("organization_record") + .set({ started_at: ago(31 * MIN) }) + .where("id", "=", id) + .execute(); + // 引擎里还有这条记录的运行(排队 / 退避中):不重投 + const list = h.engine.list; + h.engine.list = async (f) => + f?.status === "pending" ? [{ runId: "r1", input: { recordId: id } } as unknown as TaskRunView] : []; + expect((await recoverStalledRecords(h.d)).requeued).toBe(0); + h.engine.list = list; + + const before = h.engine.enqueued.length; + expect((await recoverStalledRecords(h.d)).requeued).toBe(1); + const q = h.engine.enqueued.slice(before); + expect(q).toHaveLength(1); + expect(q[0]).toMatchObject({ task: "verifyAndImport", input: { recordId: id } }); + expect(q[0]!.key).toMatch(new RegExp(`^verifyAndImport:record:${id}:`)); + const rec = await record(id); + expect(rec.requeue_count).toBe(1); + expect(rec.requeued_at).not.toBeNull(); + // 刚重投:等下一个判定窗口 + expect((await recoverStalledRecords(h.d)).requeued).toBe(0); + + // 模拟 worker 执行重投的任务:可重入,正常完成 + const done = await executeRecord(h.d, id); + expect(done.status).toBe("succeeded"); + expect((await record(id)).status).toBe("succeeded"); + }); + + it("重投超过上限仍卡住:未完成文件记失败、记录失败并开待处理", async () => { + const id = await manualRecord("cap", "重投耗尽的剧"); + await h.db + .updateTable("organization_record") + .set({ started_at: ago(3 * 3600_000), requeued_at: ago(31 * MIN), requeue_count: MAX_RECORD_REQUEUES }) + .where("id", "=", id) + .execute(); + const before = h.engine.enqueued.length; + const r = await recoverStalledRecords(h.d); + expect(r.failed).toBe(1); + expect(h.engine.enqueued.length).toBe(before); + const rec = await record(id); + expect(rec).toMatchObject({ status: "failed", reason: "organize_stalled" }); + const files = await h.db + .selectFrom("organization_file_result") + .select(["status", "error"]) + .where("organization_record_id", "=", id) + .where("role", "=", "main") + .execute(); + expect(files[0]).toMatchObject({ status: "failed", error: { code: "STALLED" } }); + expect(await decision(`organize_stalled:${id}`)).toMatchObject({ status: "open", subject_id: id }); + expect((await recoverStalledRecords(h.d)).failed).toBe(0); + }); + + it("库根离线:不抛错耗尽重试,记录等待并开待处理;恢复后巡检重投(不计上限)并完成", async () => { + const id = await manualRecord("offline", "离线的剧"); + await rm(join(lib.root, ".uhd-root")); + const r = await executeRecord(h.d, id); + expect(r.status).toBe("running"); + let rec = await record(id); + expect(rec.status).toBe("running"); + expect(rec.reason).toMatch(/^library_root_offline:/); + expect(await decision(`library_root_offline:${base.rootId}`)).toMatchObject({ status: "open", kind: "other" }); + // 再执行一次(任务重放)也只是等待 + expect((await executeRecord(h.d, id)).status).toBe("running"); + + // 离线等待的记录按较短间隔重投,不计入上限 + await h.db + .updateTable("organization_record") + .set({ started_at: ago(6 * MIN) }) + .where("id", "=", id) + .execute(); + expect((await recoverStalledRecords(h.d)).requeued).toBe(1); + rec = await record(id); + expect(rec.requeue_count).toBe(0); + + await writeFile(join(lib.root, ".uhd-root"), "sentinel\n"); + const done = await executeRecord(h.d, id); + expect(done.status).toBe("succeeded"); + rec = await record(id); + expect(rec).toMatchObject({ status: "succeeded", reason: null }); + expect((await decision(`library_root_offline:${base.rootId}`))?.status).toBe("resolved"); + }); + + it("季包里一个视频集数待确认被排除:记录 partial,Move 不移除 qB 任务,开待处理", async () => { + await h.setSettings("organize", { defaultMode: "move" }); + try { + const entityId = await series("排除视频的剧"); + const relId = await release(entityId, "[ANi] Recovery Show - 25 [1080P][Baha][WEB-DL][AAC AVC][CHT][MP4]"); + const name = "[ANi] Recovery Show S2 Pack"; + const { dtId, hash } = await completedTask(relId, name, [ani(25), ani(40)]); + const plan = await planImport(h.d, { downloadTaskId: dtId, trigger: "auto", actorType: "system" }); + expect(plan.status, JSON.stringify(plan)).toBe("running"); + expect((await record(plan.recordId!)).action).toBe("move"); + const ex = await executeRecord(h.d, plan.recordId!); + expect(ex.status).toBe("partial"); + const rec = await record(plan.recordId!); + expect(rec).toMatchObject({ status: "partial", reason: "excluded_unresolved", file_failed: 0 }); + const dt = await h.db.selectFrom("download_task").selectAll().where("id", "=", dtId).executeTakeFirstOrThrow(); + expect(dt).toMatchObject({ removed_at: null, managed: true, status: "imported" }); + expect(qb.get(hash)).toBeDefined(); + expect(qb.callsTo("torrents/delete")).toHaveLength(0); + // 被排除的视频留在下载目录 + expect(existsSync(join(cache.root, name, ani(40)))).toBe(true); + expect(existsSync(join(cache.root, name, ani(25)))).toBe(false); + const dec = await decision(`organize_excluded:${plan.recordId}`); + expect(dec).toMatchObject({ status: "open", subject_id: plan.recordId }); + expect(JSON.stringify(dec!.evidence)).toContain("excluded:episode_unresolved"); + // partial 不在补移除范围内 + await h.db + .updateTable("organization_record") + .set({ finished_at: ago(10 * MIN) }) + .where("id", "=", plan.recordId!) + .execute(); + expect((await recoverStalledRecords(h.d)).removed).toBe(0); + expect(qb.get(hash)).toBeDefined(); + } finally { + await h.setSettings("organize", { defaultMode: "copy" }); + } + }); + + it("Move 后移除 qB 任务失败:重放已结束记录与巡检都会再试,成功后结束纳管", async () => { + await h.setSettings("organize", { defaultMode: "move" }); + try { + const entityId = await series("移除失败的剧"); + const mk = async (ep: number) => { + const relId = await release(entityId, `[ANi] Recovery Show - ${ep} [1080P][Baha][WEB-DL][AAC AVC][CHT][MP4]`); + return completedTask(relId, `[ANi] Recovery Show - ${ep}`, [ani(ep)]); + }; + const removed = async (dtId: string) => + (await h.db.selectFrom("download_task").selectAll().where("id", "=", dtId).executeTakeFirstOrThrow()) + .removed_at !== null; + + // ① 重放已结束记录时补移除 + const a = await mk(14); + const pa = await planImport(h.d, { downloadTaskId: a.dtId, trigger: "auto", actorType: "system" }); + expect(pa.status, JSON.stringify(pa)).toBe("running"); + qb.fault("torrents/delete", { status: 500, times: 20 }); + expect((await executeRecord(h.d, pa.recordId!)).status).toBe("succeeded"); + expect(await removed(a.dtId)).toBe(false); + expect(qb.get(a.hash)).toBeDefined(); + qb.clearFaults(); + await executeRecord(h.d, pa.recordId!); + expect(await removed(a.dtId)).toBe(true); + expect(qb.get(a.hash)).toBeUndefined(); + + // ② 巡检补移除(收尾后留出宽限期) + const b = await mk(15); + const pb = await planImport(h.d, { downloadTaskId: b.dtId, trigger: "auto", actorType: "system" }); + qb.fault("torrents/delete", { status: 500, times: 20 }); + expect((await executeRecord(h.d, pb.recordId!)).status).toBe("succeeded"); + expect(await removed(b.dtId)).toBe(false); + qb.clearFaults(); + expect((await recoverStalledRecords(h.d)).removed).toBe(0); + await h.db + .updateTable("organization_record") + .set({ finished_at: ago(3 * MIN) }) + .where("id", "=", pb.recordId!) + .execute(); + expect((await recoverStalledRecords(h.d)).removed).toBe(1); + expect(await removed(b.dtId)).toBe(true); + expect(qb.get(b.hash)).toBeUndefined(); + const dt = await h.db.selectFrom("download_task").selectAll().where("id", "=", b.dtId).executeTakeFirstOrThrow(); + expect(dt.managed).toBe(false); + } finally { + qb.clearFaults(); + await h.setSettings("organize", { defaultMode: "copy" }); + } + }); +}); diff --git a/packages/contracts/src/reasons.ts b/packages/contracts/src/reasons.ts index 52ca2b65..102e6b52 100644 --- a/packages/contracts/src/reasons.ts +++ b/packages/contracts/src/reasons.ts @@ -38,6 +38,9 @@ const REASONS: Record = { entity_not_found: { text: "作品已不存在" }, no_library_root: { text: "没有可用的媒体库目录", action: "在设置 → 媒体库与目录里添加" }, library_root_offline: { text: "媒体库目录不可访问或未挂载", action: "检查挂载后重试" }, + target_dir_unavailable: { text: "整理目标目录建不出来", action: "检查目录权限或挂载，恢复后自动重试" }, + organize_stalled: { text: "整理多次重试仍未完成", action: "检查后重新整理" }, + excluded_unresolved: { text: "有视频没能确定季集或目标目录，留在原处", action: "确认季集后重新整理" }, target_dir_outside_root: { text: "目标目录不在媒体库目录里" }, slot_entity_mismatch: { text: "收录版本和作品对不上" }, binding_on_other_root: { text: "作品已放在另一个媒体库目录", action: "在作品详情里调整目录" }, diff --git a/packages/contracts/test/db-sync.test.ts b/packages/contracts/test/db-sync.test.ts index 9502b20b..261bb4ad 100644 --- a/packages/contracts/test/db-sync.test.ts +++ b/packages/contracts/test/db-sync.test.ts @@ -68,7 +68,8 @@ const ROWS: Record Observation: { row: C.Observation }, OrganizationFileResult: { row: C.OrganizationFileResult }, OrganizeSourceSnapshot: { row: C.OrganizeSourceSnapshotRow }, - OrganizationRecord: { row: C.OrganizationRecord, dbOnly: ["changeXid"] }, + // requeue_count / requeued_at:整理巡检的内部计数,不外露 + OrganizationRecord: { row: C.OrganizationRecord, dbOnly: ["changeXid", "requeueCount", "requeuedAt"] }, OutboxEvent: { row: C.OutboxEvent }, PathBinding: { row: C.PathBinding }, PathMapping: { row: C.PathMapping }, diff --git a/packages/db/migrations/0069_organize_requeue.ts b/packages/db/migrations/0069_organize_requeue.ts new file mode 100644 index 00000000..b88a0435 --- /dev/null +++ b/packages/db/migrations/0069_organize_requeue.ts @@ -0,0 +1,23 @@ +import type { Kysely } from "kysely"; +import { execDdl } from "../src/migration-helpers.ts"; + +// 整理记录卡在 running 的巡检重投(docs/plan/system-review.md T1-2): +// - requeue_count:巡检重新入队的次数,超过上限转失败并开待处理 +// - requeued_at:最近一次巡检重投时间,下一次判定"卡住"从这里起算 + +// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型 +export async function up(db: Kysely): Promise { + await execDdl( + db, + `alter table organization_record add column if not exists requeue_count integer not null default 0; + alter table organization_record add column if not exists requeued_at timestamptz;`, + ); +} +// biome-ignore lint/suspicious/noExplicitAny: 同上 +export async function down(db: Kysely): Promise { + await execDdl( + db, + `alter table organization_record drop column if exists requeued_at; + alter table organization_record drop column if exists requeue_count;`, + ); +} diff --git a/packages/db/src/types.ts b/packages/db/src/types.ts index 3f2a61ad..3f0a22ae 100644 --- a/packages/db/src/types.ts +++ b/packages/db/src/types.ts @@ -1081,6 +1081,8 @@ export interface OrganizationRecordTable { change_xid: Generated; created_at: GeneratedTimestamptz; updated_at: GeneratedTimestamptz; + requeue_count: Generated; + requeued_at: Timestamptz | null; } export interface OrganizeSourceSnapshotTable {