From 74f2c7649e481a17e8fe648eece993359a607baf Mon Sep 17 00:00:00 2001 From: root Date: Sat, 3 Oct 2026 23:20:28 +0800 Subject: [PATCH 1/5] =?UTF-8?q?feat(db):=20=E5=BE=85=E5=A4=84=E7=90=86?= =?UTF-8?q?=E6=96=B0=E5=A2=9E=E3=80=8C=E4=B8=8B=E8=BD=BD=E5=99=A8=E8=BF=9E?= =?UTF-8?q?=E4=B8=8D=E4=B8=8A=E3=80=8D=E3=80=8C=E4=B8=8B=E8=BD=BD=E7=9B=AE?= =?UTF-8?q?=E5=BD=95=E8=AF=BB=E4=B8=8D=E5=88=B0=E5=AE=B9=E9=87=8F=E3=80=8D?= =?UTF-8?q?=E4=B8=A4=E7=B1=BB=EF=BC=88=E8=BF=81=E7=A7=BB=200073=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit decision.kind 增加 downloader_offline / cache_volume_unknown,契约、原因文案与界面标签同步; volume_state_unknown 的原因文案改为「下载目录读不到容量」并指向路径映射与容器挂载。 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei --- apps/web/app/components/library/labels.ts | 2 + apps/web/app/lib/labels.ts | 2 + packages/contracts/src/agent.ts | 4 ++ packages/contracts/src/reasons.ts | 4 +- .../migrations/0073_downloader_visibility.ts | 40 +++++++++++++++++++ packages/db/src/types.ts | 4 +- 6 files changed, 54 insertions(+), 2 deletions(-) create mode 100644 packages/db/migrations/0073_downloader_visibility.ts diff --git a/apps/web/app/components/library/labels.ts b/apps/web/app/components/library/labels.ts index 043b9afe..08c20617 100644 --- a/apps/web/app/components/library/labels.ts +++ b/apps/web/app/components/library/labels.ts @@ -142,6 +142,8 @@ export const DECISION_KIND: Record = { approval_required: "需要审批", metadata_conflict: "元数据冲突", source_unreachable: "来源不可达", + downloader_offline: "下载器连不上", + cache_volume_unknown: "下载目录读不到容量", other: "其他", }; diff --git a/apps/web/app/lib/labels.ts b/apps/web/app/lib/labels.ts index 931e24bf..d5ceae79 100644 --- a/apps/web/app/lib/labels.ts +++ b/apps/web/app/lib/labels.ts @@ -49,6 +49,8 @@ const DECISION_KIND: Record = { approval_required: "待审批", metadata_conflict: "元数据冲突", source_unreachable: "来源不可达", + downloader_offline: "下载器连不上", + cache_volume_unknown: "下载目录读不到容量", other: "其它", }; export const decisionKindLabel = (v: string) => DECISION_KIND[v] ?? v; diff --git a/packages/contracts/src/agent.ts b/packages/contracts/src/agent.ts index 9148237f..423578e2 100644 --- a/packages/contracts/src/agent.ts +++ b/packages/contracts/src/agent.ts @@ -149,6 +149,10 @@ export const DecisionKind = z.enum([ "identity_unresolved", /** 资源已绑定作品但集数待确认(无法换算 / 映射规则失效):按作品 + 来源季 + 发布组合并,附建议规则 */ "episode_mapping", + /** 下载器连续不可达超过一段时间(按实例去重,恢复后自动关闭) */ + "downloader_offline", + /** 下载目录读不到容量(路径映射 / 挂载问题,按卷去重,恢复后自动关闭) */ + "cache_volume_unknown", ]); export type DecisionKind = z.infer; export const DecisionStatus = z.enum(["open", "investigating", "awaiting_user", "resolved", "dismissed", "expired"]); diff --git a/packages/contracts/src/reasons.ts b/packages/contracts/src/reasons.ts index fd624f6c..505f4897 100644 --- a/packages/contracts/src/reasons.ts +++ b/packages/contracts/src/reasons.ts @@ -66,7 +66,7 @@ const REASONS: Record = { not_found: { text: "没找到" }, precondition: { text: "前置条件不满足" }, internal: { text: "内部错误" }, - volume_state_unknown: { text: "下载盘用量未知", action: "在下载面板检查下载盘" }, + volume_state_unknown: { text: "下载目录读不到容量", action: "检查下载器的路径映射和容器挂载" }, hard_threshold_reached: { text: "下载盘已到暂停线" }, projected_over_threshold: { text: "下完会超过下载盘暂停线" }, awaiting_resume_watermark: { text: "等下载盘降到恢复线" }, @@ -85,6 +85,8 @@ const REASONS: Record = { download_failed: { text: "下载失败" }, download_stalled: { text: "下载无进展" }, space_insufficient: { text: "空间不足" }, + downloader_offline: { text: "下载器连不上" }, + cache_volume_unknown: { text: "下载目录读不到容量" }, unidentifiable: { text: "没识别出作品" }, user_deleted: { text: "已删除" }, // DL-3:还没提交就结束或在等的下载(计划的 statusReason、提交任务的延后原因) diff --git a/packages/db/migrations/0073_downloader_visibility.ts b/packages/db/migrations/0073_downloader_visibility.ts new file mode 100644 index 00000000..518d55e9 --- /dev/null +++ b/packages/db/migrations/0073_downloader_visibility.ts @@ -0,0 +1,40 @@ +import type { Kysely } from "kysely"; +import { execDdl, oneOf } from "../src/migration-helpers.ts"; + +// 下载侧可见性:decision.kind 增加 +// - downloader_offline:下载器连续不可达超过一段时间(按实例去重,恢复后自动关闭) +// - cache_volume_unknown:下载目录读不到容量(路径映射 / 挂载问题,按卷去重,恢复后自动关闭) + +const OLD_DECISION_KINDS = [ + "identity_ambiguous", + "download_stalled", + "download_failed", + "path_conflict", + "space_insufficient", + "approval_required", + "metadata_conflict", + "source_unreachable", + "other", + "identity_unresolved", + "episode_mapping", +]; +const NEW_DECISION_KINDS = [...OLD_DECISION_KINDS, "downloader_offline", "cache_volume_unknown"]; + +const swapDecisionKinds = (values: readonly string[]) => ` +alter table decision drop constraint decision_kind_check; +alter table decision add constraint decision_kind_check ${oneOf("kind", values)}; +`; + +// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型 +export async function up(db: Kysely): Promise { + await execDdl(db, swapDecisionKinds(NEW_DECISION_KINDS)); +} + +// biome-ignore lint/suspicious/noExplicitAny: 同上 +export async function down(db: Kysely): Promise { + await execDdl( + db, + `update decision set kind = 'other' where kind in ('downloader_offline', 'cache_volume_unknown'); + ${swapDecisionKinds(OLD_DECISION_KINDS)}`, + ); +} diff --git a/packages/db/src/types.ts b/packages/db/src/types.ts index fd96bd81..b9801dfc 100644 --- a/packages/db/src/types.ts +++ b/packages/db/src/types.ts @@ -112,7 +112,9 @@ export type DecisionKind = | "source_unreachable" | "other" | "identity_unresolved" - | "episode_mapping"; + | "episode_mapping" + | "downloader_offline" + | "cache_volume_unknown"; export type DecisionStatus = "open" | "investigating" | "awaiting_user" | "resolved" | "dismissed" | "expired"; export type DecisionSeverity = "info" | "warning" | "critical"; export type DownloadTaskOrigin = "submitted" | "adopted"; From 3d3a4cefc6c0070cf6a681bad1e7c7686ad2c0de Mon Sep 17 00:00:00 2001 From: root Date: Sat, 3 Oct 2026 23:20:28 +0800 Subject: [PATCH 2/5] =?UTF-8?q?fix(agent):=20download.recheck=20=E5=A4=8D?= =?UTF-8?q?=E7=94=A8=E4=B8=8B=E8=BD=BD=E5=90=8C=E6=AD=A5=E7=9A=84=20qB=20?= =?UTF-8?q?=E5=AE=A2=E6=88=B7=E7=AB=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit qB 客户端缓存改为按 ServerContext 共享(downloaderClientFor),Agent 不再自己 new 客户端, 与下载同步、面板共用同一登录会话和凭证处理。 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei --- .../src/modules/acquisition-organize/ports.ts | 47 +++++++++++++------ .../modules/agent/capabilities-download.ts | 16 ++----- 2 files changed, 35 insertions(+), 28 deletions(-) diff --git a/apps/server/src/modules/acquisition-organize/ports.ts b/apps/server/src/modules/acquisition-organize/ports.ts index 444ed111..d783106d 100644 --- a/apps/server/src/modules/acquisition-organize/ports.ts +++ b/apps/server/src/modules/acquisition-organize/ports.ts @@ -117,12 +117,42 @@ export const DEFAULT_OPTIONS: AoOptions = { const str = (v: unknown) => (typeof v === "string" && v.length > 0 ? v : undefined); +/** qB 客户端按 ServerContext 共享缓存(实例配置变了重建):下载同步、面板、Agent 的 download.recheck 共用同一登录会话 */ +const sharedClients = new WeakMap>(); + +/** 按实例取 qB 客户端(凭证现场经 secrets 解出,解析失败按无凭证处理并记警告) */ +export async function downloaderClientFor(ctx: ServerContext, inst: DownloaderInstanceRow): Promise { + let clients = sharedClients.get(ctx); + if (!clients) { + clients = new Map(); + sharedClients.set(ctx, clients); + } + const stamp = `${inst.updated_at.toISOString()}|${inst.base_url}`; + const hit = clients.get(inst.id); + if (hit && hit.stamp === stamp) return hit.client; + const settings = (inst.settings ?? {}) as Record; + let secret: string | undefined; + if (inst.secret_ref) { + try { + secret = await ctx.secrets.resolve(inst.secret_ref); + } catch (err) { + ctx.logger.warn({ err: { message: (err as Error).message } }, "凭证解析失败"); + } + } + const client = createQbClient({ + instanceId: inst.id, + baseUrl: inst.base_url, + ...qbAuthConfig({ authKind: settings.authKind, username: inst.username, secret: secret ?? null }), + }); + clients.set(inst.id, { stamp, client }); + return client; +} + /** 用 ctx 组装默认端口;overrides 逐项覆盖(测试) */ export function createPorts( ctx: ServerContext, overrides: Partial> & { options?: Partial } = {}, ): AoPorts { - const clients = new Map(); const resolveSecret = async (ref: string | null): Promise => { if (!ref) return undefined; try { @@ -159,20 +189,7 @@ export function createPorts( if (r.created) await enqueueNewEntityRefresh(ctx, r.mediaEntityId); return r.mediaEntityId; }, - async downloader(inst) { - const stamp = `${inst.updated_at.toISOString()}|${inst.base_url}`; - const hit = clients.get(inst.id); - if (hit && hit.stamp === stamp) return hit.client; - const settings = (inst.settings ?? {}) as Record; - const secret = await resolveSecret(inst.secret_ref); - const client = createQbClient({ - instanceId: inst.id, - baseUrl: inst.base_url, - ...qbAuthConfig({ authKind: settings.authKind, username: inst.username, secret: secret ?? null }), - }); - clients.set(inst.id, { stamp, client }); - return client; - }, + downloader: (inst) => downloaderClientFor(ctx, inst), async torrentSource({ release, site }) { // 永久性条件用 SiteError:submitDownload 据此把计划判失败并释放 Slot,而不是反复重试、Slot 一直 reserved if (!site) throw new SiteError("", "not_found", "站点不存在，无法获取种子"); diff --git a/apps/server/src/modules/agent/capabilities-download.ts b/apps/server/src/modules/agent/capabilities-download.ts index 05cb0f33..09851e95 100644 --- a/apps/server/src/modules/agent/capabilities-download.ts +++ b/apps/server/src/modules/agent/capabilities-download.ts @@ -1,7 +1,6 @@ -import { createQbClient, type DownloaderClient, qbAuthConfig } from "@umt/adapters"; import { CapabilityError, type CapabilityRegistry } from "@umt/agent"; import { capabilitySpec, type DownloadStatus } from "@umt/contracts"; -import type { DownloaderInstanceRow } from "@umt/db"; +import { downloaderClientFor } from "../acquisition-organize/ports.ts"; import type { ServerContext } from "../types.ts"; import { planBinding } from "./authorize.ts"; import type { AgentModuleOptions } from "./runner.ts"; @@ -15,16 +14,6 @@ const RECHECK_FROM: DownloadStatus[] = ["missing", "failed", "stalled", "paused" /** resume 允许的起点 */ const RESUME_FROM: DownloadStatus[] = ["paused", "stalled", "failed"]; -async function defaultClient(ctx: ServerContext, inst: DownloaderInstanceRow): Promise { - const settings = (inst.settings ?? {}) as Record; - const secret = inst.secret_ref ? await ctx.secrets.resolve(inst.secret_ref) : undefined; - return createQbClient({ - instanceId: inst.id, - baseUrl: inst.base_url, - ...qbAuthConfig({ authKind: settings.authKind, username: inst.username, secret: secret ?? null }), - }); -} - /** 持续授权的对象范围:按下载任务 → 计划 → Slot → 作品核对(authorize 对非计划类能力拿不到目标作品) */ async function checkGrantScope(ctx: ServerContext, grantId: string, planId: string | null) { const g = await ctx.db.selectFrom("automation_grant").select("scope").where("id", "=", grantId).executeTakeFirst(); @@ -75,7 +64,8 @@ export function registerDownloadRecoveryCapabilities( .where("id", "=", dt.instance_id) .executeTakeFirst(); if (!inst) throw new CapabilityError("not_found", "下载器实例不存在"); - const client = opts.downloader ? await opts.downloader(inst) : await defaultClient(ctx, inst); + // 与下载同步共用按实例缓存的客户端(同一登录会话与凭证处理),不另建 + const client = opts.downloader ? await opts.downloader(inst) : await downloaderClientFor(ctx, inst); try { if (input.action === "recheck") await client.recheck([dt.torrent_hash]); else await client.resume([dt.torrent_hash]); From 07d5f421433a9015635641c4292bccfdaecaa67e Mon Sep 17 00:00:00 2001 From: root Date: Sat, 3 Oct 2026 23:20:29 +0800 Subject: [PATCH 3/5] =?UTF-8?q?feat(download):=20=E4=B8=8B=E8=BD=BD?= =?UTF-8?q?=E5=99=A8=E5=8F=AF=E8=AE=BE=E3=80=8CH&R=20=E6=9C=AA=E7=9F=A5?= =?UTF-8?q?=E4=B9=9F=E5=8F=AF=E6=B8=85=E7=90=86=E3=80=8D=EF=BC=88M2?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 下载器策略加 hnrUnknownCleanable(缺省关,保持现状):保种天数到期、站点 H&R 要求未知 (M-Team、接管任务、站点没配 H&R)的任务视为可清理;H&R 明确未满足的不受影响。 空间不足事项里写明有多少任务因 H&R 未知没清理、可在哪里放行。 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei --- .../services/downloads.ts | 17 +++++++++++++---- .../web/app/components/settings/instances.tsx | 19 +++++++++++++++---- packages/contracts/src/instances.ts | 2 ++ 3 files changed, 30 insertions(+), 8 deletions(-) diff --git a/apps/server/src/modules/acquisition-organize/services/downloads.ts b/apps/server/src/modules/acquisition-organize/services/downloads.ts index 47223fc9..bcb93a35 100644 --- a/apps/server/src/modules/acquisition-organize/services/downloads.ts +++ b/apps/server/src/modules/acquisition-organize/services/downloads.ts @@ -1272,8 +1272,9 @@ export async function cleanupCandidates(d: AoDeps, v: VolumeState): Promise th.cleanupTargetPercent && pct >= th.hardThresholdPercent) { entry.action = entry.removed.length > 0 ? "cleaned" : "insufficient"; // 没有可清理对象:待处理 + 阈值继续阻止新下载(不突破保留要求) + const hnrUnknown = entry.skipped.filter((x) => x.reason === "hnr_unknown").length; await upsertDecision(d, db, { kind: "space_insufficient", severity: "critical", title: `下载缓存卷 ${v.volumeKey} 空间不足`, - summary: `使用率 ${pct.toFixed(1)}%，可清理任务已用尽；新下载已暂停`, + summary: + `使用率 ${pct.toFixed(1)}%，可清理任务已用尽；新下载已暂停` + + (hnrUnknown + ? `。另有 ${hnrUnknown} 个任务因为 H&R 要求未知没有清理，确认没有 H&R 要求的话，可在下载器设置里打开「H&R 未知也可清理」` + : ""), subjectType: "cache_volume", dedupeKey: `space_insufficient:${v.volumeKey}`, evidence: { usedPercent: pct, locked: entry.skipped }, diff --git a/apps/web/app/components/settings/instances.tsx b/apps/web/app/components/settings/instances.tsx index f017896a..e9f6b474 100644 --- a/apps/web/app/components/settings/instances.tsx +++ b/apps/web/app/components/settings/instances.tsx @@ -97,16 +97,20 @@ const POLICY_FIELDS = [ ["targetPct", "cleanupTargetPercent"], ["seedDays", "seedRetentionDays"], ] as const; -function policyForm(r: DownloaderInstance): Record { +function policyForm(r: DownloaderInstance): Record { const p = ((r.settings as { policy?: Record }).policy ?? {}) as Record; - return Object.fromEntries(POLICY_FIELDS.map(([f, k]) => [f, typeof p[k] === "number" ? String(p[k]) : ""])); + return { + ...Object.fromEntries(POLICY_FIELDS.map(([f, k]) => [f, typeof p[k] === "number" ? String(p[k]) : ""])), + hnrUnknownCleanable: p.hnrUnknownCleanable === true, + }; } -function policyBody(v: FormValues): Record { - const out: Record = {}; +function policyBody(v: FormValues): Record { + const out: Record = {}; for (const [f, k] of POLICY_FIELDS) { const raw = String(v[f] ?? "").trim(); if (raw !== "" && Number.isFinite(Number(raw))) out[k] = Number(raw); } + if (v.hnrUnknownCleanable === true) out.hnrUnknownCleanable = true; return out; } @@ -202,6 +206,12 @@ function DownloadersTable() { placeholder: "留空用全局", hint: "入库后至少保留几天再允许清理；公网 BT 可填 0（站点 H&R 要求仍然遵守）", }, + { + name: "hnrUnknownCleanable", + label: "H&R 未知也可清理", + kind: "switch", + hint: "保种天数到期后，没给出 H&R 要求的任务（如 M-Team、接管的任务、站点没配 H&R）也允许清理；默认不清理", + }, { name: "isDefault", label: "设为默认下载器", @@ -223,6 +233,7 @@ function DownloadersTable() { resumePct: "", targetPct: "", seedDays: "", + hnrUnknownCleanable: false, isDefault: false, enabled: true, }, diff --git a/packages/contracts/src/instances.ts b/packages/contracts/src/instances.ts index 1be15e8d..2c65dea0 100644 --- a/packages/contracts/src/instances.ts +++ b/packages/contracts/src/instances.ts @@ -68,6 +68,8 @@ export const DownloaderPolicy = z resumePercent: z.number().min(1).max(100).optional(), cleanupTargetPercent: z.number().min(1).max(100).optional(), seedRetentionDays: z.number().int().min(0).max(365).optional(), + /** 保种天数到期、但站点 H&R 要求未知(M-Team、接管任务、站点没配 H&R)时也允许清理;缺省不允许 */ + hnrUnknownCleanable: z.boolean().optional(), }) .strict(); export type DownloaderPolicy = z.infer; From 19f7bd0330083125b90b24eda939d6d4f9940ed1 Mon Sep 17 00:00:00 2001 From: root Date: Sat, 3 Oct 2026 23:20:29 +0800 Subject: [PATCH 4/5] =?UTF-8?q?fix(download):=20=E4=B8=8B=E8=BD=BD?= =?UTF-8?q?=E5=99=A8=E7=A6=BB=E7=BA=BF=E4=B8=8E=E4=B8=8B=E8=BD=BD=E7=9B=AE?= =?UTF-8?q?=E5=BD=95=E8=AF=BB=E4=B8=8D=E5=88=B0=E5=AE=B9=E9=87=8F=E5=BC=80?= =?UTF-8?q?=E5=BE=85=E5=A4=84=E7=90=86=EF=BC=8C=E5=AD=A3=E5=8C=85=E6=9A=82?= =?UTF-8?q?=E5=81=9C=E6=B7=BB=E5=8A=A0=E5=90=8E=E9=80=89=E6=96=87=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei --- .../src/modules/acquisition-organize/index.ts | 18 +- .../services/download-board.ts | 6 +- .../services/downloads.ts | 397 +++++++++++-- .../download-visibility.test.ts | 549 ++++++++++++++++++ .../app/components/downloads/plan-format.ts | 1 + docs/runbook/schedules.md | 4 +- packages/adapters/src/qbittorrent/client.ts | 6 + packages/contracts/src/download-board.ts | 3 +- 8 files changed, 924 insertions(+), 60 deletions(-) create mode 100644 apps/server/test/acquisition-organize/download-visibility.test.ts diff --git a/apps/server/src/modules/acquisition-organize/index.ts b/apps/server/src/modules/acquisition-organize/index.ts index 507f2183..5f482143 100644 --- a/apps/server/src/modules/acquisition-organize/index.ts +++ b/apps/server/src/modules/acquisition-organize/index.ts @@ -15,7 +15,7 @@ import { registerNamingRoutes } from "./routes/naming.ts"; import { registerOrganizationRoutes } from "./routes/organization.ts"; import { registerRuleRoutes } from "./routes/rules.ts"; import { type AoDeps, automationBlock, enqueueSweep, enqueueTracked, timeSlot } from "./services/common.ts"; -import { SWEEP_SLOT_MS, sweepReservedPlans, syncDownloads } from "./services/downloads.ts"; +import { reportVolumeStates, SWEEP_SLOT_MS, sweepReservedPlans, syncDownloads } from "./services/downloads.ts"; import { recoverStalledRecords } from "./services/organize-records.ts"; import { sweepDueSlots, sweepMetadataRefresh, sweepSubscriptionSearch } from "./services/subscriptions.ts"; import { organizeDecisionHook } from "./services/takeover.ts"; @@ -139,11 +139,17 @@ export function createAcquisitionOrganizeModule(overrides: AoOverrides = {}): Se }, run: guarded(d, async (d) => { const cache = await d.ctx.settings.get("cache"); - const over = (await loadVolumes(d)).some( - (v) => (usedPercent(v) ?? 0) >= volumeThresholds(v, cache).hardThresholdPercent, - ); - if (!over) return "未超阈值"; - return (await enqueueThresholdCleanup(d)) ? "已入队清理" : "清理进行中"; + const vols = await loadVolumes(d); + // 读不到容量的卷不当 0%:开待处理(同步每轮也会判定),摘要里写明 + await reportVolumeStates(d, vols); + const unknown = vols.filter((v) => usedPercent(v) === null).length; + const note = unknown ? `，${unknown} 个下载目录读不到容量` : ""; + const over = vols.some((v) => { + const pct = usedPercent(v); + return pct !== null && pct >= volumeThresholds(v, cache).hardThresholdPercent; + }); + if (!over) return `未超阈值${note}`; + return ((await enqueueThresholdCleanup(d)) ? "已入队清理" : "清理进行中") + note; }), }); registerScheduledJob(ctx, { diff --git a/apps/server/src/modules/acquisition-organize/services/download-board.ts b/apps/server/src/modules/acquisition-organize/services/download-board.ts index c18bbbdc..ef690d9a 100644 --- a/apps/server/src/modules/acquisition-organize/services/download-board.ts +++ b/apps/server/src/modules/acquisition-organize/services/download-board.ts @@ -124,7 +124,7 @@ export function instanceLive(d: AoDeps, inst: DownloaderInstanceRow): Promise { + // 复核计划仍是 reserved(取消接口同样锁计划行):qB 回查期间被取消的,不把它写回 submitted + const cur = await trx + .selectFrom("acquisition_plan") + .select("status") + .where("id", "=", plan.id) + .forUpdate() + .executeTakeFirst(); + if (cur?.status !== "reserved") { + lostPlan = cur?.status ?? "missing"; + return; + } if (res.kind === "rejected" || res.kind === "source_failed") { const r = res; await failPlanTx(d, trx, plan.id, plan.slot_id, `${r.kind}: ${r.reason}`); @@ -402,8 +425,9 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?: tags: r.tags, managed: true, origin: r.kind === "already_exists" ? "adopted" : "submitted", - status: r.kind === "accepted_unconfirmed" ? "submitting" : "queued", - pause_reason: null, + status: + r.kind === "accepted_unconfirmed" ? "submitting" : pack && r.kind === "added" ? "paused" : "queued", + pause_reason: pack && r.kind === "added" ? "policy" : null, removed_at: null, completed_at: null, no_progress_since: null, @@ -420,12 +444,13 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?: 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) + save_path, content_path, category, tags, managed, origin, status, pause_reason, size_bytes, next_check_at) values (${topo.inst.id}, ${r.hash}, ${r.hash.length === 40 ? r.hash : null}, ${r.hash.length === 64 ? r.hash : null}, ${plan.id}, ${ev.manual ? plan.id : null}, ${rel.id}, coalesce((select media_entity_id from acquisition_slot where id = ${plan.slot_id}), ${ev.mediaEntityId ?? rel.media_entity_id}::uuid), ${r.name}, ${r.savePath}, ${r.contentPath}, ${topo.category}, ${r.tags}, true, ${r.kind === "already_exists" ? "adopted" : "submitted"}, - ${r.kind === "accepted_unconfirmed" ? "submitting" : "queued"}, ${rel.size_bytes}, now()) + ${r.kind === "accepted_unconfirmed" ? "submitting" : pack && r.kind === "added" ? "paused" : "queued"}, + ${pack && r.kind === "added" ? "policy" : null}, ${rel.size_bytes}, now()) on conflict (instance_id, torrent_hash) do update set acquisition_plan_id = coalesce(download_task.acquisition_plan_id, excluded.acquisition_plan_id), import_plan_id = coalesce(excluded.import_plan_id, download_task.import_plan_id), @@ -435,9 +460,17 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?: returning id, (xmax = 0) as inserted`.execute(trx); dtId = ins.rows[0]!.id; } + // 暂停添加的季包:选文件挂起,选完(或元数据超时整包)由提交后续步骤 / 同步恢复下载 + const pendingSel = pack && r.kind !== "already_exists"; await trx .updateTable("acquisition_plan") - .set({ status: "submitted", status_reason: r.kind }) + .set({ + status: "submitted", + status_reason: r.kind, + ...(pendingSel + ? { file_selection: jsonb({ mode: "pending", since: d.ports.now().toISOString(), attempts: 0 }) } + : {}), + }) .where("id", "=", plan.id) .execute(); await applySlotEvent( @@ -456,14 +489,18 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?: }, }, ); + if (lostPlan) { + await undoLostSubmit(d, client, result, tag, tagAdded, plan.id, lostPlan); + return { status: "skipped", reason: `plan_${lostPlan}` }; + } 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) { - await selectPackFiles(d, client, hash, plan, rel).catch((err) => - d.ctx.logger.warn({ planId: plan.id, err: { message: errMessage(err) } }, "季包选文件失败，按整包下载"), - ); + // 季包/多集包:按缺集选文件再恢复下载(接管的已有任务不改文件选择);文件清单还拿不到(磁力元数据未就绪) + // 时留给同步重试,失败不影响提交结果 + if (pack && result.kind !== "already_exists") { + const fresh = await db.selectFrom("acquisition_plan").selectAll().where("id", "=", plan.id).executeTakeFirst(); + if (fresh && isPendingSelection(fresh)) await settlePackFiles(d, client, hash, fresh, rel); } if (result.kind !== "accepted_unconfirmed") await afterDownloadAccepted(d, plan.id); if (!dtId) { @@ -495,11 +532,150 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?: } const PACK_FILE_LIST = { retries: 3, intervalMs: 300 } as const; +/** 季包文件清单(磁力元数据)最多等多久;超时仍拿不到就整包下载并记录原因 */ +export const PACK_METADATA_TIMEOUT_MS = 10 * 60_000; + +/** 资源是季包(没有集号也没有日期)或多集包 */ +function isPackRelease(rel: SourceReleaseRow): boolean { + const parsed = parsedOf(rel); + const isPack = !parsed.episodes && !parsed.airDate; + const multi = !!parsed.episodes && parsed.episodes.end > parsed.episodes.start; + return isPack || multi; +} + +/** 订阅计划下季包 / 多集包、还没选过文件:提交时暂停添加,选完文件再恢复 */ +function needsPackSelection(plan: AcquisitionPlanRow, rel: SourceReleaseRow): boolean { + return !!plan.slot_id && plan.unit_keys.length > 0 && plan.file_selection === null && isPackRelease(rel); +} + +interface PendingSelection { + mode: "pending"; + since: string; + attempts: number; +} + +const isPendingSelection = (plan: Pick) => + (plan.file_selection as { mode?: string } | null)?.mode === "pending"; /** - * 季包按缺集选文件(对照 MP 从整季包按集选文件,仅 qB):提交后读文件清单,把不需要单元的文件优先级置 0; + * 暂停添加的季包:读文件清单 → 选文件 → 恢复下载 → 记录选择。 + * 文件清单还拿不到(磁力元数据未就绪)且没到 PACK_METADATA_TIMEOUT_MS:只累加重试次数,留给下一轮同步; + * 超时仍拿不到:整包下载并记 file_list_timeout。恢复下载失败时选择保持 pending,下一轮同步重做(选文件幂等)。 + * 返回 true 表示已收尾(不再 pending)。 + */ +async function settlePackFiles( + d: AoDeps, + client: DownloaderClient, + hash: string, + plan: AcquisitionPlanRow, + rel: SourceReleaseRow, +): Promise { + const pending = plan.file_selection as unknown as PendingSelection; + const now = d.ports.now(); + const waitedMs = now.getTime() - new Date(pending.since).getTime(); + const files = await client.files(hash, PACK_FILE_LIST).catch(() => [] as QbFile[]); + const record = (sel: Record, unitKeys?: string[]) => + d.ctx.db + .updateTable("acquisition_plan") + .set({ file_selection: jsonb(sel), ...(unitKeys ? { unit_keys: unitKeys } : {}) }) + .where("id", "=", plan.id) + .execute(); + if (files.length === 0 && waitedMs < PACK_METADATA_TIMEOUT_MS) { + await record({ ...pending, attempts: pending.attempts + 1 }); + return false; + } + let sel: { selection: Record; unitKeys?: string[] }; + if (files.length === 0) { + sel = { + selection: { mode: "whole", reason: "file_list_timeout", waitedSec: Math.round(waitedMs / 1000) }, + }; + } else { + try { + sel = await selectPackFiles(d, client, hash, plan, rel, files); + } catch (err) { + d.ctx.logger.warn({ planId: plan.id, err: { message: errMessage(err) } }, "季包选文件失败，按整包下载"); + sel = { selection: { mode: "whole", reason: "select_failed", error: errMessage(err).slice(0, 300) } }; + } + } + try { + await client.resume([hash]); + } catch (err) { + d.ctx.logger.warn( + { planId: plan.id, err: { message: errMessage(err) } }, + "季包选完文件后恢复下载失败，下一轮同步重试", + ); + return false; + } + await record(sel.selection, sel.unitKeys); + return true; +} + +/** 同步时收尾挂起的季包选文件(本轮在 qB 里看得到的任务) */ +async function settlePendingPacks(d: AoDeps, client: DownloaderClient, topo: InstanceTopology, seen: Set) { + const rows = await d.ctx.db + .selectFrom("acquisition_plan as p") + .innerJoin("download_task as dt", "dt.acquisition_plan_id", "p.id") + .selectAll("p") + .select("dt.torrent_hash") + .where("dt.instance_id", "=", topo.inst.id) + .where("dt.removed_at", "is", null) + .where("dt.status", "!=", "removed") + .where("p.status", "=", "submitted") + .where(sql`p.file_selection->>'mode'`, "=", "pending") + .execute(); + for (const { torrent_hash, ...plan } of rows) { + if (!seen.has(torrent_hash)) continue; + try { + const rel = await d.ctx.db + .selectFrom("source_release") + .selectAll() + .where("id", "=", plan.source_release_id) + .executeTakeFirstOrThrow(); + await settlePackFiles(d, client, torrent_hash, plan as AcquisitionPlanRow, rel); + } catch (err) { + d.ctx.logger.warn({ planId: plan.id, err: { message: errMessage(err) } }, "季包选文件重试失败"); + } + } +} + +/** + * 提交期间计划被取消(commit 复核不是 reserved):任务行没落,qB 里刚加的任务暂停、去掉管理标签、 + * 打上 <管理标签>-cancelled 留给人删(不自动删数据);原本就在 qB 里的任务只撤掉本次补上的管理标签。 + */ +async function undoLostSubmit( + d: AoDeps, + client: DownloaderClient, + result: SubmitResult, + tag: string, + tagAdded: boolean, + planId: string, + planStatus: string, +) { + if (!("hash" in result)) return; + try { + if (result.kind === "already_exists") { + if (tagAdded) await client.removeTags([result.hash], [tag]); + } else { + await client.pause([result.hash]); + await client.removeTags([result.hash], [tag]); + await client.addTags([result.hash], [`${tag}-cancelled`]); + } + d.ctx.logger.warn( + { planId, planStatus, hash: result.hash, kind: result.kind }, + "提交期间计划已结束，已撤下 qB 里的任务（暂停并去掉管理标签）", + ); + } catch (err) { + d.ctx.logger.error( + { planId, planStatus, hash: result.hash, err: { message: errMessage(err) } }, + "提交期间计划已结束，撤下 qB 任务失败，需要到 qB 里手动处理", + ); + } +} + +/** + * 季包按缺集选文件(对照 MP 从整季包按集选文件,仅 qB):按文件清单把不需要单元的文件优先级置 0; * 需要单元 = 计划预留的集(季通配 = 该季未覆盖、且不在其它在途计划里的集)。 - * 文件清单拿不到 / 有视频文件认不出集号 → 整包下载并记录理由;计划的 unit_keys 收敛为实际选中的集。 + * 有视频文件认不出集号 → 整包下载并记录理由;计划的 unit_keys 收敛为实际选中的集。返回要记录的选择(由调用方写)。 */ async function selectPackFiles( d: AoDeps, @@ -507,20 +683,12 @@ async function selectPackFiles( hash: string, plan: AcquisitionPlanRow, rel: SourceReleaseRow, -): Promise { - if (!plan.slot_id) return; + files: QbFile[], +): Promise<{ selection: Record; unitKeys?: string[] }> { const parsed = parsedOf(rel); - const isPack = !parsed.episodes && !parsed.airDate; - const multi = !!parsed.episodes && parsed.episodes.end > parsed.episodes.start; - if (!isPack && !multi) return; - const record = (sel: Record, unitKeys?: string[]) => - d.ctx.db - .updateTable("acquisition_plan") - .set({ file_selection: jsonb(sel), ...(unitKeys ? { unit_keys: unitKeys } : {}) }) - .where("id", "=", plan.id) - .execute(); - const files = await client.files(hash, PACK_FILE_LIST).catch(() => []); - if (files.length === 0) return void (await record({ mode: "whole", reason: "file_list_unavailable" })); + const whole = (selection: Record) => ({ selection: { mode: "whole", ...selection } }); + const slotId = plan.slot_id; + if (!slotId) return whole({ reason: "no_slot" }); // 季通配:展开为"该季未覆盖且不在其它在途计划里" const wildSeasons = new Set( @@ -535,7 +703,7 @@ async function selectPackFiles( const slot = await d.ctx.db .selectFrom("acquisition_slot") .select(["id", "current_asset_id", "media_entity_id"]) - .where("id", "=", plan.slot_id) + .where("id", "=", slotId) .executeTakeFirstOrThrow(); const slots = await d.ctx.db .selectFrom("acquisition_slot") @@ -543,7 +711,7 @@ async function selectPackFiles( .where("media_entity_id", "=", slot.media_entity_id) .execute(); const covered = await slotCoveredUnits(d.ctx.db, slot, { entityWide: slots.length <= 1 }); - const others = await slotInflightPlans(d.ctx.db, plan.slot_id, plan.id); + const others = await slotInflightPlans(d.ctx.db, slotId, plan.id); blocked = new Set([...covered.keys, ...others.flatMap((o) => o.unitKeys)]); } const wantedKey = (k: string) => { @@ -588,13 +756,12 @@ async function selectPackFiles( const videos = rows.filter((r) => r.kind === "video"); const unmapped = videos.filter((r) => r.keys.length === 0 && !r.special); if (videos.length === 0 || unmapped.length > 0) - return void (await record({ - mode: "whole", + return whole({ reason: videos.length === 0 ? "no_video_files" : "unmapped_video_file", files: unmapped.slice(0, 20).map((r) => r.name), - })); + }); const selected = [...new Set(videos.filter((r) => r.wanted).flatMap((r) => r.keys.filter(wantedKey)))]; - if (selected.length === 0) return void (await record({ mode: "whole", reason: "no_missing_units_in_pack" })); + if (selected.length === 0) return whole({ reason: "no_missing_units_in_pack" }); // 视频不需要 → 不下;附属文件(字幕等)有集号且不需要 → 不下;无集号的附属文件保留 const skip = rows.filter((r) => (r.kind === "video" ? !r.wanted : r.keys.length > 0 && !r.wanted)); if (skip.length > 0) @@ -603,15 +770,15 @@ async function selectPackFiles( skip.map((r) => r.index), 0, ); - await record( - { + return { + selection: { mode: skip.length > 0 ? "selected" : "all_needed", selectedUnits: selected, skippedFiles: skip.slice(0, 200).map((r) => r.name), totalFiles: files.length, }, - selected, - ); + unitKeys: selected, + }; } async function failPlanTx(d: AoDeps, trx: Exec, planId: string, slotId: string | null, reason: string) { @@ -716,8 +883,16 @@ export interface SyncStats { removed: number; thresholdPaused: number; thresholdResumed: number; + /** 本轮首次在 qB 里确认的「已收单未确认」提交的计划(同步后补调 afterDownloadAccepted) */ + confirmedPlans: string[]; } +/** 下载器连续不可达多久开 downloader_offline 待处理 */ +export const DOWNLOADER_OFFLINE_DECISION_MS = 10 * 60_000; + +/** 在途(还没下完)的任务状态:停用的实例上有这些任务时仍同步 */ +const INFLIGHT_STATUSES: DownloadStatus[] = ["queued", "submitting", "downloading", "stalled", "paused"]; + /** 连续多少轮同步(列表 + 按 hash 查)都查不到才判 removed;与资产对账的 missingConfirmAfter 同口径 */ export const MISSING_CONFIRM_ROUNDS = 3; @@ -731,23 +906,44 @@ export async function syncDownloads(d: AoDeps): Promise { removed: 0, thresholdPaused: 0, thresholdResumed: 0, + confirmedPlans: [], }; const insts = await loadInstances(d.ctx.db); for (const topo of insts) { stats.instances++; await syncInstance(d, topo, stats); } + // 停用的实例上还有在途任务:只同步这些任务的状态(不接管新任务、不开离线事项、不计入失联统计) + const enabled = new Set(insts.map((t) => t.inst.id)); + for (const topo of await loadInstances(d.ctx.db, { enabledOnly: false })) { + if (enabled.has(topo.inst.id)) continue; + const inflight = await d.ctx.db + .selectFrom("download_task") + .select("id") + .where("instance_id", "=", topo.inst.id) + .where("removed_at", "is", null) + .where("managed", "=", true) + .where("status", "in", INFLIGHT_STATUSES) + .executeTakeFirst(); + if (inflight) await syncInstance(d, topo, stats, { observeOnly: true }); + } + for (const planId of stats.confirmedPlans) { + await afterDownloadAccepted(d, planId).catch((err) => + d.ctx.logger.warn({ planId, err: { message: errMessage(err) } }, "补记订阅进度失败"), + ); + } await enforceThresholds(d, insts, stats); return stats; } -async function syncInstance(d: AoDeps, topo: InstanceTopology, stats: SyncStats) { +async function syncInstance(d: AoDeps, topo: InstanceTopology, stats: SyncStats, opts: { observeOnly?: boolean } = {}) { const db = d.ctx.db; const now = d.ports.now(); - const client = await d.ports.downloader(topo.inst); - const sourceTags = await qbSourceTags(d, topo); let list: QbTorrent[]; + let client: DownloaderClient; try { + client = await d.ports.downloader(topo.inst); + const sourceTags = await qbSourceTags(d, topo); const tags = [...new Set([topo.inst.managed_tag, ...sourceTags.map((s) => s.config.tag!)])]; const batches = await Promise.all(tags.map((tag) => client.list({ tag }))); list = [...new Map(batches.flat().map((t) => [t.hash, t])).values()].filter( @@ -755,13 +951,15 @@ async function syncInstance(d: AoDeps, topo: InstanceTopology, stats: SyncStats) ); } catch (err) { // 下载器失联:不改任何任务状态(失败 ≠ 不存在) - stats.unreachable.push(topo.inst.id); await db .updateTable("downloader_instance") .set({ health_status: "offline" }) .where("id", "=", topo.inst.id) .execute(); d.ctx.logger.warn({ instanceId: topo.inst.id, err: { message: errMessage(err) } }, "qB 同步失败"); + if (opts.observeOnly) return; + stats.unreachable.push(topo.inst.id); + await reportDownloaderOffline(d, topo, now, err); return; } await db @@ -769,6 +967,7 @@ async function syncInstance(d: AoDeps, topo: InstanceTopology, stats: SyncStats) .set({ health_status: "healthy", last_seen_at: now }) .where("id", "=", topo.inst.id) .execute(); + await closeRecoveredDecision(d, db, `downloader_offline:${topo.inst.id}`, "下载器已恢复连接"); const rows = await db .selectFrom("download_task") .selectAll() @@ -783,10 +982,11 @@ async function syncInstance(d: AoDeps, topo: InstanceTopology, stats: SyncStats) const row = byHash.get(t.hash); if (row) { if (await observe(d, row, t, now, stats)) stats.updated++; - } else { + } else if (!opts.observeOnly) { await adopt(d, topo, t, now, stats); } } + await settlePendingPacks(d, client, topo, seen); for (const row of rows) { if (seen.has(row.torrent_hash) || !row.managed) continue; let still: QbTorrent | null; @@ -920,9 +1120,21 @@ async function observeLocked( ); // 卡住 / 异常后恢复正常(进度增长或状态恢复):关闭同一任务未关闭的无进展 / 异常事项 if (changed && RECOVERED.has(merged.status)) { - if (row.status === "stalled") await closeRecoveredDecision(d, trx, `download_stalled:${row.id}`, merged.status); + const note = `下载已恢复（${merged.status}）`; + if (row.status === "stalled") await closeRecoveredDecision(d, trx, `download_stalled:${row.id}`, note); if (row.status === "failed" || row.status === "missing") - await closeRecoveredDecision(d, trx, `download_failed:${row.id}`, merged.status); + await closeRecoveredDecision(d, trx, `download_failed:${row.id}`, note); + } + // 「已收单未确认」的提交第一次在 qB 里看到:计划改记已确认,同步后补调 afterDownloadAccepted(订阅进度 / 提前完成) + if (row.status === "submitting" && row.acquisition_plan_id) { + const confirmed = await trx + .updateTable("acquisition_plan") + .set({ status_reason: "accepted_confirmed" }) + .where("id", "=", row.acquisition_plan_id) + .where("status_reason", "=", "accepted_unconfirmed") + .returning("id") + .executeTakeFirst(); + if (confirmed) stats.confirmedPlans.push(confirmed.id); } if (becameComplete) { stats.completed.push(row.id); @@ -974,7 +1186,7 @@ const RECOVERED = new Set(["queued", "downloading", "completed", * 卡住 / 异常事项自动关闭:只关 open / awaiting_user;Agent 正在调查(investigating)的由它自己核实后关闭。 * resolution 不带处理选项,executeResolution 不会再执行任何动作。 */ -async function closeRecoveredDecision(d: AoDeps, trx: Exec, dedupeKey: string, status: DownloadStatus) { +async function closeRecoveredDecision(d: AoDeps, trx: Exec, dedupeKey: string, note: string) { const by = "system:download-sync"; const rows = await trx .updateTable("decision") @@ -982,7 +1194,7 @@ async function closeRecoveredDecision(d: AoDeps, trx: Exec, dedupeKey: string, s status: "resolved", resolved_by: by, resolved_at: d.ports.now(), - resolution: jsonb({ key: "download_recovered", optionKey: null, by, note: `下载已恢复（${status}）` }), + resolution: jsonb({ key: "download_recovered", optionKey: null, by, note }), }) .where("dedupe_key", "=", dedupeKey) .where("status", "in", ["open", "awaiting_user"]) @@ -998,6 +1210,82 @@ async function closeRecoveredDecision(d: AoDeps, trx: Exec, dedupeKey: string, s } } +/** 同一 dedupeKey 已有未关闭的事项(巡检类事项每轮都会判定,已开着就不再刷新、不累加次数) */ +async function hasOpenDecision(exec: Exec, dedupeKey: string): Promise { + const r = await exec + .selectFrom("decision") + .select("id") + .where("dedupe_key", "=", dedupeKey) + .where("status", "in", ["open", "investigating", "awaiting_user"]) + .executeTakeFirst(); + return !!r; +} + +/** 下载器连续不可达超过 DOWNLOADER_OFFLINE_DECISION_MS:按实例开一条待处理,恢复连接后由同步关闭 */ +async function reportDownloaderOffline(d: AoDeps, topo: InstanceTopology, now: Date, err: unknown) { + const since = topo.inst.last_seen_at ?? topo.inst.created_at; + const offlineMs = now.getTime() - since.getTime(); + if (offlineMs < DOWNLOADER_OFFLINE_DECISION_MS) return; + const dedupeKey = `downloader_offline:${topo.inst.id}`; + if (await hasOpenDecision(d.ctx.db, dedupeKey)) return; + await upsertDecision(d, d.ctx.db, { + kind: "downloader_offline", + severity: "critical", + title: `下载器连不上：${topo.inst.name}`, + summary: + `${topo.inst.last_seen_at ? `已 ${Math.round(offlineMs / 60_000)} 分钟` : "添加后一直"}连不上 ${topo.inst.base_url}，` + + "新下载提交不了，在下的任务状态也不更新。检查下载器是否在运行、地址和登录信息是否正确", + subjectType: "downloader_instance", + subjectId: topo.inst.id, + dedupeKey, + evidence: { + instanceId: topo.inst.id, + baseUrl: topo.inst.base_url, + lastSeenAt: topo.inst.last_seen_at?.toISOString() ?? null, + error: errMessage(err).slice(0, 500), + }, + }); +} + +/** + * 下载目录读不到容量(statfs 失败、保存路径没有路径映射):按卷开 critical 待处理,读得到后关闭。 + * 读不到时新下载一律延后(不能把"未知"当"有空间"),所以这是配置问题(路径映射 / 容器挂载),不是空间问题。 + */ +export async function reportVolumeStates(d: AoDeps, vols: VolumeState[]): Promise { + for (const v of vols) { + const dedupeKey = `cache_volume_unknown:${v.volumeKey}`; + if (v.snapshot) { + await closeRecoveredDecision(d, d.ctx.db, dedupeKey, "下载目录已能读到容量"); + continue; + } + if (await hasOpenDecision(d.ctx.db, dedupeKey)) continue; + const usage = v.usage && v.usage.status !== "ok" ? (v.usage as { code?: string; message?: string }) : null; + const savePaths = v.instances.map((t) => ({ + instance: t.inst.name, + savePath: t.savePath, + localPath: t.localSavePath, + })); + const why = v.path + ? `读不到本机路径 ${v.path} 的容量${usage?.code ? `（${usage.code}）` : ""}，检查容器是否挂载了下载目录` + : "下载器的保存路径没法换算成本机路径，检查下载器的保存路径和路径映射"; + await upsertDecision(d, d.ctx.db, { + kind: "cache_volume_unknown", + severity: "critical", + title: `下载目录读不到容量：${v.instances.map((t) => t.inst.name).join("、")}`, + summary: `${why}。读到之前新下载一律延后`, + subjectType: "cache_volume", + dedupeKey, + evidence: { + volumeKey: v.volumeKey, + path: v.path, + reason: v.path ? "statfs_failed" : "save_path_unmapped", + usage: usage ? { code: usage.code ?? null, message: usage.message ?? null } : null, + instances: savePaths, + }, + }); + } +} + /** 用户给已有任务贴了管理标签:登记接管,不改保存路径与其它标签 */ async function adopt(d: AoDeps, topo: InstanceTopology, t: QbTorrent, now: Date, stats: SyncStats) { const db = d.ctx.db; @@ -1092,20 +1380,31 @@ async function adopt(d: AoDeps, topo: InstanceTopology, t: QbTorrent, now: Date, async function enforceThresholds(d: AoDeps, insts: InstanceTopology[], stats: SyncStats) { if (insts.length === 0) return; + const vols = await loadVolumes(d, d.ctx.db, insts); + await reportVolumeStates(d, vols); // 只读对账模式禁止自动写入(含对 qB 的暂停/恢复) if ((await d.ctx.settings.get("automation")).readOnlyReconcile) return; const cache = await d.ctx.settings.get("cache"); - const vols = await loadVolumes(d, d.ctx.db, insts); for (const v of vols) { const dv = decideVolume(v, cache); - if (dv.decision === "pause_managed" && dv.pauseTaskIds.length > 0) { - const n = await pauseForThreshold(d, v, dv.pauseTaskIds); + // 排队中(queuedDL)的纳管任务一并暂停:否则其它任务被暂停后 qB 的队列会把它拉起来 + const pauseIds = + dv.decision === "pause_managed" + ? [ + ...new Set([ + ...dv.pauseTaskIds, + ...v.inflight.filter((t) => t.managed && t.status === "queued").map((t) => t.taskId), + ]), + ] + : []; + if (pauseIds.length > 0) { + const n = await pauseForThreshold(d, v, pauseIds); stats.thresholdPaused += n; await d.ctx.outbox.emit(d.ctx.db, "download.threshold", { volumeKey: v.volumeKey, state: "tripped", usedPercent: dv.usedPct === undefined ? null : Math.min(100, dv.usedPct), - pausedTaskIds: dv.pauseTaskIds, + pausedTaskIds: pauseIds, }); } else if (dv.resumeThresholdPaused) { const ids = await resumeThresholdPaused(d, v); diff --git a/apps/server/test/acquisition-organize/download-visibility.test.ts b/apps/server/test/acquisition-organize/download-visibility.test.ts new file mode 100644 index 00000000..b564485c --- /dev/null +++ b/apps/server/test/acquisition-organize/download-visibility.test.ts @@ -0,0 +1,549 @@ +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(); + /** releaseId → 取种子时执行的钩子(模拟提交过程中发生的事) */ + const onSource = new Map Promise>(); + let base: Awaited>; + 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(); + } + }); +}); diff --git a/apps/web/app/components/downloads/plan-format.ts b/apps/web/app/components/downloads/plan-format.ts index 23b2d1e2..88687f99 100644 --- a/apps/web/app/components/downloads/plan-format.ts +++ b/apps/web/app/components/downloads/plan-format.ts @@ -10,6 +10,7 @@ export const PURPOSE_LABEL: Record = { fill: "补缺", upgr export const PLAN_WAIT: Record = { approval: { label: "等你批准", tone: "warning" }, space: { label: "等磁盘空间", tone: "warning" }, + volume_unknown: { label: "下载目录读不到", tone: "warning" }, automation_off: { label: "等开启自动化", tone: "warning" }, queued: { label: "准备提交", tone: "neutral" }, }; diff --git a/docs/runbook/schedules.md b/docs/runbook/schedules.md index 285351f5..0063384f 100644 --- a/docs/runbook/schedules.md +++ b/docs/runbook/schedules.md @@ -21,12 +21,12 @@ | `sites.snapshot` | 站点快照 | 采集 | 30 秒 | 扫描到周期的站点，入队 `snapshotSite` | | `identity.review` | 待识别复查 | 采集 | 30 秒 | 对 `next_review_at` 已到的条目入队 `resolveObservation` | | `subscriptions.search` | 订阅周期搜索 | 采集 | 同 qB 同步 | `sweepSubscriptionSearch` | -| `downloads.sync` | qB 同步 | 下载 | `cache.checkIntervalSec`(60 秒) | `syncDownloads`，包括按阈值暂停和恢复。所有下载器都连不上时记为失败 | +| `downloads.sync` | qB 同步 | 下载 | `cache.checkIntervalSec`(60 秒) | `syncDownloads`，包括按阈值暂停和恢复（排队中的任务一起暂停）、季包选文件的重试（磁力元数据未就绪时最多等 10 分钟，超时整包下载）。所有下载器都连不上时记为失败；单个下载器连续 10 分钟连不上开 `downloader_offline` 待处理，下载目录读不到容量开 `cache_volume_unknown` 待处理，恢复后都自动关闭。停用的下载器上还有在下的任务时仍同步这些任务，不接管新任务 | | `downloads.orchestrate` | 补提交与补入库 | 下载 | 同上 | `sweepReservedPlans` 和补入队 `verifyAndImport` | | `slots.reevaluate` | 到期重评 | 下载 | 同上 | 处理 `next_evaluate_at` 已到的收录位 | | `subscriptions.metadata` | 总集数刷新 | 整理 | 同上 | `sweepMetadataRefresh` | | `library.reconcile` | 资产对账 | 整理 | 6 小时 | 入队 `reconcileLibrary`，幂等键是 `auto:<时间槽>` | -| `cache.threshold` | 缓存阈值检查 | 维护 | 同 qB 同步 | 超过硬阈值时入队 `cleanupDownloadCache` | +| `cache.threshold` | 缓存阈值检查 | 维护 | 同 qB 同步 | 超过硬阈值时入队 `cleanupDownloadCache`；读不到容量的下载目录不当 0%，摘要里写明并开待处理。H&R 要求未知的任务缺省不清理，下载器设置打开「H&R 未知也可清理」后保种天数到期即可清理 | | `trash.purge` | 回收站清理 | 维护 | 24 小时 | 入队 `purgeTrash` | | `pi.token_refresh` | 上游登录刷新 | 维护 | 10 分钟 | 入队 `refreshPiTokens`,幂等键 `refreshPiTokens:<时间槽>`;对 30 分钟内过期的订阅登录上游刷新 token | | `auth.cleanup` | 登录会话清理 | 维护 | 1 小时 | 直接执行 `cleanupAuthTables`:删 7 天前的登录尝试、30 天前过期或吊销的会话 | diff --git a/packages/adapters/src/qbittorrent/client.ts b/packages/adapters/src/qbittorrent/client.ts index 15cda299..223b362b 100644 --- a/packages/adapters/src/qbittorrent/client.ts +++ b/packages/adapters/src/qbittorrent/client.ts @@ -120,6 +120,11 @@ export interface AddTorrentRequest { category?: string; tags?: string[]; paused?: boolean; + /** + * qB ≥ 4.5 的停止条件:MetadataReceived = 磁力拿到元数据后自动暂停(暂停添加的磁力不会去取元数据); + * 旧版本忽略该参数、照常开始下载 + */ + stopCondition?: "MetadataReceived" | "FilesChecked"; rename?: string; contentLayout?: "Original" | "Subfolder" | "NoSubfolder"; sequential?: boolean; @@ -516,6 +521,7 @@ export class QbClient implements DownloaderClient { const paused = req.paused ? "true" : "false"; form.append("paused", paused); // <= 4.x form.append("stopped", paused); // >= 5.x + if (req.stopCondition) form.append("stopCondition", req.stopCondition); if (req.rename) form.append("rename", req.rename); form.append("contentLayout", req.contentLayout ?? "Original"); form.append("autoTMM", "false"); diff --git a/packages/contracts/src/download-board.ts b/packages/contracts/src/download-board.ts index 567a0ec3..b8510d86 100644 --- a/packages/contracts/src/download-board.ts +++ b/packages/contracts/src/download-board.ts @@ -70,7 +70,8 @@ export type DownloadBoardItem = z.infer; * automation_off = 自动化总开关关闭 / 只读对账;queued = 条件已满足、等提交任务执行。 * 每日下载额度在生成计划时核对,超额不生成计划,所以没有"等额度"的计划。 */ -export const PlanWait = z.enum(["approval", "space", "automation_off", "queued"]); +/** volume_unknown:下载目录读不到容量(配置问题,不是空间不够) */ +export const PlanWait = z.enum(["approval", "space", "volume_unknown", "automation_off", "queued"]); export type PlanWait = z.infer; /** 还没提交到下载器的计划,作为「进行中」的一行(id 是计划 id) */ From 9ae287beb2092289242549b051d3d28a0e82fbe2 Mon Sep 17 00:00:00 2001 From: root Date: Sat, 3 Oct 2026 23:30:18 +0800 Subject: [PATCH 5/5] =?UTF-8?q?test:=20volume=5Fstate=5Funknown=20?= =?UTF-8?q?=E5=8E=9F=E5=9B=A0=E6=96=87=E6=A1=88=E6=94=B9=E4=B8=BA=E3=80=8C?= =?UTF-8?q?=E4=B8=8B=E8=BD=BD=E7=9B=AE=E5=BD=95=E8=AF=BB=E4=B8=8D=E5=88=B0?= =?UTF-8?q?=E5=AE=B9=E9=87=8F=E3=80=8D=E5=90=8E=E5=90=8C=E6=AD=A5=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=E6=9C=9F=E6=9C=9B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei --- apps/web/app/components/agent/report-humanize.test.ts | 2 +- packages/contracts/test/reasons.test.ts | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/web/app/components/agent/report-humanize.test.ts b/apps/web/app/components/agent/report-humanize.test.ts index 255833c6..d8b2a2a2 100644 --- a/apps/web/app/components/agent/report-humanize.test.ts +++ b/apps/web/app/components/agent/report-humanize.test.ts @@ -64,7 +64,7 @@ describe("humanizeReport(汇报兜底)", () => { names, ); expect(r.text).toBe( - "检测站点 被拒；查看下载任务 无法调用（返回 参数不对）；准备下载 返回 plan=null，拒绝原因 下载盘用量未知", + "检测站点 被拒；查看下载任务 无法调用（返回 参数不对）；准备下载 返回 plan=null，拒绝原因 下载目录读不到容量", ); expect(r.text).not.toMatch(/site\.health_check|invalid_input|volume_state_unknown/); expect(humanizeReport("被规则拒绝（unauthorized），未重试", names).text).toBe("被规则拒绝（需要你批准），未重试"); diff --git a/packages/contracts/test/reasons.test.ts b/packages/contracts/test/reasons.test.ts index 6b87b593..6f04da44 100644 --- a/packages/contracts/test/reasons.test.ts +++ b/packages/contracts/test/reasons.test.ts @@ -20,7 +20,7 @@ describe("reasonText(DEAI-2 原因码人话化)", () => { it("智能体汇报里的码(FIX-7):能力错误、带分类前缀的空间判定、站点登录", () => { expect(reasonText("invalid_input")?.text).toBe("参数不对"); - expect(reasonText("cache:volume_state_unknown")?.text).toBe("下载盘用量未知"); + expect(reasonText("cache:volume_state_unknown")?.text).toBe("下载目录读不到容量"); expect(reasonText("deferred:no_downloader_instance")?.text).toBe("没有可用的下载器"); expect(reasonText("auth_missing")?.text).toContain("Cookie"); expect(reasonText("unidentifiable")?.text).toBe("没识别出作品");