From 4d24edd94c219d8a66b0611effdc37d86ab83991 Mon Sep 17 00:00:00 2001 From: root Date: Sat, 3 Oct 2026 20:46:40 +0800 Subject: [PATCH 1/3] =?UTF-8?q?fix(acquisition):=20=E9=9B=86=E6=95=B0?= =?UTF-8?q?=E5=BE=85=E7=A1=AE=E8=AE=A4=E7=9A=84=E8=B5=84=E6=BA=90=E5=9C=A8?= =?UTF-8?q?=E8=AE=A2=E9=98=85=E4=BE=A7=E5=86=99=E5=80=99=E9=80=89=E8=AE=B0?= =?UTF-8?q?=E5=BD=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 评估跳过 episode_unresolved 时，给仍在追踪的收录版本各写一条 skipped 候选评估 （原因 episode_unresolved），作品「为什么选 / 不选」与订阅记录能看到"集数待确认"。 确认后重发评估按 (收录版本, 资源) 覆盖；换算到别的季不再进入原收录版本时， 先清掉旧的待确认记录，不留陈旧原因。 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei --- .../services/acquisition.ts | 40 ++++++++++++++++- .../episode-resolution-evaluate.test.ts | 43 +++++++++++++++++++ 2 files changed, 82 insertions(+), 1 deletion(-) diff --git a/apps/server/src/modules/acquisition-organize/services/acquisition.ts b/apps/server/src/modules/acquisition-organize/services/acquisition.ts index c8b1929b..2fbb2c5e 100644 --- a/apps/server/src/modules/acquisition-organize/services/acquisition.ts +++ b/apps/server/src/modules/acquisition-organize/services/acquisition.ts @@ -34,6 +34,7 @@ import { hashJson, jsonb, startOfDay, toApi } from "../util.ts"; import { type AoDeps, automationBlock, + CANDIDATE_ARTIFACT_TITLE, downloaderPathOf, type Exec, enqueueTracked, @@ -364,7 +365,13 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon if (rel.availability !== "active") return skip(`release_${rel.availability}`); const entity = await resolveEntity(db, rel.media_entity_id); if (!entity || entity.status === "archived") return skip("entity_unavailable"); - if (entity.kind !== "movie" && episodeUnresolved(rel)) return skip("episode_unresolved"); + if (entity.kind !== "movie" && episodeUnresolved(rel)) { + // 订阅侧可见:相关收录版本各记一条候选评估;确认后重发 release.resolved 正常评估时按 (收录版本, 资源) 覆盖 + await noteEpisodeUnresolved(c, rel, entity.id, onlySlotId); + return skip("episode_unresolved"); + } + // 已确认的资源:先清掉之前的"集数待确认"记录(换算到别的季时下面可能不再进入该收录版本,不会被覆盖) + if (entity.kind !== "movie" && episodeResolutionOf(rel)) await clearEpisodeUnresolvedNotes(c, rel.id, entity.id); const parsed = parsedOf(rel); const units = await releaseUnits( db, @@ -523,6 +530,37 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon } } +/** 集数待确认:给仍在追踪的收录版本各写一条 skipped 候选评估(原因 episode_unresolved) */ +async function noteEpisodeUnresolved(c: Ctx, rel: SourceReleaseRow, entityId: string, onlySlotId?: string) { + let q = c.d.ctx.db + .selectFrom("acquisition_slot as s") + .leftJoin("subscription as sub", "sub.id", "s.subscription_id") + .select("s.id") + .where("s.media_entity_id", "=", entityId) + .where((eb) => eb.or([eb("s.subscription_id", "is", null), eb("sub.status", "in", [...TRACKING_SUB_STATUSES])])); + if (onlySlotId) q = q.where("s.id", "=", onlySlotId); + for (const s of await q.execute()) + await note(c, { + slotId: s.id, + sourceReleaseId: rel.id, + title: rel.title, + planId: null, + decision: "skipped", + mode: null, + rejected: [{ code: "episode_unresolved", message: "集数待确认，确认后自动重新评估" }], + reasons: ["episode_unresolved"], + }); +} + +async function clearEpisodeUnresolvedNotes(c: Ctx, sourceReleaseId: string, entityId: string) { + await sql` + delete from artifact + where title = ${CANDIDATE_ARTIFACT_TITLE} and subject_type = 'acquisition_slot' + and subject_id in (select id from acquisition_slot where media_entity_id = ${entityId}) + and content->>'sourceReleaseId' = ${sourceReleaseId} + and content->'reasons' ? 'episode_unresolved'`.execute(c.d.ctx.db); +} + /** 站点字段 → 评估上下文(做种数、促销下载系数、副标题/标签、站点 id、发布后分钟数) */ export function contextOf(rel: SourceReleaseRow, now: Date = new Date()): EvaluationContext { const promo = (rel.promotion ?? {}) as { kind?: string; raw?: { downloadFactor?: number | null } }; diff --git a/apps/server/test/acquisition-organize/episode-resolution-evaluate.test.ts b/apps/server/test/acquisition-organize/episode-resolution-evaluate.test.ts index 85c65d47..392a6c09 100644 --- a/apps/server/test/acquisition-organize/episode-resolution-evaluate.test.ts +++ b/apps/server/test/acquisition-organize/episode-resolution-evaluate.test.ts @@ -156,4 +156,47 @@ describe.skipIf(!hasPg)("订阅评估跳过集数待确认的资源", () => { expect(ev.plans).toHaveLength(0); expect(ev.skipped).toEqual([{ sourceReleaseId: id, reason: "episode_unresolved" }]); }); + + it("集数待确认在订阅侧可见:写候选记录 episode_unresolved;确认后正常评估并覆盖", async () => { + const slot = await h.db + .selectFrom("acquisition_slot") + .select("id") + .where("media_entity_id", "=", entityId) + .executeTakeFirstOrThrow(); + const candidate = async (releaseId: string) => + ( + await h.db + .selectFrom("artifact") + .select("content") + .where("title", "=", "candidate_evaluation") + .where("subject_id", "=", slot.id) + .execute() + ) + .map( + (r) => r.content as { sourceReleaseId: string; decision: string; reasons: string[]; planId: string | null }, + ) + .find((c) => c.sourceReleaseId === releaseId); + const id = await seed(LILITH_40); + await evaluateAcquisition(h.d, { sourceReleaseId: id }); + expect(await candidate(id)).toMatchObject({ decision: "skipped", reasons: ["episode_unresolved"] }); + + // 人工确认为 S02E12(与 `- 24` 的换算结果相同),重发评估:正常建计划,记录被覆盖 + const fixed = resolved("[Lilith-Raws] Kusuriya no Hitorigoto - 24 [Baha][WEB-DL][1080p][AVC AAC][CHT][MP4]"); + const cols = toParsedColumns(fixed.parsed); + await h.db + .updateTable("source_release") + .set({ + parsed: JSON.stringify(cols.parsed), + parser_version: cols.parserVersion, + episode_resolution: JSON.stringify(fixed.resolution), + }) + .where("id", "=", id) + .execute(); + const ev = await evaluateAcquisition(h.d, { sourceReleaseId: id }); + expect(ev.plans, JSON.stringify(ev)).toHaveLength(1); + const c = await candidate(id); + expect(c?.decision).toBe("accept"); + expect(c?.planId).toBe(ev.plans[0]); + expect(c?.reasons).not.toContain("episode_unresolved"); + }); }); From 8c10c12bd7e04891ddaf8e38dda2238cb9b8e621 Mon Sep 17 00:00:00 2001 From: root Date: Sat, 3 Oct 2026 20:47:23 +0800 Subject: [PATCH 2/3] =?UTF-8?q?fix(acquisition):=20=E9=97=B8=E9=97=A8?= =?UTF-8?q?=E6=8C=A1=E4=BD=8F=E7=9A=84=E8=B5=84=E6=BA=90=E5=86=99=E5=BB=B6?= =?UTF-8?q?=E5=90=8E=E5=B9=B6=E6=8E=92=E9=87=8D=E8=AF=84=EF=BC=9B=E5=B9=B6?= =?UTF-8?q?=E5=8F=91=E9=A2=84=E7=AE=97=E4=B8=8D=E8=AE=A1=20stalled?= =?UTF-8?q?=EF=BC=9B=E6=B8=85=E7=90=86=E7=A9=BA=E8=BD=AC=E5=BB=B6=E5=90=8E?= =?UTF-8?q?=E8=A1=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 自动化开关 / 下载器 / 预算 / 缓存卷挡住时，除候选记录（改为 deferred）外写 slot_deferral， 原因区分 gate:automation / gate:downloader / gate:budget / gate:cache，并排该收录版本重评； 电影收录版本与已满足的剧集收录版本也会被重评（scheduleSlotRetry，不推迟已排得更早的）。 - 重评时间按延后次数退避：30 分钟起翻倍、封顶 6 小时；预算释放 / 缓存卷恢复没有统一事件，不做事件触发。 - 并发预算：stalled 任务不占名额；已预留计划只算还没建下载任务的，不与任务重复计。 - 延后行清理：资源过期 / 不可评估 / 集数待确认时清该资源的延后；订阅完成（含 stopAfter）清其收录版本的延后； 订阅暂停 / 取消后重评跳过时清对应延后。 - continuous-series e2e：三集被接收后 once 订阅即完成、延后行随之清掉，改以候选评估记录断言"另一版本延后"。 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei --- .../services/acquisition.ts | 68 ++++- .../services/subscriptions.ts | 47 +++- .../blocked-reevaluate.test.ts | 249 ++++++++++++++++++ .../server/test/e2e/continuous-series.test.ts | 24 +- 4 files changed, 366 insertions(+), 22 deletions(-) create mode 100644 apps/server/test/acquisition-organize/blocked-reevaluate.test.ts diff --git a/apps/server/src/modules/acquisition-organize/services/acquisition.ts b/apps/server/src/modules/acquisition-organize/services/acquisition.ts index 2fbb2c5e..6ff90aff 100644 --- a/apps/server/src/modules/acquisition-organize/services/acquisition.ts +++ b/apps/server/src/modules/acquisition-organize/services/acquisition.ts @@ -48,9 +48,13 @@ import { autoFillCheck, type ForceFill } from "./evaluate-autofill.ts"; import { entityOriginalLanguage, loadRuleCatalog, mergeHard, type RuleCatalog, resolveRuleChain } from "./rules.ts"; import { clearDeferral, + clearReleaseDeferrals, + clearSubscriptionDeferrals, + gateRetryDelayMs, rangeOf, recordDeferral, scheduleSlotReevaluate, + scheduleSlotRetry, seasonLedgers, slotCoveredUnits, slotInflightPlans, @@ -246,6 +250,11 @@ export async function createSlotsFromTemplates( } } +/** + * 预算用量。并发 = 在跑的下载任务(不含 stalled:无进展任务不占名额,否则卡住几个就把追更全挡住) + * + 已预留但还没建下载任务的计划(刚预留、等补投;防止一轮评估连续预留超过并发上限)。 + * 已建下载任务的预留计划按任务计,不重复算。 + */ async function budgetUsage(exec: Exec, now: Date) { const day = startOfDay(now); const r = await sql<{ new_titles: number; bytes: number; reserved: number; active: number; reserved_count: number }>` @@ -254,8 +263,11 @@ async function budgetUsage(exec: Exec, now: Date) { and evaluation->>'mode' = 'new_title')::int as new_titles, (select coalesce(sum(reserved_bytes),0) from acquisition_plan where created_at >= ${day} and status in ('submitted','completed'))::bigint as bytes, (select coalesce(sum(reserved_bytes),0) from acquisition_plan where created_at >= ${day} and status = 'reserved')::bigint as reserved, - (select count(*) from download_task where managed and removed_at is null and status in ('queued','submitting','downloading','stalled'))::int as active, - (select count(*) from acquisition_plan where status = 'reserved')::int as reserved_count`.execute(exec); + (select count(*) from download_task where managed and removed_at is null and status in ('queued','submitting','downloading'))::int as active, + (select count(*) from acquisition_plan p where p.status = 'reserved' + and not exists (select 1 from download_task dt where dt.acquisition_plan_id = p.id and dt.removed_at is null))::int as reserved_count`.execute( + exec, + ); const row = r.rows[0]!; return { newTitlesToday: row.new_titles, @@ -361,14 +373,19 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon const { d, now } = c; const db = d.ctx.db; const skip = (reason: string) => c.out.skipped.push({ sourceReleaseId: rel.id, reason }); - if (!rel.media_entity_id || rel.resolution_status !== "resolved") return skip("release_unresolved"); - if (rel.availability !== "active") return skip(`release_${rel.availability}`); + // 不再参与评估的资源:延后行一并清掉,免得到期空转(L4) + const drop = async (reason: string) => { + await clearReleaseDeferrals(db, rel.id); + return skip(reason); + }; + if (!rel.media_entity_id || rel.resolution_status !== "resolved") return drop("release_unresolved"); + if (rel.availability !== "active") return drop(`release_${rel.availability}`); const entity = await resolveEntity(db, rel.media_entity_id); - if (!entity || entity.status === "archived") return skip("entity_unavailable"); + if (!entity || entity.status === "archived") return drop("entity_unavailable"); if (entity.kind !== "movie" && episodeUnresolved(rel)) { // 订阅侧可见:相关收录版本各记一条候选评估;确认后重发 release.resolved 正常评估时按 (收录版本, 资源) 覆盖 await noteEpisodeUnresolved(c, rel, entity.id, onlySlotId); - return skip("episode_unresolved"); + return drop("episode_unresolved"); } // 已确认的资源:先清掉之前的"集数待确认"记录(换算到别的季时下面可能不再进入该收录版本,不会被覆盖) if (entity.kind !== "movie" && episodeResolutionOf(rel)) await clearEpisodeUnresolvedNotes(c, rel.id, entity.id); @@ -470,6 +487,7 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon .set({ status: "completed", status_reason: "stop_after_reached" }) .where("id", "=", s.id) .execute(); + await clearSubscriptionDeferrals(db, s.id); s.status = "completed"; } } @@ -484,7 +502,11 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon for (const slot of slots) { if (onlySlotId && slot.id !== onlySlotId) continue; const sub = slot.subscription_id ? activeSubs.get(slot.subscription_id) : undefined; - if (slot.subscription_id && !sub) continue; // 订阅已暂停/完成/取消 + if (slot.subscription_id && !sub) { + // 订阅已暂停/完成/取消:不评估,也不留延后行 + await clearDeferral(db, slot.id, rel.id); + continue; + } if (slot.content_unit_id && !units.includes(slot.content_unit_id)) { // Slot 限定某季:季单元是否是本发布单元的父级 const seasonOk = await db @@ -645,6 +667,10 @@ async function evaluateForSlot(c: Ctx, p: SlotInput): Promise<{ planId: string | } const inflight = await slotInflightPlans(db, slot.id); const inflightKeys = inflight.flatMap((x) => x.unitKeys); + const deferWithRetry = async (slotId: string, releaseId: string, reason: string, unitKeys: string[]) => { + const attempts = await recordDeferral(db, { slotId, sourceReleaseId: releaseId, reason, unitKeys }); + await scheduleSlotRetry(db, slotId, new Date(now.getTime() + gateRetryDelayMs(attempts))); + }; const defer = async (reason: string, unitKeys: string[], mode: AcquisitionMode | null = null) => { await recordDeferral(db, { slotId: slot.id, sourceReleaseId: rel.id, reason, unitKeys }); await note(c, { @@ -823,9 +849,21 @@ async function evaluateForSlot(c: Ctx, p: SlotInput): Promise<{ planId: string | planUnits = fill.planUnits; const gate = await gates(c, rel, mode, sub); if (!gate.ok) { - await note(c, { ...base, decision: "skipped", mode, rejected: gate.rejected, reasons: gate.reasons, ...evalExtra }); + // 被挡住不是"不要":写延后并按次数退避排重评(电影 / 已满足的剧集收录版本没有周期重评,全靠这里) + const reason = `gate:${gate.kind}`; + if (!c.draft) await deferWithRetry(slot.id, rel.id, reason, planUnits); + await note(c, { + ...base, + decision: "deferred", + mode, + rejected: gate.rejected, + reasons: [`deferred:${reason}`, ...gate.reasons], + ...evalExtra, + unitKeys: planUnits, + }); c.out.skipped.push({ sourceReleaseId: rel.id, reason: gate.reasons[0] ?? "gated" }); - return done; + if (!c.draft) c.out.deferred.push({ sourceReleaseId: rel.id, slotId: slot.id, reason }); + return { planId: null, deferred: !c.draft }; } const r = await reserve(c, { slot, @@ -943,8 +981,11 @@ interface GateOk { volumeKey: string; savePath: string; } +/** 闸门类型:写进延后原因(gate:automation / gate:downloader / gate:budget / gate:cache) */ +type GateKind = "automation" | "downloader" | "budget" | "cache"; interface GateNo { ok: false; + kind: GateKind; reasons: string[]; rejected: { code: string; message: string }[]; } @@ -958,13 +999,13 @@ async function gates( ): Promise { const { d } = c; const blocked = await automationBlock(d); - if (blocked) return { ok: false, reasons: [blocked], rejected: [] }; + if (blocked) return { ok: false, kind: "automation", reasons: [blocked], rejected: [] }; const sp = (sub?.params ?? {}) as { downloaderInstanceId?: string; downloadSavePath?: string }; const inst = await pickInstance(d.ctx.db, await loadInstances(d.ctx.db), { explicitId: sp.downloaderInstanceId, siteId: rel.site_id, }); - if (!inst?.savePath) return { ok: false, reasons: ["no_downloader_instance"], rejected: [] }; + if (!inst?.savePath) return { ok: false, kind: "downloader", reasons: ["no_downloader_instance"], rejected: [] }; // 订阅指定的保存路径:本机路径经下载器路径映射换成下载器侧路径;本来就是下载器侧路径的原样用 let savePath = inst.savePath; if (sp.downloadSavePath) { @@ -972,6 +1013,7 @@ async function gates( if (!mapped) return { ok: false, + kind: "downloader", reasons: ["download_save_path_unmapped"], rejected: [ { @@ -996,6 +1038,7 @@ async function gates( if (b.decision !== "allow") { return { ok: false, + kind: "budget", reasons: [b.decision === "unknown" ? "budget_unknown_size" : "budget_exceeded"], rejected: b.violations.map((v) => ({ code: `budget.${v.kind}`, @@ -1007,11 +1050,12 @@ async function gates( const vol = (await loadVolumes(d, d.ctx.db, await loadInstances(d.ctx.db))).find( (v) => v.volumeKey === inst.volumeKey, ); - if (!vol) return { ok: false, reasons: ["volume_state_unknown"], rejected: [] }; + if (!vol) return { ok: false, kind: "cache", reasons: ["volume_state_unknown"], rejected: [] }; const dv = decideVolume(vol, cache, rel.size_bytes !== null ? { bytes: rel.size_bytes } : {}); if (dv.decision !== "allow") { return { ok: false, + kind: "cache", reasons: dv.reasons.length ? dv.reasons.map((r) => `cache:${r}`) : ["cache:blocked"], rejected: [ { diff --git a/apps/server/src/modules/acquisition-organize/services/subscriptions.ts b/apps/server/src/modules/acquisition-organize/services/subscriptions.ts index 4e795fac..16fda6df 100644 --- a/apps/server/src/modules/acquisition-organize/services/subscriptions.ts +++ b/apps/server/src/modules/acquisition-organize/services/subscriptions.ts @@ -577,6 +577,7 @@ export async function completeSubscription( .returningAll() .executeTakeFirst(); if (!done) return false; + await clearSubscriptionDeferrals(trx, done.id); await trx .insertInto("subscription_history") .values({ @@ -597,16 +598,56 @@ export async function completeSubscription( // ── 延后与重评调度 ── +/** 写 / 更新延后行,返回累计次数(首次 1) */ export async function recordDeferral( exec: Exec, input: { slotId: string; sourceReleaseId: string; reason: string; unitKeys: readonly string[] }, -): Promise { - await sql` +): Promise { + const r = await sql<{ attempts: number }>` insert into slot_deferral (slot_id, source_release_id, reason, unit_keys) values (${input.slotId}, ${input.sourceReleaseId}, ${input.reason}, ${[...input.unitKeys]}) on conflict (slot_id, source_release_id) do update set reason = excluded.reason, unit_keys = excluded.unit_keys, attempts = slot_deferral.attempts + 1, - updated_at = now()`.execute(exec); + updated_at = now() + returning attempts`.execute(exec); + return r.rows[0]?.attempts ?? 1; +} + +/** 某资源的全部延后行(资源过期 / 不可评估时不再留着空转) */ +export async function clearReleaseDeferrals(exec: Exec, sourceReleaseId: string): Promise { + await exec.deleteFrom("slot_deferral").where("source_release_id", "=", sourceReleaseId).execute(); +} + +/** 某订阅下收录版本的全部延后行(订阅完成 / 暂停 / 取消后不再重评) */ +export async function clearSubscriptionDeferrals(exec: Exec, subscriptionId: string): Promise { + await exec + .deleteFrom("slot_deferral") + .where("slot_id", "in", (eb) => + eb.selectFrom("acquisition_slot").select("id").where("subscription_id", "=", subscriptionId), + ) + .execute(); +} + +/** + * 被闸门挡住(自动化开关 / 下载器 / 预算 / 缓存卷)或失败计划冷却中的延后:按次数退避, + * 30 分钟起翻倍,封顶 6 小时。不等事件触发(预算释放、缓存卷恢复没有统一事件),到点由 sweepDueSlots 重评。 + */ +export const GATE_RETRY_BASE_MS = 30 * 60_000; +export const GATE_RETRY_MAX_MS = 6 * 3600_000; +export const gateRetryDelayMs = (attempts: number): number => + Math.min(GATE_RETRY_BASE_MS * 2 ** Math.max(0, Math.min(attempts, 16) - 1), GATE_RETRY_MAX_MS); + +/** + * 最晚在 at 重评该收录版本(已排得更早的不推迟);电影收录版本与已满足的剧集收录版本也适用 + * (scheduleSlotReevaluate 只给未满足的剧集排周期)。 + */ +export async function scheduleSlotRetry(exec: Exec, slotId: string, at: Date): Promise { + await exec + .updateTable("acquisition_slot") + .set({ next_evaluate_at: at }) + .where("id", "=", slotId) + .where((eb) => eb.or([eb("next_evaluate_at", "is", null), eb("next_evaluate_at", ">", at)])) + .execute(); } export async function clearDeferral(exec: Exec, slotId: string, sourceReleaseId: string): Promise { diff --git a/apps/server/test/acquisition-organize/blocked-reevaluate.test.ts b/apps/server/test/acquisition-organize/blocked-reevaluate.test.ts new file mode 100644 index 00000000..414d5798 --- /dev/null +++ b/apps/server/test/acquisition-organize/blocked-reevaluate.test.ts @@ -0,0 +1,249 @@ +import { createMockQb, type MockQb } from "@umt/testkit"; +import { sql } from "kysely"; +import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest"; +import { evaluateAcquisition } from "../../src/modules/acquisition-organize/services/acquisition.ts"; +import { + completeSubscription, + GATE_RETRY_BASE_MS, + sweepDueSlots, +} from "../../src/modules/acquisition-organize/services/subscriptions.ts"; +import { createHarness, fakeDisk, type Harness, pgAvailable, seedBase, seedEntity, seedRelease } from "./harness.ts"; + +// 获取环节被挡住 / 失败后的重评(T1-3):闸门不过写延后并排重评(电影、已满足剧集收录版本); +// 并发预算不计 stalled;订阅完成 / 资源过期清延后行。 + +const hasPg = await pgAvailable(); +const SIZE = 300_000_000; + +describe.skipIf(!hasPg)("被挡住与失败后的重评", () => { + let h: Harness; + let qb: MockQb; + let base: Awaited>; + const disk = fakeDisk(10_000_000_000); + + beforeAll(async () => { + qb = await createMockQb({ addVisibleAfterPolls: 0 }); + disk.setPct(10); + h = await createHarness("blocked_reevaluate", { diskUsage: disk.fn }); + base = await seedBase(h.db, { libraryPath: "/tmp/umt-none-lib", cachePath: "/tmp/umt-none-cache", qbUrl: qb.url }); + }, 60_000); + + afterAll(async () => { + await h?.close(); + await qb?.close(); + }); + + const db = () => h.db; + // 用例之间不共享在途:上一个用例留下的预留计划会占并发预算 + beforeEach(async () => { + await db().updateTable("acquisition_plan").set({ status: "cancelled" }).where("status", "=", "reserved").execute(); + }); + const hash = () => Math.random().toString(16).slice(2).padEnd(40, "0").slice(0, 40); + + async function movieSlot(title: string) { + const entityId = await seedEntity(db(), { kind: "movie", title, year: 2024 }); + const sub = await db() + .insertInto("subscription") + .values({ media_entity_id: entityId, media_kind: "movie", mode: "once", status: "active" }) + .returning("id") + .executeTakeFirstOrThrow(); + const slot = await db() + .insertInto("acquisition_slot") + .values({ + media_entity_id: entityId, + subscription_id: sub.id, + name: "1080p", + hard_conditions: JSON.stringify({ resolutions: ["1080p"] }), + upgrade_threshold: JSON.stringify({}), + }) + .returning("id") + .executeTakeFirstOrThrow(); + return { entityId, subId: sub.id, slotId: slot.id }; + } + + async function seriesSlot(title: string, status: "empty" | "satisfied" = "empty") { + const entityId = await seedEntity(db(), { kind: "series", title, year: 2024 }); + const sub = await db() + .insertInto("subscription") + .values({ media_entity_id: entityId, media_kind: "series", mode: "continuous", status: "active" }) + .returning("id") + .executeTakeFirstOrThrow(); + const slot = await db() + .insertInto("acquisition_slot") + .values({ + media_entity_id: entityId, + subscription_id: sub.id, + name: "1080p", + status, + hard_conditions: JSON.stringify({ resolutions: ["1080p"] }), + upgrade_threshold: JSON.stringify({}), + }) + .returning("id") + .executeTakeFirstOrThrow(); + return { entityId, subId: sub.id, slotId: slot.id }; + } + + const release = (entityId: string, title: string) => + seedRelease(db(), { siteId: base.siteId, entityId, title, size: SIZE, hash: hash() }); + const deferrals = (slotId: string) => + db().selectFrom("slot_deferral").selectAll().where("slot_id", "=", slotId).execute(); + const slotRow = (slotId: string) => + db().selectFrom("acquisition_slot").selectAll().where("id", "=", slotId).executeTakeFirstOrThrow(); + const candidate = async (slotId: string, releaseId: string) => + ( + await sql<{ content: { decision: string; reasons: string[] } }>` + select content from artifact where title = 'candidate_evaluation' and subject_id = ${slotId} + and content->>'sourceReleaseId' = ${releaseId}`.execute(db()) + ).rows[0]?.content; + /** 缓存卷超阈值挡住 → 延后 + 排重评;恢复后到期重评能建计划 */ + async function gateThenRecover(slotId: string, releaseId: string) { + disk.setPct(95); + const t0 = Date.now(); + const ev = await evaluateAcquisition(h.d, { sourceReleaseId: releaseId }); + expect(ev.plans).toHaveLength(0); + expect(ev.skipped[0]?.reason).toMatch(/^cache:/); + expect(ev.deferred).toContainEqual({ sourceReleaseId: releaseId, slotId, reason: "gate:cache" }); + const rows = await deferrals(slotId); + expect(rows.map((r) => [r.source_release_id, r.reason])).toEqual([[releaseId, "gate:cache"]]); + expect((await candidate(slotId, releaseId))?.decision).toBe("deferred"); + const due = (await slotRow(slotId)).next_evaluate_at; + expect(due).not.toBeNull(); + const delay = due!.getTime() - t0; + expect(delay).toBeGreaterThan(GATE_RETRY_BASE_MS - 60_000); + expect(delay).toBeLessThanOrEqual(GATE_RETRY_BASE_MS + 60_000); + + // 恢复后到期:巡检入队按收录版本重评,先评延后的资源 + disk.setPct(10); + await db() + .updateTable("acquisition_slot") + .set({ next_evaluate_at: new Date(Date.now() - 1000) }) + .where("id", "=", slotId) + .execute(); + h.engine.enqueued.length = 0; + expect(await sweepDueSlots(h.d)).toBeGreaterThanOrEqual(1); + const job = h.engine.enqueued.find( + (e) => + e.task === "evaluateAcquisition" && (e.input as { acquisitionSlotId?: string }).acquisitionSlotId === slotId, + ); + expect(job).toBeDefined(); + const ev2 = await evaluateAcquisition(h.d, job!.input as { acquisitionSlotId: string }); + expect(ev2.plans, JSON.stringify(ev2)).toHaveLength(1); + expect(await deferrals(slotId)).toHaveLength(0); + expect((await candidate(slotId, releaseId))?.decision).toBe("accept"); + } + + it("电影收录版本:缓存卷超阈值写延后 gate:cache 并排重评,恢复后重评建计划", async () => { + const { entityId, slotId } = await movieSlot("Gate Movie"); + const rel = await release(entityId, "Gate.Movie.2024.1080p.WEB-DL.x264-GRP"); + await gateThenRecover(slotId, rel); + }); + + it("已满足的剧集收录版本:新集被挡住同样写延后并重评", async () => { + const { entityId, slotId } = await seriesSlot("Gate Series", "satisfied"); + const rel = await release(entityId, "Gate.Series.S01E05.1080p.WEB-DL.x264-GRP"); + await gateThenRecover(slotId, rel); + }); + + it("再次被挡住按次数退避(第 2 次 60 分钟)", async () => { + const { entityId, slotId } = await movieSlot("Backoff Movie"); + const rel = await release(entityId, "Backoff.Movie.2024.1080p.WEB-DL.x264-GRP"); + await h.setSettings("automation", { enabled: false }); + try { + await evaluateAcquisition(h.d, { sourceReleaseId: rel }); + await db().updateTable("acquisition_slot").set({ next_evaluate_at: null }).where("id", "=", slotId).execute(); + const t0 = Date.now(); + await evaluateAcquisition(h.d, { sourceReleaseId: rel }); + const [row] = await deferrals(slotId); + expect(row).toMatchObject({ reason: "gate:automation", attempts: 2 }); + const delay = (await slotRow(slotId)).next_evaluate_at!.getTime() - t0; + expect(delay).toBeGreaterThan(2 * GATE_RETRY_BASE_MS - 60_000); + expect(delay).toBeLessThanOrEqual(2 * GATE_RETRY_BASE_MS + 60_000); + } finally { + await h.setSettings("automation", { enabled: true }); + } + }); + + it("预算并发:stalled 任务不占并发名额,在跑的任务占", async () => { + await h.setSettings("budget", { maxConcurrentDownloads: 1 }); + try { + const busyTask = await db() + .insertInto("download_task") + .values({ + instance_id: base.instanceId, + torrent_hash: hash(), + name: "stalled-one", + size_bytes: 1_000_000, + status: "stalled", + managed: true, + }) + .returning("id") + .executeTakeFirstOrThrow(); + const a = await movieSlot("Budget Movie A"); + const relA = await release(a.entityId, "Budget.Movie.A.2024.1080p.WEB-DL.x264-GRP"); + const evA = await evaluateAcquisition(h.d, { sourceReleaseId: relA }); + expect(evA.plans, JSON.stringify(evA)).toHaveLength(1); + // 刚预留、还没建下载任务的计划算一个名额 + const b = await movieSlot("Budget Movie B"); + const relB = await release(b.entityId, "Budget.Movie.B.2024.1080p.WEB-DL.x264-GRP"); + const evB = await evaluateAcquisition(h.d, { sourceReleaseId: relB }); + expect(evB.plans).toHaveLength(0); + expect(evB.skipped[0]?.reason).toBe("budget_exceeded"); + expect((await deferrals(b.slotId))[0]?.reason).toBe("gate:budget"); + // A 结束、stalled 任务变成下载中:仍占满 + await db().updateTable("acquisition_plan").set({ status: "cancelled" }).where("id", "=", evA.plans[0]!).execute(); + await db().updateTable("download_task").set({ status: "downloading" }).where("id", "=", busyTask.id).execute(); + expect((await evaluateAcquisition(h.d, { sourceReleaseId: relB })).plans).toHaveLength(0); + await db().updateTable("download_task").set({ status: "stalled" }).where("id", "=", busyTask.id).execute(); + const evB2 = await evaluateAcquisition(h.d, { sourceReleaseId: relB }); + expect(evB2.plans).toHaveLength(1); + expect(await deferrals(b.slotId)).toHaveLength(0); + await db().deleteFrom("download_task").where("id", "=", busyTask.id).execute(); + await db() + .updateTable("acquisition_plan") + .set({ status: "cancelled" }) + .where("id", "=", evB2.plans[0]!) + .execute(); + } finally { + await h.setSettings("budget", {}); + } + }); + + it("订阅完成、订阅暂停、资源过期时清掉延后行", async () => { + disk.setPct(95); + try { + // 完成 + const a = await movieSlot("Cleanup Movie A"); + const relA = await release(a.entityId, "Cleanup.Movie.A.2024.1080p.WEB-DL.x264-GRP"); + await evaluateAcquisition(h.d, { sourceReleaseId: relA }); + expect(await deferrals(a.slotId)).toHaveLength(1); + const subA = await db() + .selectFrom("subscription") + .selectAll() + .where("id", "=", a.subId) + .executeTakeFirstOrThrow(); + expect(await completeSubscription(h.d, db(), subA, "test")).toBe(true); + expect(await deferrals(a.slotId)).toHaveLength(0); + + // 暂停:到期重评时跳过并清掉 + const b = await movieSlot("Cleanup Movie B"); + const relB = await release(b.entityId, "Cleanup.Movie.B.2024.1080p.WEB-DL.x264-GRP"); + await evaluateAcquisition(h.d, { sourceReleaseId: relB }); + expect(await deferrals(b.slotId)).toHaveLength(1); + await db().updateTable("subscription").set({ status: "paused" }).where("id", "=", b.subId).execute(); + await evaluateAcquisition(h.d, { acquisitionSlotId: b.slotId }); + expect(await deferrals(b.slotId)).toHaveLength(0); + + // 资源过期 + const c = await movieSlot("Cleanup Movie C"); + const relC = await release(c.entityId, "Cleanup.Movie.C.2024.1080p.WEB-DL.x264-GRP"); + await evaluateAcquisition(h.d, { sourceReleaseId: relC }); + expect(await deferrals(c.slotId)).toHaveLength(1); + await db().updateTable("source_release").set({ availability: "expired" }).where("id", "=", relC).execute(); + const ev = await evaluateAcquisition(h.d, { sourceReleaseId: relC }); + expect(ev.skipped[0]?.reason).toBe("release_expired"); + expect(await deferrals(c.slotId)).toHaveLength(0); + } finally { + disk.setPct(10); + } + }); +}); diff --git a/apps/server/test/e2e/continuous-series.test.ts b/apps/server/test/e2e/continuous-series.test.ts index 7a38ac7b..54259e96 100644 --- a/apps/server/test/e2e/continuous-series.test.ts +++ b/apps/server/test/e2e/continuous-series.test.ts @@ -12,7 +12,7 @@ import { completeDownload, createE2E, type E2E, pgAvailable, seedBase, waitFor } // E9 连续更新剧集(W5-C,对照 MP 缺集 − note 与批量下载):全部经真实任务引擎 + outbox 中继。 // 季共 3 集(mock TMDB),新作品策略按 once 订阅。E01 还在下载(Slot 忙)时 E02、E03 与 E02 的另一个版本先后到达: // · E02/E03 与在途的 E01 单元不相交 → 各自得到计划并同时下载(Slot×单元预留),不再因 slot_busy 丢集; -// · E02 另一版本与在途 E02 单元冲突 → 延后(slot_deferral),计划结束后重评时已覆盖,不重复下载; +// · E02 另一版本与在途 E02 单元冲突 → 延后(候选评估 deferred;订阅接收即完成后延后行清掉),不重复下载; // · refreshMetadata(编排循环按周期入队)把 TMDB 季集数写进季单元 → 三集全部入库后 once 订阅完成并写历史。 const hasPg = await pgAvailable(); @@ -216,14 +216,24 @@ describe.skipIf(!hasPg)("E9 连续更新剧集:Slot 忙时到达的集全部入 60_000, "另一版本识别", ); + // 三集都被下载器接收后 once 订阅即完成,完成时延后行随之清掉;以候选评估记录为准 await waitFor( async () => - e.db - .selectFrom("slot_deferral") - .selectAll() - .where("slot_id", "=", slotId) - .where("source_release_id", "=", loserRel.id) - .executeTakeFirst(), + ( + await e.db + .selectFrom("artifact") + .select("content") + .where("title", "=", "candidate_evaluation") + .where("subject_id", "=", slotId) + .execute() + ).find((r) => { + const c = r.content as { sourceReleaseId: string; decision: string; reasons: string[] }; + return ( + c.sourceReleaseId === loserRel.id && + c.decision === "deferred" && + c.reasons.includes("deferred:units_inflight") + ); + }), 60_000, "另一版本延后", ); From 4532587ad5ed8bcc42a8be61ebfc43e794e2f581 Mon Sep 17 00:00:00 2001 From: root Date: Sat, 3 Oct 2026 20:47:48 +0800 Subject: [PATCH 3/3] =?UTF-8?q?fix(acquisition):=20=E8=AE=A1=E5=88=92?= =?UTF-8?q?=E5=B9=82=E7=AD=89=E9=94=AE=E5=8F=AA=E6=8C=A1=E5=9C=A8=E9=80=94?= =?UTF-8?q?=20/=20=E5=B7=B2=E5=AE=8C=E6=88=90=EF=BC=8C=E5=A4=B1=E8=B4=A5?= =?UTF-8?q?=20/=20=E5=8F=96=E6=B6=88=E7=9A=84=E6=97=A7=E8=AE=A1=E5=88=92?= =?UTF-8?q?=E8=AE=A9=E4=BD=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit acq:{slot}:{release} 原先对 failed / cancelled 也冲突，站点 Cookie 修好、下载器恢复、 重置订阅后同一资源对该收录版本永不再建计划。现在在收录版本行锁内查旧计划： - reserved / submitted / completed：照旧挡住； - failed / cancelled / expired：旧计划幂等键改名为 …:retired:{id} 让位，再插新计划（同一事务，唯一约束兜底）； - 用户明确不要的（download_removed、cancelled_by_user）和整理阶段失败（organize:*）仍挡住； - failed 的按该资源已失败次数冷却（30 分钟起翻倍，封顶 6 小时），冷却中写延后 retry_cooldown 并排到冷却结束重评， 避免持续失败时每轮都建计划、开待处理。 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei --- .../services/acquisition.ts | 64 ++++++++++- .../blocked-reevaluate.test.ts | 107 +++++++++++++++++- 2 files changed, 168 insertions(+), 3 deletions(-) diff --git a/apps/server/src/modules/acquisition-organize/services/acquisition.ts b/apps/server/src/modules/acquisition-organize/services/acquisition.ts index 6ff90aff..f6cd5bf5 100644 --- a/apps/server/src/modules/acquisition-organize/services/acquisition.ts +++ b/apps/server/src/modules/acquisition-organize/services/acquisition.ts @@ -879,6 +879,27 @@ async function evaluateForSlot(c: Ctx, p: SlotInput): Promise<{ planId: string | ...(fill.action === "fill" ? { priority: fill.priority, evaluationExtra: fill.evaluation } : {}), }); if (r.conflict) return defer("units_inflight", r.conflict, mode); + if (r.retryAt) { + // 同一资源上次计划失败,冷却中:延后到冷却结束再试 + await recordDeferral(db, { + slotId: slot.id, + sourceReleaseId: rel.id, + reason: "retry_cooldown", + unitKeys: planUnits, + }); + await scheduleSlotRetry(db, slot.id, r.retryAt); + await note(c, { + ...base, + decision: "deferred", + mode, + rejected: [], + reasons: ["deferred:retry_cooldown", `retry_at:${r.retryAt.toISOString()}`], + ...evalExtra, + unitKeys: planUnits, + }); + c.out.deferred.push({ sourceReleaseId: rel.id, slotId: slot.id, reason: "retry_cooldown" }); + return { planId: null, deferred: true }; + } const planId = r.planId; if (!planId) { await note(c, { ...base, decision: "skipped", mode, rejected: [], reasons: ["slot_changed_or_reserved"] }); @@ -1080,6 +1101,16 @@ export function subscriptionSavePath(t: Pick; }, -): Promise<{ planId: string | null; conflict?: string[] }> { +): Promise<{ planId: string | null; conflict?: string[]; retryAt?: Date }> { const { d, now } = c; const immediate = p.sub?.priority === "immediate"; const priority = immediate ? PRIORITY.immediate : (p.priority ?? PRIORITY[priorityOf(p.mode, immediate)]); @@ -1191,6 +1222,35 @@ async function reserve( } else if (slot.revision !== p.slot.revision || busy) { return { planId: null }; } + // 幂等键只挡在途 / 已完成的计划;失败 / 取消 / 过期的旧计划改名让位(Slot 行锁内,同 Slot 的预留串行; + // 改名与插入在同一事务,唯一约束仍兜底) + const key = `acq:${slot.id}:${p.rel.id}`; + const prev = await trx + .selectFrom("acquisition_plan") + .select(["id", "status", "status_reason", "updated_at"]) + .where("idempotency_key", "=", key) + .forUpdate() + .executeTakeFirst(); + if (prev) { + if (!planRetryable(prev)) return { planId: null }; + if (prev.status === "failed") { + // 失败的冷却一段再重试(按该资源已失败次数翻倍),免得持续失败时每轮都建计划、开待处理 + const retired = await trx + .selectFrom("acquisition_plan") + .select((eb) => eb.fn.countAll().as("n")) + .where("slot_id", "=", slot.id) + .where("source_release_id", "=", p.rel.id) + .where("idempotency_key", "like", `${key}:retired:%`) + .executeTakeFirstOrThrow(); + const retryAt = new Date(prev.updated_at.getTime() + gateRetryDelayMs(Number(retired.n) + 1)); + if (retryAt > now) return { planId: null, retryAt }; + } + await trx + .updateTable("acquisition_plan") + .set({ idempotency_key: `${key}:retired:${prev.id}` }) + .where("id", "=", prev.id) + .execute(); + } let st: SlotState = { status: slot.status, hasAsset: slot.current_asset_id !== null }; if (!busy) { if (st.status === "satisfied") st = transition(st, { type: "upgrade_found" }); @@ -1213,7 +1273,7 @@ async function reserve( }), payload_hash: hashJson(payload), expected_revision: slot.revision + 1, - idempotency_key: `acq:${slot.id}:${p.rel.id}`, + idempotency_key: key, unit_keys: p.unitKeys, }) .onConflict((oc) => oc.column("idempotency_key").doNothing()) diff --git a/apps/server/test/acquisition-organize/blocked-reevaluate.test.ts b/apps/server/test/acquisition-organize/blocked-reevaluate.test.ts index 414d5798..ea956b38 100644 --- a/apps/server/test/acquisition-organize/blocked-reevaluate.test.ts +++ b/apps/server/test/acquisition-organize/blocked-reevaluate.test.ts @@ -10,7 +10,7 @@ import { import { createHarness, fakeDisk, type Harness, pgAvailable, seedBase, seedEntity, seedRelease } from "./harness.ts"; // 获取环节被挡住 / 失败后的重评(T1-3):闸门不过写延后并排重评(电影、已满足剧集收录版本); -// 并发预算不计 stalled;订阅完成 / 资源过期清延后行。 +// 并发预算不计 stalled;失败 / 取消的旧计划不挡同一资源重建计划;订阅完成 / 资源过期清延后行。 const hasPg = await pgAvailable(); const SIZE = 300_000_000; @@ -95,6 +95,14 @@ describe.skipIf(!hasPg)("被挡住与失败后的重评", () => { select content from artifact where title = 'candidate_evaluation' and subject_id = ${slotId} and content->>'sourceReleaseId' = ${releaseId}`.execute(db()) ).rows[0]?.content; + /** 模拟计划结束后收录版本释放(applySlotEvent release 的效果) */ + const releaseSlot = (slotId: string) => + db() + .updateTable("acquisition_slot") + .set({ status: "empty", reserved_plan_id: null }) + .where("id", "=", slotId) + .execute(); + /** 缓存卷超阈值挡住 → 延后 + 排重评;恢复后到期重评能建计划 */ async function gateThenRecover(slotId: string, releaseId: string) { disk.setPct(95); @@ -208,6 +216,103 @@ describe.skipIf(!hasPg)("被挡住与失败后的重评", () => { } }); + it("计划失败后同一资源能重建计划(冷却后);在途 / 已完成 / 用户移除的仍挡住", async () => { + const { entityId, slotId } = await movieSlot("Retry Movie"); + const rel = await release(entityId, "Retry.Movie.2024.1080p.WEB-DL.x264-GRP"); + const ev1 = await evaluateAcquisition(h.d, { sourceReleaseId: rel }); + expect(ev1.plans, JSON.stringify(ev1)).toHaveLength(1); + const p1 = ev1.plans[0]!; + // 失败时间 = updated_at;触发器会把它改成 now(),回拨时在事务里临时停用本 schema 的触发器 + const fail = (id: string, reason: string, agoMs: number) => + db() + .transaction() + .execute(async (trx) => { + await sql`alter table acquisition_plan disable trigger user`.execute(trx); + await trx + .updateTable("acquisition_plan") + .set({ status: "failed", status_reason: reason, updated_at: new Date(Date.now() - agoMs) }) + .where("id", "=", id) + .execute(); + await sql`alter table acquisition_plan enable trigger user`.execute(trx); + }); + + // 刚失败:冷却中 → 延后 retry_cooldown,排到冷却结束 + await fail(p1, "source_failed: cookie expired", 0); + await releaseSlot(slotId); + const ev2 = await evaluateAcquisition(h.d, { sourceReleaseId: rel }); + expect(ev2.plans).toHaveLength(0); + expect(ev2.deferred).toContainEqual({ sourceReleaseId: rel, slotId, reason: "retry_cooldown" }); + expect((await deferrals(slotId))[0]?.reason).toBe("retry_cooldown"); + expect((await slotRow(slotId)).next_evaluate_at).not.toBeNull(); + + // 冷却过后:建新计划,旧计划的幂等键让位 + await fail(p1, "source_failed: cookie expired", GATE_RETRY_BASE_MS + 60_000); + const ev3 = await evaluateAcquisition(h.d, { sourceReleaseId: rel }); + expect(ev3.plans, JSON.stringify(ev3)).toHaveLength(1); + const p2 = ev3.plans[0]!; + expect(p2).not.toBe(p1); + const keys = await db() + .selectFrom("acquisition_plan") + .select(["id", "idempotency_key"]) + .where("id", "in", [p1, p2]) + .execute(); + expect(keys.find((k) => k.id === p2)?.idempotency_key).toBe(`acq:${slotId}:${rel}`); + expect(keys.find((k) => k.id === p1)?.idempotency_key).toBe(`acq:${slotId}:${rel}:retired:${p1}`); + expect(await deferrals(slotId)).toHaveLength(0); + + // 第二次失败:冷却翻倍(30 分钟前失败仍在冷却) + await fail(p2, "rejected", GATE_RETRY_BASE_MS + 60_000); + await releaseSlot(slotId); + expect((await evaluateAcquisition(h.d, { sourceReleaseId: rel })).plans).toHaveLength(0); + await fail(p2, "rejected", 2 * GATE_RETRY_BASE_MS + 60_000); + const ev4 = await evaluateAcquisition(h.d, { sourceReleaseId: rel }); + expect(ev4.plans).toHaveLength(1); + const p3 = ev4.plans[0]!; + + // 已完成:挡住 + await db().updateTable("acquisition_plan").set({ status: "completed" }).where("id", "=", p3).execute(); + await releaseSlot(slotId); + const ev5 = await evaluateAcquisition(h.d, { sourceReleaseId: rel }); + expect(ev5.plans).toHaveLength(0); + expect((await candidate(slotId, rel))?.reasons).toContain("slot_changed_or_reserved"); + + // 用户在下载器里移除:不自动重下 + await fail(p3, "download_removed", 24 * 3600_000); + const ev6 = await evaluateAcquisition(h.d, { sourceReleaseId: rel }); + expect(ev6.plans).toHaveLength(0); + }); + + it("取消的旧计划:重置订阅取消的立即让位,用户手动取消的仍挡住", async () => { + const { entityId, slotId } = await movieSlot("Cancel Movie"); + const rel = await release(entityId, "Cancel.Movie.2024.1080p.WEB-DL.x264-GRP"); + const ev1 = await evaluateAcquisition(h.d, { sourceReleaseId: rel }); + expect(ev1.plans).toHaveLength(1); + const cancel = (id: string, reason: string) => + db() + .updateTable("acquisition_plan") + .set({ status: "cancelled", status_reason: reason }) + .where("id", "=", id) + .execute(); + await cancel(ev1.plans[0]!, "subscription_reset"); + await releaseSlot(slotId); + const ev2 = await evaluateAcquisition(h.d, { sourceReleaseId: rel }); + expect(ev2.plans, JSON.stringify(ev2)).toHaveLength(1); + await cancel(ev2.plans[0]!, "cancelled_by_user"); + await releaseSlot(slotId); + expect((await evaluateAcquisition(h.d, { sourceReleaseId: rel })).plans).toHaveLength(0); + }); + + it("在途计划(已提交)仍挡住同一资源", async () => { + const { entityId } = await seriesSlot("Inflight Series"); + const rel = await release(entityId, "Inflight.Series.S01E01.1080p.WEB-DL.x264-GRP"); + const ev1 = await evaluateAcquisition(h.d, { sourceReleaseId: rel }); + expect(ev1.plans, JSON.stringify(ev1)).toHaveLength(1); + await db().updateTable("acquisition_plan").set({ status: "submitted" }).where("id", "=", ev1.plans[0]!).execute(); + const ev2 = await evaluateAcquisition(h.d, { sourceReleaseId: rel }); + expect(ev2.plans).toHaveLength(0); + expect(ev2.deferred[0]?.reason).toBe("units_inflight"); + }); + it("订阅完成、订阅暂停、资源过期时清掉延后行", async () => { disk.setPct(95); try {