fix(acquisition): 闸门挡住的资源写延后并排重评；并发预算不计 stalled；清理空转延后行

- 自动化开关 / 下载器 / 预算 / 缓存卷挡住时，除候选记录（改为 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 <[email protected]> Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei
2026-10-03 20:47:23 +08:00
1 parent 4d24edd94c
commit 8c10c12bd7
@@ -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<GateOk | GateNo> {
  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: [
        {
@@ -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<void> {
-  await sql`
+): Promise<number> {
+  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<void> {
+  await exec.deleteFrom("slot_deferral").where("source_release_id", "=", sourceReleaseId).execute();
+}
+
+/** 某订阅下收录版本的全部延后行(订阅完成 / 暂停 / 取消后不再重评) */
+export async function clearSubscriptionDeferrals(exec: Exec, subscriptionId: string): Promise<void> {
+  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<void> {
+  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<void> {