pixel_pop/UHD_Media_Tool
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(download): removed 行再次提交 / 接管时复活，同 hash 第二个订阅计划取消（H1、M3/M6）

Browse Source 
- 提交遇到同 (实例, 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 <[email protected]>
Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei
This commit is contained in:
rootandClaude Opus 5.5 committed 2026-10-03 21:20:08 +08:00
1 parent 2159765949
commit cba5de731b
1 file changed
+136 -9
Download Patch File Download Diff File Expand all files Collapse all files
apps/server/src/modules/acquisition-organize/services/downloads.ts
+136 -9
View File
@@ -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<SubmitResult>(
db,
`submit:${plan.id}`,
@@ -337,7 +342,82 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?:
return;
}
const r = res as Extract<SubmitResult, { hash: string }>;
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<SubmitResult, { hash: string }>).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);
});
}