fix(download): 卡住 / 异常恢复后自动关闭事项（L2），补下载卡死修复测试
stalled → 恢复下载 / 完成、failed / missing → 恢复正常时，关闭同一任务 open / awaiting_user 的 download_stalled / download_failed 事项（resolved，system:download-sync）；Agent 调查中的由其自行核实关闭。 新增 download-sync-recovery 测试覆盖 removed 行复活、同 hash 第二个计划、imported 回退、多轮缺席、事项自动关闭。 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
d96f0ff19f
commit
1c2fff694b
2 files changed
+361
No files matched your search
@@ -918,6 +918,12 @@ async function observeLocked(
|
||||
merged.status === "paused" ? merged.pauseReason : null,
|
||||
row.generation,
|
||||
);
|
||||
// 卡住 / 异常后恢复正常(进度增长或状态恢复):关闭同一任务未关闭的无进展 / 异常事项
|
||||
if (changed && RECOVERED.has(merged.status)) {
|
||||
if (row.status === "stalled") await closeRecoveredDecision(d, trx, `download_stalled:${row.id}`, merged.status);
|
||||
if (row.status === "failed" || row.status === "missing")
|
||||
await closeRecoveredDecision(d, trx, `download_failed:${row.id}`, merged.status);
|
||||
}
|
||||
if (becameComplete) {
|
||||
stats.completed.push(row.id);
|
||||
const plan = row.acquisition_plan_id
|
||||
@@ -961,6 +967,37 @@ async function observeLocked(
|
||||
return changed;
|
||||
}
|
||||
|
||||
/** 视为"恢复正常"的状态(用户暂停不算恢复,事项留给人看) */
|
||||
const RECOVERED = new Set<DownloadStatus>(["queued", "downloading", "completed", "importing", "imported"]);
|
||||
|
||||
/**
|
||||
* 卡住 / 异常事项自动关闭:只关 open / awaiting_user;Agent 正在调查(investigating)的由它自己核实后关闭。
|
||||
* resolution 不带处理选项,executeResolution 不会再执行任何动作。
|
||||
*/
|
||||
async function closeRecoveredDecision(d: AoDeps, trx: Exec, dedupeKey: string, status: DownloadStatus) {
|
||||
const by = "system:download-sync";
|
||||
const rows = await trx
|
||||
.updateTable("decision")
|
||||
.set({
|
||||
status: "resolved",
|
||||
resolved_by: by,
|
||||
resolved_at: d.ports.now(),
|
||||
resolution: jsonb({ key: "download_recovered", optionKey: null, by, note: `下载已恢复（${status}）` }),
|
||||
})
|
||||
.where("dedupe_key", "=", dedupeKey)
|
||||
.where("status", "in", ["open", "awaiting_user"])
|
||||
.returning("id")
|
||||
.execute();
|
||||
for (const r of rows) {
|
||||
await d.ctx.outbox.emit(
|
||||
trx,
|
||||
"decision.updated",
|
||||
{ decisionId: r.id, status: "resolved" },
|
||||
{ idempotencyKey: `decision.updated:${r.id}:resolved` },
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/** 用户给已有任务贴了管理标签:登记接管,不改保存路径与其它标签 */
|
||||
async function adopt(d: AoDeps, topo: InstanceTopology, t: QbTorrent, now: Date, stats: SyncStats) {
|
||||
const db = d.ctx.db;
|
||||
|
||||
@@ -0,0 +1,324 @@
|
||||
import { createQbClient } from "@umt/adapters";
|
||||
import { sql } from "@umt/db";
|
||||
import { createMockQb, createTempTree, type MockQb, makeTorrent, type TempTree } from "@umt/testkit";
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import { evaluateAcquisition } from "../../src/modules/acquisition-organize/services/acquisition.ts";
|
||||
import {
|
||||
cleanupCandidates,
|
||||
MISSING_CONFIRM_ROUNDS,
|
||||
submitDownload,
|
||||
syncDownloads,
|
||||
} from "../../src/modules/acquisition-organize/services/downloads.ts";
|
||||
import { executeRecord, planImport } from "../../src/modules/acquisition-organize/services/organize.ts";
|
||||
import { loadVolumes } from "../../src/modules/acquisition-organize/services/volumes.ts";
|
||||
import {
|
||||
createHarness,
|
||||
fakeDisk,
|
||||
type Harness,
|
||||
pgAvailable,
|
||||
seedBase,
|
||||
seedEntity,
|
||||
seedRelease,
|
||||
seedSlot,
|
||||
} from "./harness.ts";
|
||||
|
||||
// 下载调度卡死修复:removed 行再次提交复活、同 hash 第二个计划不成孤儿、已入库任务不被 qB 中间态改回、
|
||||
// 多轮缺席才判移除、卡住 / 异常恢复后事项自动关闭。真实 PG + mock qB;无进展阈值设为 0 方便构造 stalled。
|
||||
|
||||
const hasPg = await pgAvailable();
|
||||
const SIZE = 300_000;
|
||||
|
||||
describe.skipIf(!hasPg)("下载同步与提交的卡死修复(真实 PG + mock qB)", () => {
|
||||
let h: Harness;
|
||||
let qb: MockQb;
|
||||
let lib: TempTree;
|
||||
let cache: TempTree;
|
||||
const disk = fakeDisk(10_000_000);
|
||||
const torrents = new Map<string, Uint8Array>();
|
||||
let base: Awaited<ReturnType<typeof seedBase>>;
|
||||
|
||||
beforeAll(async () => {
|
||||
qb = await createMockQb({ addVisibleAfterPolls: 0 });
|
||||
lib = await createTempTree({}, { sentinel: true, prefix: "umt-lib-" });
|
||||
cache = await createTempTree({}, { prefix: "umt-cache-" });
|
||||
disk.state.watch = [cache.root];
|
||||
disk.setPct(40);
|
||||
h = await createHarness("dl_recovery", {
|
||||
diskUsage: disk.fn,
|
||||
torrentSource: async ({ release }) => ({ kind: "file", bytes: torrents.get(release.id)! }),
|
||||
options: { noProgressMs: 0 },
|
||||
});
|
||||
base = await seedBase(h.db, { libraryPath: lib.root, cachePath: cache.root, qbUrl: qb.url });
|
||||
await h.setSettings("retention", { seedRetentionDays: 0 });
|
||||
}, 60_000);
|
||||
|
||||
afterAll(async () => {
|
||||
await h?.close();
|
||||
await qb?.close();
|
||||
await lib?.cleanup();
|
||||
await cache?.cleanup();
|
||||
});
|
||||
|
||||
const ext = () => createQbClient({ instanceId: "t", baseUrl: qb.url, username: "admin", password: "adminadmin" });
|
||||
const dtRow = (id: string) =>
|
||||
h.db.selectFrom("download_task").selectAll().where("id", "=", id).executeTakeFirstOrThrow();
|
||||
const planRow = (id: string) =>
|
||||
h.db.selectFrom("acquisition_plan").selectAll().where("id", "=", id).executeTakeFirstOrThrow();
|
||||
const slotRow = (id: string) =>
|
||||
h.db.selectFrom("acquisition_slot").selectAll().where("id", "=", id).executeTakeFirstOrThrow();
|
||||
const decision = (key: string) =>
|
||||
h.db
|
||||
.selectFrom("decision")
|
||||
.selectAll()
|
||||
.where("dedupe_key", "=", key)
|
||||
.orderBy("created_at", "desc")
|
||||
.executeTakeFirst();
|
||||
|
||||
async function movie(title: string, year: number) {
|
||||
const entityId = await seedEntity(h.db, { kind: "movie", title, year });
|
||||
const slotId = await seedSlot(h.db, { entityId });
|
||||
return { entityId, slotId };
|
||||
}
|
||||
|
||||
async function release(entityId: string, title: string) {
|
||||
const t = makeTorrent({ name: `${title}.mkv`, length: SIZE, pieceSeed: Math.floor(Math.random() * 1e6) });
|
||||
const id = await seedRelease(h.db, { siteId: base.siteId, entityId, title, size: SIZE, hash: t.hash });
|
||||
torrents.set(id, t.bytes);
|
||||
return { id, hash: t.hash };
|
||||
}
|
||||
|
||||
async function planFor(releaseId: string) {
|
||||
const ev = await evaluateAcquisition(h.d, { sourceReleaseId: releaseId });
|
||||
expect(ev.plans).toHaveLength(1);
|
||||
return ev.plans[0]!;
|
||||
}
|
||||
|
||||
/** 手动获取:同一资源可再次明确选择(slot_id 为空、evaluation.manual) */
|
||||
async function manualPlanLike(planId: string, entityId: string) {
|
||||
const r = await sql<{ id: string }>`
|
||||
insert into acquisition_plan (slot_id, source_release_id, purpose, status, priority, evaluation, reserved_bytes,
|
||||
budget_reservation, payload_hash, idempotency_key, status_reason)
|
||||
select null, source_release_id, 'fill', 'reserved', 'immediate',
|
||||
evaluation || ${JSON.stringify({ manual: true, mediaEntityId: entityId, importMode: "add" })}::jsonb,
|
||||
reserved_bytes, '{}'::jsonb, payload_hash, ${`manual:${crypto.randomUUID()}`}, 'manual_selection'
|
||||
from acquisition_plan where id = ${planId}
|
||||
returning id`.execute(h.db);
|
||||
return r.rows[0]!.id;
|
||||
}
|
||||
|
||||
async function syncTimes(n: number) {
|
||||
for (let i = 0; i < n; i++) await syncDownloads(h.d);
|
||||
}
|
||||
|
||||
it("removed 行再次提交:行复活挂到新计划,完成后换代发完成事件并能入库", async () => {
|
||||
const { entityId } = await movie("复活片", 2021);
|
||||
const rel = await release(entityId, "Revive.Movie.2021.1080p.BluRay.x264-GRP");
|
||||
const p1 = await planFor(rel.id);
|
||||
const s1 = await submitDownload(h.d, p1);
|
||||
expect(s1).toMatchObject({ status: "submitted", reason: "added" });
|
||||
const dtId = s1.downloadTaskId!;
|
||||
|
||||
await ext().delete([rel.hash], { deleteFiles: false });
|
||||
await syncTimes(MISSING_CONFIRM_ROUNDS);
|
||||
expect(await dtRow(dtId)).toMatchObject({ status: "removed" });
|
||||
expect((await dtRow(dtId)).removed_at).not.toBeNull();
|
||||
expect(await planRow(p1)).toMatchObject({ status: "failed", status_reason: "download_removed" });
|
||||
|
||||
// 用户再次明确选择同一资源
|
||||
const p2 = await manualPlanLike(p1, entityId);
|
||||
const s2 = await submitDownload(h.d, p2);
|
||||
expect(s2).toMatchObject({ status: "submitted", reason: "added", downloadTaskId: dtId });
|
||||
const revived = await dtRow(dtId);
|
||||
expect(revived).toMatchObject({
|
||||
status: "queued",
|
||||
removed_at: null,
|
||||
completed_at: null,
|
||||
acquisition_plan_id: p2,
|
||||
import_plan_id: p2,
|
||||
origin: "submitted",
|
||||
missing_count: 0,
|
||||
generation: 1,
|
||||
});
|
||||
expect(await planRow(p2)).toMatchObject({ status: "submitted", status_reason: "added" });
|
||||
|
||||
await qb.complete(rel.hash, { writeFiles: true });
|
||||
const sync = await syncDownloads(h.d);
|
||||
expect(sync.completed).toContain(dtId);
|
||||
const done = (await h.outbox("download.status_changed")).filter(
|
||||
(e) => e.payload.downloadTaskId === dtId && e.payload.to === "completed",
|
||||
);
|
||||
expect(done.map((e) => e.idempotency_key)).toEqual([`download.completed:${dtId}:g1`]);
|
||||
|
||||
const plan = await planImport(h.d, { downloadTaskId: dtId, trigger: "auto", actorType: "system" });
|
||||
expect(plan.status).toBe("running");
|
||||
expect(await executeRecord(h.d, plan.recordId!)).toMatchObject({ status: "succeeded", succeeded: 1 });
|
||||
expect((await dtRow(dtId)).status).toBe("imported");
|
||||
}, 60_000);
|
||||
|
||||
it("removed 行被重新贴管理标签:接管时复活为 adopted 行", async () => {
|
||||
const { entityId } = await movie("接管复活", 2020);
|
||||
const rel = await release(entityId, "Adopt.Revive.2020.1080p.WEB-DL.x264-GRP");
|
||||
const s = await submitDownload(h.d, await planFor(rel.id));
|
||||
const dtId = s.downloadTaskId!;
|
||||
await ext().delete([rel.hash], { deleteFiles: false });
|
||||
await syncTimes(MISSING_CONFIRM_ROUNDS);
|
||||
expect((await dtRow(dtId)).status).toBe("removed");
|
||||
|
||||
qb.seed({ hash: rel.hash, name: "Adopt.Revive.2020.1080p.WEB-DL.x264-GRP.mkv", tags: ["umt"], size: SIZE });
|
||||
qb.setProgress(rel.hash, 0.2);
|
||||
const st = await syncDownloads(h.d);
|
||||
expect(st.adopted).toBe(1);
|
||||
expect(await dtRow(dtId)).toMatchObject({
|
||||
origin: "adopted",
|
||||
acquisition_plan_id: null,
|
||||
removed_at: null,
|
||||
status: "downloading",
|
||||
generation: 1,
|
||||
});
|
||||
}, 60_000);
|
||||
|
||||
it("同 hash 已挂在途计划:第二个计划取消并释放 Slot 预留,不留 submitted 孤儿", async () => {
|
||||
const { entityId } = await movie("同种双计划", 2022);
|
||||
const rel = await release(entityId, "Shared.Hash.2022.1080p.BluRay.x264-GRP");
|
||||
const p1 = await planFor(rel.id);
|
||||
const s1 = await submitDownload(h.d, p1);
|
||||
expect(s1.status).toBe("submitted");
|
||||
|
||||
// 另一个 Slot(另一订阅 / 版本)跨轮次选中同一资源
|
||||
const slot2 = await seedSlot(h.db, { entityId, name: "second" });
|
||||
const ev = await evaluateAcquisition(h.d, { acquisitionSlotId: slot2 });
|
||||
expect(ev.plans).toHaveLength(1);
|
||||
const p2 = ev.plans[0]!;
|
||||
expect((await slotRow(slot2)).status).toBe("reserved");
|
||||
const out = await submitDownload(h.d, p2);
|
||||
expect(out).toEqual({ status: "skipped", reason: "torrent_in_use_by_plan" });
|
||||
expect(await planRow(p2)).toMatchObject({ status: "cancelled", status_reason: `torrent_in_use_by_plan:${p1}` });
|
||||
const s2 = await slotRow(slot2);
|
||||
expect(["reserved", "downloading", "verifying"]).not.toContain(s2.status);
|
||||
expect(s2.reserved_plan_id).toBeNull();
|
||||
// 任务行仍只推进第一个计划;同一资源只剩一个 submitted 计划
|
||||
expect((await dtRow(s1.downloadTaskId!)).acquisition_plan_id).toBe(p1);
|
||||
expect(
|
||||
await h.db
|
||||
.selectFrom("acquisition_plan")
|
||||
.select("id")
|
||||
.where("source_release_id", "=", rel.id)
|
||||
.where("status", "=", "submitted")
|
||||
.execute(),
|
||||
).toEqual([{ id: p1 }]);
|
||||
}, 60_000);
|
||||
|
||||
it("已入库任务经 checkingResumeData / checkingUP / moving 再回到完成:仍是 imported,completed_at 不变,清理仍能选中", async () => {
|
||||
const { entityId } = await movie("回退片", 2019);
|
||||
const rel = await release(entityId, "Regress.Movie.2019.1080p.BluRay.x264-GRP");
|
||||
const s = await submitDownload(h.d, await planFor(rel.id));
|
||||
const dtId = s.downloadTaskId!;
|
||||
await qb.complete(rel.hash, { writeFiles: true });
|
||||
await syncDownloads(h.d);
|
||||
const plan = await planImport(h.d, { downloadTaskId: dtId, trigger: "auto", actorType: "system" });
|
||||
expect(await executeRecord(h.d, plan.recordId!)).toMatchObject({ status: "succeeded" });
|
||||
const before = await dtRow(dtId);
|
||||
expect(before.status).toBe("imported");
|
||||
expect(before.completed_at).not.toBeNull();
|
||||
|
||||
const t = qb.get(rel.hash)!;
|
||||
const done = t.state;
|
||||
for (const [state, progress] of [
|
||||
["checkingResumeData", 1],
|
||||
["queuedDL", 1],
|
||||
["checkingUP", 0.4],
|
||||
["moving", 1],
|
||||
] as const) {
|
||||
t.state = state;
|
||||
t.progress = progress;
|
||||
await syncDownloads(h.d);
|
||||
const row = await dtRow(dtId);
|
||||
expect(row.status, state).toBe("imported");
|
||||
expect(row.raw_state, state).toBe(state);
|
||||
}
|
||||
qb.setProgress(rel.hash, 1);
|
||||
t.state = done;
|
||||
await syncDownloads(h.d);
|
||||
const after = await dtRow(dtId);
|
||||
expect(after.status).toBe("imported");
|
||||
expect(after.completed_at?.getTime()).toBe(before.completed_at!.getTime());
|
||||
expect(after.no_progress_since).toBeNull();
|
||||
const completedEvents = (await h.outbox("download.status_changed")).filter(
|
||||
(e) => e.payload.downloadTaskId === dtId && e.payload.from === "imported",
|
||||
);
|
||||
expect(completedEvents).toHaveLength(0);
|
||||
|
||||
const vol = (await loadVolumes(h.d))[0]!;
|
||||
const cand = (await cleanupCandidates(h.d, vol)).find((c) => c.row.id === dtId);
|
||||
expect(cand).toMatchObject({ cleanable: true, lockReason: null });
|
||||
}, 60_000);
|
||||
|
||||
it("qB 一轮查不到不判移除;连续缺席达到阈值才判移除;中途重新出现清掉缺席", async () => {
|
||||
const { entityId } = await movie("缺席片", 2018);
|
||||
const rel = await release(entityId, "Missing.Movie.2018.1080p.BluRay.x264-GRP");
|
||||
const p = await planFor(rel.id);
|
||||
const s = await submitDownload(h.d, p);
|
||||
const dtId = s.downloadTaskId!;
|
||||
const name = qb.get(rel.hash)!.name;
|
||||
|
||||
// qB 刚重启:列表与按 hash 都查不到
|
||||
await ext().delete([rel.hash], { deleteFiles: false });
|
||||
const st1 = await syncDownloads(h.d);
|
||||
expect(st1.removed).toBe(0);
|
||||
const r1 = await dtRow(dtId);
|
||||
expect(r1).toMatchObject({ removed_at: null, missing_count: 1 });
|
||||
expect(r1.status).not.toBe("removed");
|
||||
expect(r1.missing_since).not.toBeNull();
|
||||
expect((await planRow(p)).status).toBe("submitted");
|
||||
|
||||
// 任务加载回来:缺席清零
|
||||
qb.seed({ hash: rel.hash, name, tags: ["umt"], size: SIZE });
|
||||
qb.setProgress(rel.hash, 0.3);
|
||||
await syncDownloads(h.d);
|
||||
expect(await dtRow(dtId)).toMatchObject({ missing_count: 0, missing_since: null, removed_at: null });
|
||||
|
||||
// 真被删:连续缺席到阈值才判移除
|
||||
await ext().delete([rel.hash], { deleteFiles: false });
|
||||
await syncTimes(MISSING_CONFIRM_ROUNDS - 1);
|
||||
expect((await dtRow(dtId)).removed_at).toBeNull();
|
||||
expect((await dtRow(dtId)).missing_count).toBe(MISSING_CONFIRM_ROUNDS - 1);
|
||||
const st = await syncDownloads(h.d);
|
||||
expect(st.removed).toBe(1);
|
||||
expect(await dtRow(dtId)).toMatchObject({ status: "removed" });
|
||||
expect(await planRow(p)).toMatchObject({ status: "failed", status_reason: "download_removed" });
|
||||
}, 60_000);
|
||||
|
||||
it("stalled / 异常后恢复:同一任务未关闭的事项自动关闭为 resolved", async () => {
|
||||
const { entityId } = await movie("卡住片", 2017);
|
||||
const rel = await release(entityId, "Stalled.Movie.2017.1080p.BluRay.x264-GRP");
|
||||
const s = await submitDownload(h.d, await planFor(rel.id));
|
||||
const dtId = s.downloadTaskId!;
|
||||
qb.setState(rel.hash, "downloading");
|
||||
qb.setProgress(rel.hash, 0.2);
|
||||
await syncDownloads(h.d); // 有进展
|
||||
await syncDownloads(h.d); // 字节不增长,阈值 0 → stalled
|
||||
expect((await dtRow(dtId)).status).toBe("stalled");
|
||||
expect(await decision(`download_stalled:${dtId}`)).toMatchObject({ kind: "download_stalled", status: "open" });
|
||||
|
||||
qb.setProgress(rel.hash, 0.5);
|
||||
await syncDownloads(h.d);
|
||||
expect((await dtRow(dtId)).status).toBe("downloading");
|
||||
const closed = await decision(`download_stalled:${dtId}`);
|
||||
expect(closed).toMatchObject({ status: "resolved", resolved_by: "system:download-sync" });
|
||||
expect((await h.outbox("decision.updated")).map((e) => e.payload)).toContainEqual({
|
||||
decisionId: closed!.id,
|
||||
status: "resolved",
|
||||
});
|
||||
|
||||
// qB 报错 → download_failed;恢复下载 → 关闭
|
||||
qb.setError(rel.hash);
|
||||
await syncDownloads(h.d);
|
||||
expect((await dtRow(dtId)).status).toBe("failed");
|
||||
expect(await decision(`download_failed:${dtId}`)).toMatchObject({ status: "open" });
|
||||
qb.setState(rel.hash, "downloading");
|
||||
qb.setProgress(rel.hash, 0.7);
|
||||
await syncDownloads(h.d);
|
||||
expect((await dtRow(dtId)).status).toBe("downloading");
|
||||
expect(await decision(`download_failed:${dtId}`)).toMatchObject({ status: "resolved" });
|
||||
}, 60_000);
|
||||
});
|
||||
Reference in new issue
Block a user