整理：卡住记录巡检重投、库根离线暂停、排除视频开待处理、Move 后补移除 qB 任务
- 巡检(挂在"补提交与补入库"):running 且无活跃 verifyAndImport 运行超过 30 分钟的记录以 record:<id> 重新入队(迁移 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 <[email protected]> Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei
This commit is contained in:
rootandClaude Opus 5.5 committed
1 parent
70935f4c17
commit
0744be5c68
10 files changed
+674
-37
No files matched your search
@@ -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<ReturnType<typeof seedBase>>;
|
||||
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:<id> 重投,执行后完成", 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" });
|
||||
}
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user