From cba5de731bb1305ebcc656caa954bc39952095cc Mon Sep 17 00:00:00 2001 From: root Date: Sat, 3 Oct 2026 21:20:08 +0800 Subject: [PATCH] =?UTF-8?q?fix(download):=20removed=20=E8=A1=8C=E5=86=8D?= =?UTF-8?q?=E6=AC=A1=E6=8F=90=E4=BA=A4=20/=20=E6=8E=A5=E7=AE=A1=E6=97=B6?= =?UTF-8?q?=E5=A4=8D=E6=B4=BB=EF=BC=8C=E5=90=8C=20hash=20=E7=AC=AC?= =?UTF-8?q?=E4=BA=8C=E4=B8=AA=E8=AE=A2=E9=98=85=E8=AE=A1=E5=88=92=E5=8F=96?= =?UTF-8?q?=E6=B6=88=EF=BC=88H1=E3=80=81M3/M6=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 提交遇到同 (实例, hash) 的 removed 行：清 removed_at、状态回到 queued / submitting、改挂新计划、generation+1； - 接管（adopt）冲突时 removed 行同样复活为 adopted 行； - 完成事件幂等键按 generation 换代，复活后完成仍能触发整理； - 订阅计划遇到任务行已挂着别的计划：本计划取消（torrent_in_use_by_plan:<计划>）并释放 Slot 预留，不再留下没有任务的 submitted 孤儿；手动计划仍经 import_plan_id 挂到已有任务。 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei --- .../services/downloads.ts | 145 ++++++++++++++++-- 1 file changed, 136 insertions(+), 9 deletions(-) diff --git a/apps/server/src/modules/acquisition-organize/services/downloads.ts b/apps/server/src/modules/acquisition-organize/services/downloads.ts index bfd47283..a13f98ea 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) { @@ -815,7 +897,15 @@ async function observeLocked( .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 (becameComplete) { stats.completed.push(row.id); const plan = row.acquisition_plan_id @@ -902,12 +992,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); }); }