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 {