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, "另一版本延后", );