diff --git a/apps/server/src/modules/acquisition-organize/services/downloads.ts b/apps/server/src/modules/acquisition-organize/services/downloads.ts index 081a5693..47223fc9 100644 --- a/apps/server/src/modules/acquisition-organize/services/downloads.ts +++ b/apps/server/src/modules/acquisition-organize/services/downloads.ts @@ -198,13 +198,17 @@ async function emitStatus( from: DownloadStatus | null, to: DownloadStatus, pauseReason: PauseReason | null, + /** 任务行的代数:removed 行复活后 +1,完成事件换一把幂等键,复活后的完成才能再次触发整理 */ + generation = 0, ) { await d.ctx.outbox.emit( exec, "download.status_changed", { downloadTaskId: id, from, to, pauseReason }, - // 完成事件幂等:只触发一次入库;其它状态变化不去重 - to === "completed" ? { idempotencyKey: `download.completed:${id}` } : {}, + // 完成事件幂等:同一代只触发一次入库;其它状态变化不去重 + to === "completed" + ? { idempotencyKey: generation > 0 ? `download.completed:${id}:g${generation}` : `download.completed:${id}` } + : {}, ); } @@ -275,6 +279,7 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?: tags: t.tags, }); let dtId: string | undefined; + let inUseBy: string | undefined; const result = await withAttemptToken( db, `submit:${plan.id}`, @@ -337,7 +342,82 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?: return; } const r = res as Extract; - const ins = await sql<{ id: string; inserted: boolean }>` + const existing = await trx + .selectFrom("download_task") + .select(["id", "acquisition_plan_id", "status", "removed_at"]) + .where("instance_id", "=", topo.inst.id) + .where("torrent_hash", "=", r.hash) + .forUpdate() + .executeTakeFirst(); + const removed = !!existing && (existing.removed_at !== null || existing.status === "removed"); + if ( + !ev.manual && + existing && + !removed && + existing.acquisition_plan_id && + existing.acquisition_plan_id !== plan.id + ) { + // 同一种子的任务行已挂着另一个计划(手动先下 / 另一 Slot 跨轮次选中同一资源):任务行只推进它挂着的计划, + // 本计划挂不上去,取消并释放 Slot 预留,不留没有任务的 submitted 孤儿(该资源对本 Slot 不再计划)。 + // 手动计划不受影响:它经 import_plan_id 挂到已有任务,由整理收尾结束(organize-finalize 的 importPlanId) + const owner = existing.acquisition_plan_id; + inUseBy = owner; + await trx + .updateTable("acquisition_plan") + .set({ status: "cancelled", status_reason: `torrent_in_use_by_plan:${owner}` }) + .where("id", "=", plan.id) + .execute(); + await applySlotEvent( + d, + trx, + plan.slot_id, + { type: "release" }, + { reason: "torrent_in_use_by_plan", planId: plan.id }, + ); + return; + } + if (existing && removed) { + // 同一种子的旧任务行已 removed(用户删过 / 清理删过 / Move 后删过):复活并改挂本计划, + // 换代让完成事件能再次触发整理 + const slotEntity = plan.slot_id + ? ( + await trx + .selectFrom("acquisition_slot") + .select("media_entity_id") + .where("id", "=", plan.slot_id) + .executeTakeFirst() + )?.media_entity_id + : null; + await trx + .updateTable("download_task") + .set((eb) => ({ + acquisition_plan_id: plan.id, + import_plan_id: ev.manual ? plan.id : null, + source_release_id: rel.id, + media_entity_id: slotEntity ?? ev.mediaEntityId ?? rel.media_entity_id, + name: r.name, + save_path: r.savePath, + content_path: r.contentPath, + category: topo.category, + tags: r.tags, + managed: true, + origin: r.kind === "already_exists" ? "adopted" : "submitted", + status: r.kind === "accepted_unconfirmed" ? "submitting" : "queued", + pause_reason: null, + removed_at: null, + completed_at: null, + no_progress_since: null, + last_progress_at: null, + missing_since: null, + missing_count: 0, + generation: eb("generation", "+", 1), + next_check_at: d.ports.now(), + })) + .where("id", "=", existing.id) + .execute(); + dtId = existing.id; + } else { + const ins = await sql<{ id: string; inserted: boolean }>` insert into download_task (instance_id, torrent_hash, infohash_v1, infohash_v2, acquisition_plan_id, import_plan_id, source_release_id, media_entity_id, name, save_path, content_path, category, tags, managed, origin, status, size_bytes, next_check_at) @@ -353,7 +433,8 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?: media_entity_id = coalesce(download_task.media_entity_id, excluded.media_entity_id), managed = true, tags = excluded.tags, updated_at = now() returning id, (xmax = 0) as inserted`.execute(trx); - dtId = ins.rows[0]!.id; + dtId = ins.rows[0]!.id; + } await trx .updateTable("acquisition_plan") .set({ status: "submitted", status_reason: r.kind }) @@ -369,13 +450,14 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?: await d.ctx.outbox.emit( trx, "download.submitted", - { downloadTaskId: dtId, instanceId: topo.inst.id, torrentHash: r.hash, planId: plan.id, name: r.name }, + { downloadTaskId: dtId!, instanceId: topo.inst.id, torrentHash: r.hash, planId: plan.id, name: r.name }, { idempotencyKey: `download.submitted:${plan.id}` }, ); }, }, ); if (result.kind === "rejected" || result.kind === "source_failed") return { status: "failed", reason: result.reason }; + if (inUseBy) return { status: "skipped", reason: "torrent_in_use_by_plan" }; const hash = (result as Extract).hash; // 季包/多集包:按缺集选文件(接管的已有任务不改文件选择);失败不影响提交结果 if (result.kind === "added" && plan.unit_keys.length > 0 && plan.file_selection === null) { @@ -636,6 +718,9 @@ export interface SyncStats { thresholdResumed: number; } +/** 连续多少轮同步(列表 + 按 hash 查)都查不到才判 removed;与资产对账的 missingConfirmAfter 同口径 */ +export const MISSING_CONFIRM_ROUNDS = 3; + export async function syncDownloads(d: AoDeps): Promise { const stats: SyncStats = { instances: 0, @@ -715,13 +800,20 @@ async function syncInstance(d: AoDeps, topo: InstanceTopology, stats: SyncStats) // 管理标签被移除:停止后续自动动作,保留历史与资产 await trx .updateTable("download_task") - .set({ managed: false, tags: still.tags, last_synced_at: now }) + .set({ managed: false, tags: still.tags, last_synced_at: now, missing_since: null, missing_count: 0 }) + .where("id", "=", row.id) + .execute(); + } else if (row.missing_count + 1 < MISSING_CONFIRM_ROUNDS) { + // 列表与按 hash 都查不到:先记缺席(qB 刚重启、任务还没加载完时会短暂返回空 / 部分列表) + await trx + .updateTable("download_task") + .set({ missing_since: row.missing_since ?? now, missing_count: row.missing_count + 1, last_synced_at: now }) .where("id", "=", row.id) .execute(); } else { await trx .updateTable("download_task") - .set({ status: "removed", removed_at: now, last_synced_at: now }) + .set({ status: "removed", removed_at: now, last_synced_at: now, missing_count: row.missing_count + 1 }) .where("id", "=", row.id) .execute(); await emitStatus(d, trx, row.id, row.status, "removed", null); @@ -771,7 +863,8 @@ async function observeLocked( let noProgressSince = row.no_progress_since; let lastProgressAt = row.last_progress_at; let stalledNow = false; - if (obs.status === "downloading") { + // importing / imported 期间(qB 重新校验 / 移动等中间态)不做无进展判定 + if (obs.status === "downloading" && (merged.status === "downloading" || merged.status === "stalled")) { if (downloaded > row.downloaded_bytes || !lastProgressAt) { lastProgressAt = now; noProgressSince = null; @@ -809,12 +902,28 @@ async function observeLocked( last_progress_at: lastProgressAt, no_progress_since: noProgressSince, last_synced_at: now, + missing_since: null, + missing_count: 0, ...(becameComplete ? { completed_at: now } : {}), }) .where("id", "=", row.id) .execute(); if (changed) - await emitStatus(d, trx, row.id, row.status, merged.status, merged.status === "paused" ? merged.pauseReason : null); + await emitStatus( + d, + trx, + row.id, + row.status, + merged.status, + 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 @@ -858,6 +967,37 @@ async function observeLocked( return changed; } +/** 视为"恢复正常"的状态(用户暂停不算恢复,事项留给人看) */ +const RECOVERED = new Set(["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; @@ -901,12 +1041,49 @@ async function adopt(d: AoDeps, topo: InstanceTopology, t: QbTorrent, now: Date, last_progress_at: now, ...(st.status === "completed" ? { completed_at: now } : {}), }) - .onConflict((oc) => oc.columns(["instance_id", "torrent_hash"]).doNothing()) - .returning("id") + // 同一种子的旧任务行已 removed:复活为接管任务(原计划已在移除时结束,不再挂回);未移除的冲突不动 + .onConflict((oc) => + oc + .columns(["instance_id", "torrent_hash"]) + .doUpdateSet((eb) => ({ + acquisition_plan_id: null, + import_plan_id: null, + source_release_id: eb.fn.coalesce("download_task.source_release_id", "excluded.source_release_id"), + media_entity_id: eb.fn.coalesce("download_task.media_entity_id", "excluded.media_entity_id"), + name: eb.ref("excluded.name"), + save_path: eb.ref("excluded.save_path"), + content_path: eb.ref("excluded.content_path"), + category: eb.ref("excluded.category"), + tags: eb.ref("excluded.tags"), + managed: true, + origin: "adopted", + status: eb.ref("excluded.status"), + pause_reason: eb.ref("excluded.pause_reason"), + raw_state: eb.ref("excluded.raw_state"), + progress: eb.ref("excluded.progress"), + size_bytes: eb.ref("excluded.size_bytes"), + downloaded_bytes: eb.ref("excluded.downloaded_bytes"), + remaining_bytes: eb.ref("excluded.remaining_bytes"), + ratio: eb.ref("excluded.ratio"), + seeding_time_sec: eb.ref("excluded.seeding_time_sec"), + last_synced_at: eb.ref("excluded.last_synced_at"), + last_progress_at: eb.ref("excluded.last_progress_at"), + completed_at: eb.ref("excluded.completed_at"), + no_progress_since: null, + removed_at: null, + missing_since: null, + missing_count: 0, + generation: eb("download_task.generation", "+", 1), + })) + .where((eb) => + eb.or([eb("download_task.removed_at", "is not", null), eb("download_task.status", "=", "removed")]), + ), + ) + .returning(["id", "generation"]) .executeTakeFirst(); if (!ins) return; stats.adopted++; - await emitStatus(d, trx, ins.id, null, st.status, st.status === "paused" ? st.pauseReason : null); + await emitStatus(d, trx, ins.id, null, st.status, st.status === "paused" ? st.pauseReason : null, ins.generation); if (st.status === "completed") stats.completed.push(ins.id); }); } diff --git a/apps/server/test/acquisition-organize/download-sync-recovery.test.ts b/apps/server/test/acquisition-organize/download-sync-recovery.test.ts new file mode 100644 index 00000000..c318bffc --- /dev/null +++ b/apps/server/test/acquisition-organize/download-sync-recovery.test.ts @@ -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(); + let base: Awaited>; + + 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); +}); diff --git a/apps/server/test/acquisition-organize/subscription.test.ts b/apps/server/test/acquisition-organize/subscription.test.ts index a779fc9e..776fe353 100644 --- a/apps/server/test/acquisition-organize/subscription.test.ts +++ b/apps/server/test/acquisition-organize/subscription.test.ts @@ -2,7 +2,11 @@ import { createQbClient } from "@umt/adapters"; 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 { submitDownload, syncDownloads } from "../../src/modules/acquisition-organize/services/downloads.ts"; +import { + 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 { applyEpisodeTotal, @@ -159,7 +163,7 @@ describe.skipIf(!hasPg)("订阅增强(真实 PG + mock qB + 临时目录)", () = // 用户在 qB 里删掉 E02:确认移除 → 计划失败释放单元 → Slot 到期重评 → 延后的 E02 另一版本得到计划 const qc = createQbClient({ instanceId: "t", baseUrl: qb.url, username: "admin", password: "adminadmin" }); await qc.delete([e2.hash], { deleteFiles: false }); - await syncDownloads(h.d); + for (let i = 0; i < MISSING_CONFIRM_ROUNDS; i++) await syncDownloads(h.d); expect(await planRow(p2)).toMatchObject({ status: "failed", status_reason: "download_removed" }); expect((await slotRow(slotId)).status).toBe("attention"); h.engine.enqueued.length = 0; diff --git a/apps/server/test/e2e/agent-quality.test.ts b/apps/server/test/e2e/agent-quality.test.ts index 7485fb35..c2c00ffd 100644 --- a/apps/server/test/e2e/agent-quality.test.ts +++ b/apps/server/test/e2e/agent-quality.test.ts @@ -41,6 +41,8 @@ describe.skipIf(!hasPg)("E15 Agent 质量(fake 模型 + 真实能力)", () => { name: "Dune.Part.Two.2024.1080p.BluRay.x264-GRP", status: "failed", last_error: JSON.stringify({ message: "qB 状态 missingFiles" }), + // 已连续缺席两轮:下一轮同步即确认移除(多轮缺席才判移除) + missing_count: 2, }) .returning("id") .executeTakeFirstOrThrow() diff --git a/apps/server/test/e2e/tag-removal.test.ts b/apps/server/test/e2e/tag-removal.test.ts index 2be080c8..8aab6f51 100644 --- a/apps/server/test/e2e/tag-removal.test.ts +++ b/apps/server/test/e2e/tag-removal.test.ts @@ -2,7 +2,7 @@ import { createQbClient } from "@umt/adapters"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; import type { createAcquisitionOrganizeModule } from "../../src/modules/acquisition-organize/index.ts"; import { evaluateAcquisition } from "../../src/modules/acquisition-organize/services/acquisition.ts"; -import { syncDownloads } from "../../src/modules/acquisition-organize/services/downloads.ts"; +import { MISSING_CONFIRM_ROUNDS, syncDownloads } from "../../src/modules/acquisition-organize/services/downloads.ts"; import { torrentFor } from "./fixtures/modules.ts"; import { createE2E, type E2E, pgAvailable, seedBase, waitFor } from "./harness.ts"; @@ -206,7 +206,8 @@ describe.skipIf(!hasPg)("E8 AT24 标签移除 / 任务消失", () => { }); await ext.delete([ids.D.hash], { deleteFiles: false }); const externalDeletes = e.qb.callsTo("torrents/delete").length; - await syncDownloads(deps()); + // 连续多轮查不到才确认移除 + for (let i = 0; i < MISSING_CONFIRM_ROUNDS; i++) await syncDownloads(deps()); const row = await dt("D"); expect(row.status).toBe("removed"); expect(row.removed_at).not.toBeNull(); diff --git a/packages/contracts/src/execution.ts b/packages/contracts/src/execution.ts index 3760a9e3..f3d67091 100644 --- a/packages/contracts/src/execution.ts +++ b/packages/contracts/src/execution.ts @@ -223,6 +223,11 @@ export const DownloadTask = EntityBase.extend({ removedAt: Timestamp.nullable(), fileList: z.array(z.object({ path: z.string(), size: NonNegInt, progress: z.number().optional() })).nullable(), lastError: JsonObject.nullable(), + /** 下载器里连续查不到的起点与轮数:达到阈值才判 removed,中途出现即清零 */ + missingSince: Timestamp.nullable().optional(), + missingCount: NonNegInt.optional(), + /** removed 行被再次提交 / 接管复活的次数(完成事件幂等键随之换代) */ + generation: NonNegInt.optional(), }); export type DownloadTask = z.infer; export const DownloadTaskCreate = DownloadTask.pick({ diff --git a/packages/db/migrations/0067_download_missing.ts b/packages/db/migrations/0067_download_missing.ts new file mode 100644 index 00000000..cb07d1ea --- /dev/null +++ b/packages/db/migrations/0067_download_missing.ts @@ -0,0 +1,25 @@ +import type { Kysely } from "kysely"; +import { execDdl } from "../src/migration-helpers.ts"; + +// 下载调度卡死修复(docs/plan/system-review.md ③): +// - missing_since / missing_count:qB 列表与按 hash 查都查不到时先记缺席,连续多轮才判 removed,中途出现即清零 +// - generation:同一 (实例, hash) 的 removed 行被再次提交 / 接管而复活时 +1,完成事件幂等键随之换代, +// 复活后的完成仍能触发整理 +// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型 +export async function up(db: Kysely): Promise { + await execDdl( + db, + `alter table download_task add column if not exists missing_since timestamptz; + alter table download_task add column if not exists missing_count integer not null default 0; + alter table download_task add column if not exists generation integer not null default 0;`, + ); +} +// biome-ignore lint/suspicious/noExplicitAny: 同上 +export async function down(db: Kysely): Promise { + await execDdl( + db, + `alter table download_task drop column if exists missing_since; + alter table download_task drop column if exists missing_count; + alter table download_task drop column if exists generation;`, + ); +} diff --git a/packages/db/src/types.ts b/packages/db/src/types.ts index f94ac4d5..fd96bd81 100644 --- a/packages/db/src/types.ts +++ b/packages/db/src/types.ts @@ -719,6 +719,9 @@ export interface DownloadTaskTable { updated_at: GeneratedTimestamptz; media_entity_id: string | null; import_plan_id: string | null; + missing_since: Timestamptz | null; + missing_count: Generated; + generation: Generated; } export interface DownloaderInstanceTable { diff --git a/packages/mapping/src/download.ts b/packages/mapping/src/download.ts index 35b65b89..7ee3f338 100644 --- a/packages/mapping/src/download.ts +++ b/packages/mapping/src/download.ts @@ -62,7 +62,9 @@ export function toDomainDownloadStatus(status: DownloadStatus, pauseReason: Paus /** * 同步轮询时合并"下载器观察"与"当前业务状态": - * - 业务叠加态在其底层下载状态不变时保留(importing/imported 遇 completed、stalled 遇 downloading、submitting 遇 queued); + * - importing / imported 只由整理侧推进:下载器的任何中间态(checkingResumeData / queuedDL / checkingUP / moving, + * 乃至出错 / 缺文件)都不把它改回下载阶段,同步只更新原始字段;只有 removed(调用方确认移除)例外; + * - 其余业务叠加态在其底层下载状态不变时保留(stalled 遇 downloading、submitting 遇 queued); * - 暂停原因为 policy/agent 且下载器仍是暂停时保留原因,不被改写成 user; * - 其余以观察为准。 */ @@ -72,7 +74,7 @@ export function mergeObservedDownloadState( ): ContractDownloadState { const next = TO_CONTRACT[observed]; const keep = - ((current.status === "importing" || current.status === "imported") && observed === "completed") || + ((current.status === "importing" || current.status === "imported") && observed !== "removed") || (current.status === "stalled" && observed === "downloading") || (current.status === "submitting" && observed === "queued"); if (keep) return { ...current }; diff --git a/packages/mapping/test/enums.test.ts b/packages/mapping/test/enums.test.ts index fa15264f..eb812c79 100644 --- a/packages/mapping/test/enums.test.ts +++ b/packages/mapping/test/enums.test.ts @@ -57,6 +57,11 @@ describe("下载状态", () => { st("paused", "threshold"), ); expect(M.mergeObservedDownloadState(st("imported"), "removed")).toEqual(st("removed")); + // 已进入整理侧:qB 重启 / 重新校验 / 移动的中间态不把状态改回下载阶段 + for (const o of ["queued", "downloading", "paused_by_user", "error", "missing_files"] as const) { + expect(M.mergeObservedDownloadState(st("imported"), o), o).toEqual(st("imported")); + expect(M.mergeObservedDownloadState(st("importing"), o), o).toEqual(st("importing")); + } }); });