fix(acquisition): 计划幂等键只挡在途 / 已完成，失败 / 取消的旧计划让位

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 <[email protected]>
Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei

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<InstanceTopology, "inst" | "mapping
   return hostPathOf(t, p) ? p : null;
 }
 
+/**
+ * 旧计划是否让位给新计划(T1-3c):失败 / 取消 / 过期可以重建(站点 Cookie 修好、下载器恢复、集数映射改号后重评),
+ * 但用户明确不要的(在下载器里移除、手动取消)和整理阶段失败的(文件已下好,重下同一种子无意义,交给整理重试)仍挡住。
+ */
+function planRetryable(p: { status: string; status_reason: string | null }): boolean {
+  if (!["failed", "cancelled", "expired"].includes(p.status)) return false;
+  const r = p.status_reason ?? "";
+  return r !== "download_removed" && r !== "cancelled_by_user" && !r.startsWith("organize:");
+}
+
 async function reserve(
   c: Ctx,
   p: {
@@ -1099,7 +1130,7 @@ async function reserve(
     priority?: AcquisitionPriority;
     evaluationExtra?: Record<string, unknown>;
   },
-): 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<number>().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())

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 {