Files
UHD_Media_Tool/apps/server/test/acquisition-organize/download-visibility.test.ts
T
rootandClaude Opus 5.5 19f7bd0330 fix(download): 下载器离线与下载目录读不到容量开待处理,季包暂停添加后选文件
- M5:单个下载器连续 10 分钟连不上开 downloader_offline(按实例去重),恢复连接后自动关闭
- M3:下载目录读不到容量按卷开 critical 的 cache_volume_unknown,读到后关闭;面板单独显示
  「下载目录读不到」不再归为等空间;cache.threshold 不再把未知当 0%
- M4:季包 / 多集包 .torrent 暂停添加、磁力用「拿到元数据后停止」,选完文件再恢复;文件清单
  拿不到时留给同步重试,超过 10 分钟整包下载并记 file_list_timeout
- L1:提交 commit 时锁计划复核仍是 reserved,被取消的不写回 submitted,qB 里刚加的任务暂停、
  去掉管理标签并打 <管理标签>-cancelled
- L3:已收单未确认的提交在同步首次看到任务时补调 afterDownloadAccepted
- L4:排队中(queuedDL)的纳管任务一并受阈值暂停
- L6:停用的下载器上还有在下的任务时仍同步这些任务(不接管新任务、不开离线事项)

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei
2026-10-03 23:20:29 +08:00

550 lines
22 KiB
TypeScript

import { type EpisodeResolution, parseRelease, resolveEpisodes } from "@umt/domain";
import { toParsedColumns } from "@umt/mapping";
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,
DOWNLOADER_OFFLINE_DECISION_MS,
PACK_METADATA_TIMEOUT_MS,
submitDownload,
syncDownloads,
} from "../../src/modules/acquisition-organize/services/downloads.ts";
import { loadVolumes } from "../../src/modules/acquisition-organize/services/volumes.ts";
import { canonicalParsed } from "../../src/modules/sources-identity/service/episode-resolution.ts";
import {
createHarness,
fakeDisk,
type Harness,
pgAvailable,
seedBase,
seedEntity,
seedRelease,
seedSlot,
} from "./harness.ts";
// 下载侧可见性与剩余隐患:下载器离线 / 下载目录读不到容量开待处理并自动关闭、H&R 未知按下载器设置放行、
// 季包暂停添加选完文件再恢复(磁力元数据延后 / 超时整包)、取消与提交并发、已收单未确认的提交补记订阅进度、
// 排队中的任务受阈值暂停、停用实例上的在途任务仍同步。真实 PG + mock qB。
const hasPg = await pgAvailable();
const SIZE = 300_000;
const TABLE = { seasons: [{ season: 1, total: 12 }], latestSeason: 1, ended: true };
const ep = (n: number) => `Pack.Show.S01E${String(n).padStart(2, "0")}.1080p.WEB-DL.x264-GRP.mkv`;
describe.skipIf(!hasPg)("下载侧可见性(真实 PG + mock qB)", () => {
let h: Harness;
let qb: MockQb;
let lib: TempTree;
let cache: TempTree;
const disk = fakeDisk(10_000_000_000);
/** releaseId → 种子来源(.torrent 字节或磁力) */
const sources = new Map<string, { kind: "file"; bytes: Uint8Array } | { kind: "magnet"; uri: string }>();
/** releaseId → 取种子时执行的钩子(模拟提交过程中发生的事) */
const onSource = new Map<string, () => Promise<void>>();
let base: Awaited<ReturnType<typeof seedBase>>;
let n = 0;
beforeAll(async () => {
qb = await createMockQb({ addVisibleAfterPolls: 0 });
lib = await createTempTree({}, { sentinel: true, prefix: "umt-vislib-" });
cache = await createTempTree({}, { prefix: "umt-viscache-" });
disk.state.watch = [cache.root];
disk.setPct(40);
h = await createHarness(
"dl_visibility",
{
diskUsage: disk.fn,
torrentSource: async ({ release }) => {
await onSource.get(release.id)?.();
return sources.get(release.id)!;
},
},
{ withSources: true },
);
base = await seedBase(h.db, { libraryPath: lib.root, cachePath: cache.root, qbUrl: qb.url });
}, 60_000);
afterAll(async () => {
await h?.close();
await qb?.close();
await lib?.cleanup();
await cache?.cleanup();
});
const decision = (key: string) =>
h.db
.selectFrom("decision")
.selectAll()
.where("dedupe_key", "=", key)
.orderBy("created_at", "desc")
.executeTakeFirst();
const planRow = (id: string) =>
h.db.selectFrom("acquisition_plan").selectAll().where("id", "=", id).executeTakeFirstOrThrow();
const dtByHash = (hash: string) =>
h.db.selectFrom("download_task").selectAll().where("torrent_hash", "=", hash).executeTakeFirst();
async function movieRelease(title: string, hr: boolean | "unknown" = false) {
const entityId = await seedEntity(h.db, { kind: "movie", title, year: 2020 + (n++ % 5) });
const slotId = await seedSlot(h.db, { entityId });
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: `${title}.2021.1080p.BluRay.x264-GRP`,
size: SIZE,
hash: t.hash,
hr,
});
sources.set(id, { kind: "file", bytes: t.bytes });
return { entityId, slotId, releaseId: 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]!;
}
/** 季包:剧集 + 季表 + 订阅收录版本 + 只缺 S01E03 的计划 */
async function seasonPack(opts: { magnet?: boolean } = {}) {
const title = `Pack Show ${n++}`;
const entityId = await seedEntity(h.db, { kind: "series", title, year: 2023 });
await h.db
.insertInto("content_unit")
.values({
media_entity_id: entityId,
kind: "season",
season_number: 1,
episode_total: 12,
status: "confirmed",
source: "tmdb",
})
.execute();
const sub = await h.db
.insertInto("subscription")
.values({ media_entity_id: entityId, media_kind: "series", mode: "continuous", status: "active" })
.returning("id")
.executeTakeFirstOrThrow();
const slot = await h.db
.insertInto("acquisition_slot")
.values({
media_entity_id: entityId,
subscription_id: sub.id,
name: "1080p",
hard_conditions: JSON.stringify({}),
upgrade_threshold: JSON.stringify({}),
})
.returning("id")
.executeTakeFirstOrThrow();
const relTitle = `Pack.Show.${n}.S01.1080p.WEB-DL.x264-GRP`;
const raw = parseRelease(relTitle);
const resolution: EpisodeResolution = resolveEpisodes({ parsed: raw, kind: "series", table: TABLE, rules: [] });
const cols = toParsedColumns(canonicalParsed(raw, resolution));
const files = [1, 2, 3, 4].map((e) => ({ path: ep(e), length: SIZE }));
const t = makeTorrent({ name: relTitle, files, pieceSeed: Math.floor(Math.random() * 1e6) });
const rel = await h.db
.insertInto("source_release")
.values({
site_id: base.siteId,
site_release_id: `${Math.floor(Math.random() * 1e9)}`,
title: relTitle,
size_bytes: SIZE * 4,
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();
sources.set(
rel.id,
opts.magnet
? { kind: "magnet", uri: `magnet:?xt=urn:btih:${t.hash}&dn=${relTitle}` }
: { kind: "file", bytes: t.bytes },
);
const plan = await h.db
.insertInto("acquisition_plan")
.values({
slot_id: slot.id,
source_release_id: rel.id,
status: "reserved",
evaluation: JSON.stringify({ instanceId: base.instanceId, savePath: cache.root, mediaEntityId: entityId }),
payload_hash: `ph-${rel.id}`,
idempotency_key: `ik-${rel.id}`,
unit_keys: ["s1e3"],
})
.returning("id")
.executeTakeFirstOrThrow();
return { planId: plan.id, hash: t.hash, relTitle };
}
/** 磁力的 mock 种子:元数据到达后换成真实文件清单 */
const giveMetadata = (hash: string) => {
const t = qb.get(hash)!;
t.files = [1, 2, 3, 4].map((e) => ({ name: ep(e), size: SIZE, progress: 0, priority: 1 }));
};
const priorities = (hash: string) =>
Object.fromEntries(qb.get(hash)!.files.map((f) => [f.name.split("/").pop()!, f.priority]));
const resumedHashes = () =>
[...qb.callsTo("torrents/resume"), ...qb.callsTo("torrents/start")].flatMap((c) =>
(c.form.hashes ?? "").split("|"),
);
it("下载器连不上超过 10 分钟开 downloader_offline 待处理(按实例去重),恢复连接后自动关闭", async () => {
const qb2 = await createMockQb();
const inst = await h.db
.insertInto("downloader_instance")
.values({
name: "离线的 qB",
base_url: "http://127.0.0.1:9",
username: "admin",
managed_tag: "umt-offline",
settings: JSON.stringify({ savePath: cache.root }),
last_seen_at: new Date(Date.now() - 2 * 60_000),
})
.returning("id")
.executeTakeFirstOrThrow();
const key = `downloader_offline:${inst.id}`;
try {
// 刚断 2 分钟:只标离线,不开事项
let st = await syncDownloads(h.d);
expect(st.unreachable).toEqual([inst.id]);
expect(await decision(key)).toBeUndefined();
await h.db
.updateTable("downloader_instance")
.set({ last_seen_at: new Date(Date.now() - DOWNLOADER_OFFLINE_DECISION_MS - 60_000) })
.where("id", "=", inst.id)
.execute();
await syncDownloads(h.d);
await syncDownloads(h.d);
const open = await h.db.selectFrom("decision").selectAll().where("dedupe_key", "=", key).execute();
expect(open).toHaveLength(1);
expect(open[0]).toMatchObject({
kind: "downloader_offline",
severity: "critical",
status: "open",
subject_type: "downloader_instance",
subject_id: inst.id,
});
expect(open[0]!.summary).toContain("分钟连不上");
// 恢复:地址改对
await h.db.updateTable("downloader_instance").set({ base_url: qb2.url }).where("id", "=", inst.id).execute();
st = await syncDownloads(h.d);
expect(st.unreachable).toEqual([]);
expect(await decision(key)).toMatchObject({ status: "resolved", resolved_by: "system:download-sync" });
const health = await h.db
.selectFrom("downloader_instance")
.select(["health_status", "last_seen_at"])
.where("id", "=", inst.id)
.executeTakeFirstOrThrow();
expect(health.health_status).toBe("healthy");
} finally {
await h.db.deleteFrom("downloader_instance").where("id", "=", inst.id).execute();
await qb2.close();
}
});
it("下载目录读不到容量:提交延后、面板显示「下载目录读不到」而不是等空间,按卷开 critical 事项,恢复后关闭", async () => {
const { releaseId } = await movieRelease("读不到容量片");
const planId = await planFor(releaseId);
disk.state.fail = true;
try {
const s = await submitDownload(h.d, planId);
expect(s).toMatchObject({ status: "deferred", reason: "cache:volume_state_unknown" });
// 评估时已建好提交任务行(假引擎不执行):按处理器的做法记下延后结果
await h.db
.updateTable("task")
.set({ status: "succeeded", output: JSON.stringify(s) })
.where("kind", "=", "submitDownload")
.where("idempotency_key", "=", `plan:${planId}`)
.execute();
await syncDownloads(h.d);
const vol = (await loadVolumes(h.d))[0]!;
const key = `cache_volume_unknown:${vol.volumeKey}`;
const dec = await decision(key);
expect(dec).toMatchObject({ kind: "cache_volume_unknown", severity: "critical", status: "open" });
expect(dec!.evidence).toMatchObject({ reason: "statfs_failed", path: cache.root, usage: { code: "EIO" } });
// 再同步一轮不重复开
await syncDownloads(h.d);
expect(await h.db.selectFrom("decision").select("id").where("dedupe_key", "=", key).execute()).toHaveLength(1);
const board = (await h.api("GET", "/download-tasks/board")).body.data;
const row = board.items.find((x: { id: string }) => x.id === planId);
expect(row).toMatchObject({ kind: "plan", wait: "volume_unknown", waitReason: "cache:volume_state_unknown" });
disk.state.fail = false;
await syncDownloads(h.d);
expect(await decision(key)).toMatchObject({ status: "resolved" });
} finally {
disk.state.fail = false;
await h.db
.updateTable("acquisition_plan")
.set({ status: "cancelled", status_reason: "test_cleanup" })
.where("id", "=", planId)
.execute();
}
});
it("H&R 未知:缺省锁定(hnr_unknown);下载器打开「H&R 未知也可清理」后保种到期即可清理", async () => {
const { releaseId, hash, entityId } = await movieRelease("H&R 未知片", "unknown");
const dt = await h.db
.insertInto("download_task")
.values({
instance_id: base.instanceId,
torrent_hash: hash,
source_release_id: releaseId,
media_entity_id: entityId,
name: "H&R 未知片",
save_path: cache.root,
content_path: `${cache.root}/H&R 未知片.mkv`,
tags: ["umt"],
managed: true,
status: "imported",
size_bytes: SIZE,
completed_at: new Date(Date.now() - 30 * 86_400_000),
})
.returning("id")
.executeTakeFirstOrThrow();
await h.db
.insertInto("organization_record")
.values({
download_task_id: dt.id,
trigger: "auto",
actor_type: "system",
action: "copy",
status: "succeeded",
} as never)
.execute();
const lockOf = async () => {
const vol = (await loadVolumes(h.d))[0]!;
return (await cleanupCandidates(h.d, vol)).find((c) => c.row.id === dt.id)!;
};
expect(await lockOf()).toMatchObject({ cleanable: false, lockReason: "hnr_unknown" });
await h.db
.updateTable("downloader_instance")
.set({ settings: JSON.stringify({ savePath: cache.root, policy: { hnrUnknownCleanable: true } }) })
.where("id", "=", base.instanceId)
.execute();
try {
expect(await lockOf()).toMatchObject({ cleanable: true, lockReason: null });
// H&R 明确未满足的不受这个开关影响
await h.db
.updateTable("source_release")
.set({ raw: JSON.stringify({ _site: { hr: true } }) })
.where("id", "=", releaseId)
.execute();
expect(await lockOf()).toMatchObject({ cleanable: false, lockReason: "hnr_unsatisfied" });
} finally {
await h.db
.updateTable("downloader_instance")
.set({ settings: JSON.stringify({ savePath: cache.root }) })
.where("id", "=", base.instanceId)
.execute();
await h.db.updateTable("download_task").set({ managed: false }).where("id", "=", dt.id).execute();
}
});
it("季包(.torrent)暂停添加:选完文件再恢复下载", async () => {
const p = await seasonPack();
const s = await submitDownload(h.d, p.planId);
expect(s, JSON.stringify(s)).toMatchObject({ status: "submitted", reason: "added" });
const add = qb.callsTo("torrents/add").at(-1)!;
expect(add.status).toBe(200);
const plan = await planRow(p.planId);
expect(plan.file_selection).toMatchObject({ mode: "selected", selectedUnits: ["s1e3"] });
expect(priorities(p.hash)).toEqual({ [ep(1)]: 0, [ep(2)]: 0, [ep(3)]: 1, [ep(4)]: 0 });
// 先设文件优先级再恢复
const prioAt = qb.callsTo("torrents/filePrio").at(-1)!.at;
const resume = qb
.callsTo("torrents/resume")
.concat(qb.callsTo("torrents/start"))
.find((c) => (c.form.hashes ?? "").includes(p.hash));
expect(resume).toBeDefined();
expect(resume!.at).toBeGreaterThanOrEqual(prioAt);
expect(qb.get(p.hash)!.state).toBe("downloading");
// 提交时任务行记为策略暂停,恢复后同步回到下载中
await syncDownloads(h.d);
expect(await dtByHash(p.hash)).toMatchObject({ status: "downloading", pause_reason: null });
});
it("磁力季包:不暂停添加而是「拿到元数据后停止」;元数据未就绪时延后选文件,同步里拿到后选完再恢复", async () => {
const p = await seasonPack({ magnet: true });
// 磁力元数据还没到:文件清单为空
qb.fault("torrents/files", { status: 200, body: "[]", times: 3 });
const s = await submitDownload(h.d, p.planId);
expect(s).toMatchObject({ status: "submitted", reason: "added" });
expect(await planRow(p.planId)).toMatchObject({ file_selection: { mode: "pending", attempts: 1 } });
expect(resumedHashes()).not.toContain(p.hash);
qb.clearFaults();
giveMetadata(p.hash);
await syncDownloads(h.d);
const plan = await planRow(p.planId);
expect(plan.file_selection).toMatchObject({ mode: "selected", selectedUnits: ["s1e3"] });
expect(plan.unit_keys).toEqual(["s1e3"]);
expect(priorities(p.hash)[ep(3)]).toBe(1);
expect(priorities(p.hash)[ep(1)]).toBe(0);
expect(resumedHashes()).toContain(p.hash);
});
it("磁力季包元数据一直拿不到:超时后整包下载并记录原因", async () => {
const p = await seasonPack({ magnet: true });
qb.fault("torrents/files", { status: 200, body: "[]", times: 1000 });
const realNow = h.d.ports.now;
try {
await submitDownload(h.d, p.planId);
await syncDownloads(h.d);
expect(await planRow(p.planId)).toMatchObject({ file_selection: { mode: "pending", attempts: 2 } });
h.d.ports.now = () => new Date(Date.now() + PACK_METADATA_TIMEOUT_MS + 60_000);
await syncDownloads(h.d);
} finally {
h.d.ports.now = realNow;
qb.clearFaults();
}
const plan = await planRow(p.planId);
expect(plan.file_selection).toMatchObject({ mode: "whole", reason: "file_list_timeout" });
expect(resumedHashes()).toContain(p.hash);
});
it("取消与提交并发:qB 已收单时计划被取消,commit 复核后不写回 submitted,qB 里的任务暂停并去掉管理标签", async () => {
const { releaseId, hash } = await movieRelease("并发取消片");
const planId = await planFor(releaseId);
onSource.set(releaseId, async () => {
const r = await h.api("POST", `/acquisition-plans/${planId}/cancel`);
expect(r.status).toBe(200);
});
const s = await submitDownload(h.d, planId);
expect(s).toMatchObject({ status: "skipped", reason: "plan_cancelled" });
expect(await planRow(planId)).toMatchObject({ status: "cancelled", status_reason: "cancelled_by_user" });
expect(await dtByHash(hash)).toBeUndefined();
const t = qb.get(hash)!;
expect(t.tags).not.toContain("umt");
expect(t.tags).toContain("umt-cancelled");
expect(t.state).toMatch(/^(paused|stopped)/);
// 同步不接管它,计划也不被复活
await syncDownloads(h.d);
expect(await dtByHash(hash)).toBeUndefined();
expect(await planRow(planId)).toMatchObject({ status: "cancelled" });
});
it("已收单未确认的提交:同步首次在 qB 里看到时补记订阅进度,计划改记已确认", async () => {
const { releaseId, hash, slotId } = await movieRelease("未确认片");
const plan = await h.db
.insertInto("acquisition_plan")
.values({
slot_id: slotId,
source_release_id: releaseId,
status: "submitted",
status_reason: "accepted_unconfirmed",
evaluation: JSON.stringify({ instanceId: base.instanceId }),
payload_hash: `ph-${releaseId}`,
idempotency_key: `ik-${releaseId}`,
})
.returning("id")
.executeTakeFirstOrThrow();
await h.db
.insertInto("download_task")
.values({
instance_id: base.instanceId,
torrent_hash: hash,
acquisition_plan_id: plan.id,
source_release_id: releaseId,
name: "未确认片",
tags: ["umt"],
managed: true,
status: "submitting",
})
.execute();
const sub = () =>
h.db
.selectFrom("subscription as s")
.innerJoin("acquisition_slot as sl", "sl.subscription_id", "s.id")
.select("s.last_progress_at")
.where("sl.id", "=", slotId)
.executeTakeFirstOrThrow();
expect((await sub()).last_progress_at).toBeNull();
// 还没出现在 qB 里:不补记
await syncDownloads(h.d);
expect(await planRow(plan.id)).toMatchObject({ status_reason: "accepted_unconfirmed" });
qb.seed({ hash, name: "未确认片", tags: ["umt"], save_path: cache.root });
const st = await syncDownloads(h.d);
expect(st.confirmedPlans).toEqual([plan.id]);
expect(await planRow(plan.id)).toMatchObject({ status: "submitted", status_reason: "accepted_confirmed" });
expect((await sub()).last_progress_at).not.toBeNull();
// 只补一次
expect((await syncDownloads(h.d)).confirmedPlans).toEqual([]);
await h.db.updateTable("download_task").set({ managed: false }).where("torrent_hash", "=", hash).execute();
});
it("排队中(queuedDL)的纳管任务也受阈值暂停,降到恢复线后恢复", async () => {
const { releaseId, hash } = await movieRelease("排队片");
const planId = await planFor(releaseId);
expect(await submitDownload(h.d, planId)).toMatchObject({ status: "submitted" });
qb.setState(hash, "queuedDL");
await syncDownloads(h.d);
expect(await dtByHash(hash)).toMatchObject({ status: "queued" });
disk.setPct(95);
try {
const st = await syncDownloads(h.d);
expect(st.thresholdPaused).toBeGreaterThan(0);
expect(await dtByHash(hash)).toMatchObject({ status: "paused", pause_reason: "threshold" });
expect(qb.get(hash)!.state).toMatch(/^(paused|stopped)/);
} finally {
disk.setPct(40);
}
await syncDownloads(h.d);
expect(await dtByHash(hash)).toMatchObject({ pause_reason: null });
expect((await dtByHash(hash))!.status).not.toBe("paused");
});
it("停用的实例上还有在途任务:仍同步其状态,但不接管新任务", async () => {
const inst = await h.db
.insertInto("downloader_instance")
.values({
name: "停用的 qB",
base_url: qb.url,
username: "admin",
managed_tag: "umt-old",
enabled: false,
settings: JSON.stringify({ savePath: cache.root }),
})
.returning("id")
.executeTakeFirstOrThrow();
const hash = (900 + n++).toString(16).padStart(40, "e");
const other = (901 + n++).toString(16).padStart(40, "e");
qb.seed({ hash, name: "停用实例上的任务", tags: ["umt-old"], save_path: cache.root });
qb.seed({ hash: other, name: "停用实例上的新任务", tags: ["umt-old"], save_path: cache.root });
qb.setProgress(hash, 0.5);
const dt = await h.db
.insertInto("download_task")
.values({
instance_id: inst.id,
torrent_hash: hash,
name: "停用实例上的任务",
tags: ["umt-old"],
managed: true,
status: "downloading",
progress: 0.1,
})
.returning("id")
.executeTakeFirstOrThrow();
try {
const st = await syncDownloads(h.d);
expect(st.unreachable).toEqual([]);
const row = await h.db.selectFrom("download_task").selectAll().where("id", "=", dt.id).executeTakeFirstOrThrow();
expect(row.progress).toBeCloseTo(0.5);
expect(await dtByHash(other)).toBeUndefined();
} finally {
await h.db.deleteFrom("download_task").where("instance_id", "=", inst.id).execute();
await h.db.deleteFrom("downloader_instance").where("id", "=", inst.id).execute();
}
});
});