From fe9693d4caa4a18da101f34b9dc5d83e3ba67414 Mon Sep 17 00:00:00 2001 From: root Date: Sun, 27 Sep 2026 22:12:16 +0800 Subject: [PATCH] refactor(ao): background loops only orchestrate; cleanup/reconcile endpoints enqueue - separate qB sync timer from the planning timer - threshold cleanup, reserved-plan and pending-import sweeps, 6h reconcile enqueue engine tasks with time-slot idempotency keys instead of running inline - pending-import sweep selects by completed_at (sync refreshes updated_at every round) - POST /download-cache/cleanup and /library/reconcile return {taskId, runId} (202); web jumps to the engine run page - test: a 50MB verifyAndImport in progress does not delay qB sync Co-Authored-By: Claude Opus 5.5 --- .../__tests__/background-loop.test.ts | 122 +++++++++++++++++ .../acquisition-organize/__tests__/harness.ts | 2 + .../acquisition-organize/capabilities.ts | 2 +- .../src/modules/acquisition-organize/index.ts | 126 +++++++++++++----- .../src/modules/acquisition-organize/ports.ts | 7 +- .../acquisition-organize/routes/downloads.ts | 42 +++--- .../acquisition-organize/routes/helpers.ts | 2 +- .../services/acquisition.ts | 2 +- .../acquisition-organize/services/common.ts | 43 +++++- .../services/downloads.ts | 38 +++++- apps/web/app/api/downloads.ts | 4 +- .../components/downloads/cache-volumes.tsx | 8 +- packages/testkit/src/qbittorrent/mock-qb.ts | 4 +- 13 files changed, 334 insertions(+), 68 deletions(-) create mode 100644 apps/server/src/modules/acquisition-organize/__tests__/background-loop.test.ts diff --git a/apps/server/src/modules/acquisition-organize/__tests__/background-loop.test.ts b/apps/server/src/modules/acquisition-organize/__tests__/background-loop.test.ts new file mode 100644 index 00000000..743cb7cd --- /dev/null +++ b/apps/server/src/modules/acquisition-organize/__tests__/background-loop.test.ts @@ -0,0 +1,122 @@ +import { sql } from "@umt/db"; +import { createMockQb, createTempTree, type MockQb, makeTorrent, type TempTree } from "@umt/testkit"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { evaluateAcquisition } from "../services/acquisition.ts"; +import { submitDownload, syncDownloads } from "../services/downloads.ts"; +import { executeRecord, planImport } from "../services/organize.ts"; +import { + createHarness, + fakeDisk, + type Harness, + pgAvailable, + seedBase, + seedEntity, + seedRelease, + seedSlot, +} from "./harness.ts"; + +// W4-D P0:后台循环只编排。补入库在巡检里只入队 verifyAndImport;一个 50MB 的入库(复制 + sha256 + 慢 ffprobe) +// 进行中时,qB 同步定时器仍按周期打到 mock qB(用 mock qB 的请求时间戳断言)。 + +const hasPg = await pgAvailable(); +const SIZE = 50 * 1024 * 1024; +const SYNC_MS = 200; +const PROBE_DELAY_MS = 2500; + +describe.skipIf(!hasPg)("acquisition-organize 后台循环不被入库阻塞", () => { + let h: Harness; + let qb: MockQb; + let lib: TempTree; + let cache: TempTree; + const disk = fakeDisk(10 * 1024 ** 3); + const torrents = new Map(); + let base: Awaited>; + + beforeAll(async () => { + qb = await createMockQb({ addVisibleAfterPolls: 0 }); + lib = await createTempTree({}, { sentinel: true, prefix: "umt-lib-" }); + cache = await createTempTree({}, { prefix: "umt-cache-" }); + disk.state.watch = [cache.root]; + disk.setPct(40); + h = await createHarness("bgloop", { + diskUsage: disk.fn, + torrentSource: async ({ release }) => ({ kind: "file", bytes: torrents.get(release.id)! }), + // 慢 ffprobe:让一次入库稳定地持续数秒 + probe: async () => { + await new Promise((r) => setTimeout(r, PROBE_DELAY_MS)); + return { status: "unknown", reason: "ffprobe_unavailable" }; + }, + options: { backgroundLoops: true, syncIntervalMs: SYNC_MS, planIntervalMs: SYNC_MS, reconcileIntervalMs: 0 }, + }); + base = await seedBase(h.db, { libraryPath: lib.root, cachePath: cache.root, qbUrl: qb.url }); + await h.setSettings("retention", { seedRetentionDays: 0 }); + }, 60_000); + + afterAll(async () => { + await h?.module.stop?.(); + await h?.close(); + await qb?.close(); + await lib?.cleanup(); + await cache?.cleanup(); + }); + + it("巡检只入队 verifyAndImport;50MB 入库进行中 qB 同步仍按时执行", async () => { + const entityId = await seedEntity(h.db, { kind: "movie", title: "长片", year: 2023 }); + await seedSlot(h.db, { entityId }); + const t = makeTorrent({ name: "Long.Movie.2023.1080p.BluRay.x264-GRP.mkv", length: SIZE, pieceSeed: 7 }); + const relId = await seedRelease(h.db, { + siteId: base.siteId, + entityId, + title: "Long.Movie.2023.1080p.BluRay.x264-GRP", + size: SIZE, + hash: t.hash, + }); + torrents.set(relId, t.bytes); + const ev = await evaluateAcquisition(h.d, { sourceReleaseId: relId }); + const sub = await submitDownload(h.d, ev.plans[0]!); + expect(sub.status).toBe("submitted"); + await qb.complete(t.hash, { writeFiles: true }); + expect((await syncDownloads(h.d)).completed).toContain(sub.downloadTaskId); + const dtId = sub.downloadTaskId!; + // 完成超过 5 分钟且没有整理记录 → 巡检对象 + await sql`update download_task set completed_at = now() - interval '10 minutes' where id = ${dtId}`.execute(h.db); + + await h.module.start?.(h.ctx); + const deadline = Date.now() + 10_000; + while (!h.engine.enqueued.some((e) => e.task === "verifyAndImport") && Date.now() < deadline) { + await new Promise((r) => setTimeout(r, 50)); + } + const enq = h.engine.enqueued.find((e) => e.task === "verifyAndImport"); + expect(enq?.input).toMatchObject({ downloadTaskId: dtId }); + expect(enq?.key).toMatch(new RegExp(`^verifyAndImport:import:${dtId}:\\d+$`)); + // 循环没有自己执行入库 + const recs = await h.db + .selectFrom("organization_record") + .select("id") + .where("download_task_id", "=", dtId) + .execute(); + expect(recs).toHaveLength(0); + // 同一时间槽重复巡检用同一个引擎幂等键(真实引擎去重;假引擎只记录) + await new Promise((r) => setTimeout(r, SYNC_MS * 3)); + const keys = new Set(h.engine.enqueued.filter((e) => e.task === "verifyAndImport").map((e) => e.key)); + expect(keys.size).toBe(1); + + // 模拟 worker 执行这次 verifyAndImport(同进程,与同步循环并发) + const t0 = Date.now(); + const plan = await planImport(h.d, { downloadTaskId: dtId, trigger: "auto", actorType: "system" }); + expect(plan.status).toBe("running"); + const ex = await executeRecord(h.d, plan.recordId!); + const t1 = Date.now(); + expect(ex).toMatchObject({ status: "succeeded", succeeded: 1 }); + expect(t1 - t0).toBeGreaterThanOrEqual(PROBE_DELAY_MS); + + const stamps = qb + .callsTo("torrents/info") + .map((c) => c.at) + .filter((at) => at >= t0 && at <= t1); + // 每 200ms 一轮(同步本身几十 ms);入库期间至少有 1/3 的轮次、最大间隔不超过 1s + expect(stamps.length).toBeGreaterThanOrEqual(Math.floor((t1 - t0) / SYNC_MS / 3)); + const gaps = [t0, ...stamps, t1].slice(1).map((v, i) => v - [t0, ...stamps][i]!); + expect(Math.max(...gaps)).toBeLessThan(1000); + }, 90_000); +}); diff --git a/apps/server/src/modules/acquisition-organize/__tests__/harness.ts b/apps/server/src/modules/acquisition-organize/__tests__/harness.ts index 376a57b3..6ef2e1ad 100644 --- a/apps/server/src/modules/acquisition-organize/__tests__/harness.ts +++ b/apps/server/src/modules/acquisition-organize/__tests__/harness.ts @@ -117,6 +117,7 @@ export function fakeDisk(total: number) { type Loose = any; export interface Harness { + module: ReturnType; db: Kysely; ctx: ServerContext; d: AoDeps; @@ -213,6 +214,7 @@ export async function createHarness(name: string, overrides: AoOverrides = {}): }); const d = module.deps(); return { + module, db, ctx, d, diff --git a/apps/server/src/modules/acquisition-organize/capabilities.ts b/apps/server/src/modules/acquisition-organize/capabilities.ts index de5fa904..3ba43ad9 100644 --- a/apps/server/src/modules/acquisition-organize/capabilities.ts +++ b/apps/server/src/modules/acquisition-organize/capabilities.ts @@ -180,7 +180,7 @@ export function registerAoCapabilities(registry: CapabilityRegistry, deps: () => throw new CapabilityError("precondition", `计划状态为 ${plan.status},不能提交`); } // 与自动路径同一幂等键:同一计划只会有一个 submitDownload 任务 - const taskId = await enqueueTracked(d, { + const { taskId } = await enqueueTracked(d, { kind: "submitDownload", idempotencyKey: `plan:${plan.id}`, input: { acquisitionPlanId: plan.id }, diff --git a/apps/server/src/modules/acquisition-organize/index.ts b/apps/server/src/modules/acquisition-organize/index.ts index edcdb828..2f5c74d4 100644 --- a/apps/server/src/modules/acquisition-organize/index.ts +++ b/apps/server/src/modules/acquisition-organize/index.ts @@ -1,4 +1,5 @@ import { type EventPayload, EventType } from "@umt/contracts"; +import { sql } from "@umt/db"; import type { ServerContext, ServerModule } from "../types.ts"; import { registerAoCapabilities } from "./capabilities.ts"; import { registerAoTasks } from "./handlers.ts"; @@ -9,10 +10,8 @@ import { registerDownloadRoutes } from "./routes/downloads.ts"; import { registerLibraryRoutes } from "./routes/library.ts"; import { registerOrganizationRoutes } from "./routes/organization.ts"; import { registerRuleRoutes } from "./routes/rules.ts"; -import { type AoDeps, automationBlock } from "./services/common.ts"; -import { cleanupDownloadCache, sweepReservedPlans, syncDownloads } from "./services/downloads.ts"; -import { executeRecord, planImport } from "./services/organize.ts"; -import { reconcileLibrary } from "./services/reconcile.ts"; +import { type AoDeps, automationBlock, enqueueSweep, enqueueTracked, timeSlot } from "./services/common.ts"; +import { SWEEP_SLOT_MS, sweepReservedPlans, syncDownloads } from "./services/downloads.ts"; import { loadVolumes, usedPercent } from "./services/volumes.ts"; import { errMessage } from "./util.ts"; @@ -25,33 +24,49 @@ import { errMessage } from "./util.ts"; export type AoOverrides = Partial> & { options?: Partial }; -/** 模块工厂:测试注入 mock qB / 固定磁盘用量 / 假种子来源;正式装配用 acquisitionOrganizeModule */ +/** 超阈值自动清理的时间槽(同槽只入队一次) */ +const CLEANUP_SLOT_MS = 5 * 60_000; + +/** + * 模块工厂:测试注入 mock qB / 固定磁盘用量 / 假种子来源;正式装配用 acquisitionOrganizeModule。 + * + * 后台只做编排,两个独立定时器互不阻塞: + * sync —— qB 同步 + 缓存阈值暂停/恢复(syncDownloads),只有轻量 HTTP 与 DB 写; + * plan —— 阈值判断后入队 cleanupDownloadCache、巡检补投 submitDownload / verifyAndImport、按周期入队 reconcileLibrary。 + * 复制/删种/全量 stat 这些重活都进任务引擎(租约 + attemptToken),不在循环里 await。 + */ export function createAcquisitionOrganizeModule(overrides: AoOverrides = {}): ServerModule & { deps(): AoDeps } { let deps: AoDeps | null = null; - let loop: ReturnType | null = null; + const timers = new Set>(); let stopped = false; - let lastReconcile = 0; + let lastReconcileSlot: string | null = null; // register* 阶段只保存 ctx;端口在首次使用时组装(ctx 其它字段此时才完整) const lazy = (ctx: ServerContext): AoDeps => { deps ??= { ctx, ports: createPorts(ctx, overrides) }; return deps; }; - /** worker 后台一轮:同步 qB → 阈值 → 必要时清理 → 补提交 → 补入库 → 低频资产对账 */ - const tick = async (d: AoDeps) => { - await syncDownloads(d); + /** 编排一轮:阈值 → 入队清理;补投提交/入库;周期入队资产对账 */ + const planTick = async (d: AoDeps) => { const blocked = await automationBlock(d); if (!blocked) { const cache = await d.ctx.settings.get("cache"); const over = (await loadVolumes(d)).some((v) => (usedPercent(v) ?? 0) >= cache.hardThresholdPercent); - if (over) await cleanupDownloadCache(d); + if (over) await enqueueThresholdCleanup(d); await sweepReservedPlans(d); await sweepPendingImports(d); } const every = d.ports.options.reconcileIntervalMs; - if (every > 0 && Date.now() - lastReconcile >= every) { - lastReconcile = Date.now(); - await reconcileLibrary(d); + if (every > 0) { + const slot = timeSlot(d.ports.now(), every); + if (slot !== lastReconcileSlot) { + await enqueueTracked(d, { + kind: "reconcileLibrary", + idempotencyKey: `auto:${slot}`, + input: {}, + }); + lastReconcileSlot = slot; + } } }; @@ -94,30 +109,64 @@ export function createAcquisitionOrganizeModule(overrides: AoOverrides = {}): Se const d = lazy(ctx); if (!d.ports.options.backgroundLoops) return; stopped = false; - const schedule = async () => { - if (stopped) return; + const intervalMs = async () => { const cache = await ctx.settings.get("cache").catch(() => null); - const ms = d.ports.options.syncIntervalMs ?? (cache?.checkIntervalSec ?? 60) * 1000; - loop = setTimeout(async () => { - try { - await tick(d); - } catch (err) { - ctx.logger.error({ err: { message: errMessage(err) } }, "acquisition-organize 后台轮询失败"); - } - void schedule(); - }, ms); + return d.ports.options.syncIntervalMs ?? (cache?.checkIntervalSec ?? 60) * 1000; }; - void schedule(); + /** 自调度循环:上一轮结束后再排下一轮,单个循环内不重入;两个循环互不等待 */ + const loop = (name: string, fn: (d: AoDeps) => Promise, period: () => Promise) => { + const schedule = async () => { + if (stopped) return; + const t = setTimeout( + async () => { + timers.delete(t); + try { + await fn(d); + } catch (err) { + ctx.logger.error( + { err: { message: errMessage(err) }, loop: name }, + "acquisition-organize 后台轮询失败", + ); + } + void schedule(); + }, + await period(), + ); + timers.add(t); + if (stopped) clearTimeout(t); + }; + void schedule(); + }; + loop("sync", syncDownloads, intervalMs); + loop("plan", planTick, async () => d.ports.options.planIntervalMs ?? (await intervalMs())); }, stop() { stopped = true; - if (loop) clearTimeout(loop); + for (const t of timers) clearTimeout(t); + timers.clear(); }, }; return module; } -/** 下载已完成、没有整理记录(入库事件被总开关延后等)的纳管任务:直接补做入库 */ +/** 超阈值:按时间槽入队一次 cleanupDownloadCache;上一次清理仍在进行则跳过 */ +async function enqueueThresholdCleanup(d: AoDeps): Promise { + const busy = await d.ctx.db + .selectFrom("task") + .select("id") + .where("kind", "=", "cleanupDownloadCache") + .where("status", "in", ["pending", "running"]) + .where("updated_at", ">", new Date(d.ports.now().getTime() - 30 * 60_000)) + .executeTakeFirst(); + if (busy) return; + await enqueueTracked(d, { + kind: "cleanupDownloadCache", + idempotencyKey: `threshold:${timeSlot(d.ports.now(), CLEANUP_SLOT_MS)}`, + input: {}, + }); +} + +/** 下载已完成、没有整理记录(入库事件被总开关延后等)的纳管任务:入队 verifyAndImport(业务行 import:) */ async function sweepPendingImports(d: AoDeps): Promise { const rows = await d.ctx.db .selectFrom("download_task as dt") @@ -125,7 +174,8 @@ async function sweepPendingImports(d: AoDeps): Promise { .where("dt.status", "=", "completed") .where("dt.managed", "=", true) .where("dt.removed_at", "is", null) - .where("dt.updated_at", "<", new Date(d.ports.now().getTime() - 5 * 60_000)) + // 用 completed_at:qB 同步每轮都会写 download_task(updated_at 触发器随之刷新),按 updated_at 永远选不中 + .where(sql`coalesce(dt.completed_at, dt.updated_at)`, "<", new Date(d.ports.now().getTime() - 5 * 60_000)) .where((eb) => eb.not( eb.exists( @@ -135,14 +185,24 @@ async function sweepPendingImports(d: AoDeps): Promise { ) .limit(10) .execute(); + const slot = timeSlot(d.ports.now(), SWEEP_SLOT_MS); let n = 0; for (const r of rows) { try { - const p = await planImport(d, { downloadTaskId: r.id, trigger: "auto", actorType: "system", actorId: null }); - if (p.recordId && (p.status === "running" || p.status === "queued")) await executeRecord(d, p.recordId); - n++; + const q = await enqueueSweep( + d, + { + kind: "verifyAndImport", + idempotencyKey: `import:${r.id}`, + input: { downloadTaskId: r.id }, + subjectType: "download_task", + subjectId: r.id, + }, + slot, + ); + if (q) n++; } catch (err) { - d.ctx.logger.warn({ downloadTaskId: r.id, err: { message: errMessage(err) } }, "补入库失败"); + d.ctx.logger.warn({ downloadTaskId: r.id, err: { message: errMessage(err) } }, "补入库入队失败"); } } return n; diff --git a/apps/server/src/modules/acquisition-organize/ports.ts b/apps/server/src/modules/acquisition-organize/ports.ts index 7fba8b29..445fc693 100644 --- a/apps/server/src/modules/acquisition-organize/ports.ts +++ b/apps/server/src/modules/acquisition-organize/ports.ts @@ -23,6 +23,10 @@ export type { DownloaderInstanceRow, MediaServerInstanceRow, SiteRow, SourceRele export interface AoOptions { /** 下载同步/阈值巡检周期;缺省按 settings.cache.checkIntervalSec */ syncIntervalMs?: number; + /** 编排循环(阈值清理入队/巡检补投/对账入队)周期;缺省同 syncIntervalMs */ + planIntervalMs?: number; + /** 手动整理按路径递归遍历的文件数上限(超出截断;dryRun 返回 truncated,正式执行拒绝) */ + walkMaxFiles: number; /** worker 启动时是否跑后台循环(测试里手动调用) */ backgroundLoops: boolean; /** 下载中且字节不增长超过该时长 → 记证据 + 待处理(不换源) */ @@ -59,6 +63,7 @@ export const DEFAULT_OPTIONS: AoOptions = { newTitleWindowDays: 30, missingConfirmAfter: 3, reconcileIntervalMs: 6 * 3600_000, + walkMaxFiles: 5000, }; const str = (v: unknown) => (typeof v === "string" && v.length > 0 ? v : undefined); @@ -137,7 +142,7 @@ export function createPorts( ...(str(settings.username) ? { username: str(settings.username)! } : {}), }); }, - options: { ...DEFAULT_OPTIONS }, + options: { ...DEFAULT_OPTIONS, walkMaxFiles: ctx.config?.ORGANIZE_WALK_MAX_FILES ?? DEFAULT_OPTIONS.walkMaxFiles }, }; const { options, ...rest } = overrides; return { ...ports, ...rest, options: { ...ports.options, ...options } }; diff --git a/apps/server/src/modules/acquisition-organize/routes/downloads.ts b/apps/server/src/modules/acquisition-organize/routes/downloads.ts index e87fea6c..d99c7dd3 100644 --- a/apps/server/src/modules/acquisition-organize/routes/downloads.ts +++ b/apps/server/src/modules/acquisition-organize/routes/downloads.ts @@ -2,10 +2,9 @@ import { type AcquisitionPlan, Api, type DownloadTask } from "@umt/contracts"; import { z } from "zod"; import { ApiException } from "../../../http/errors.ts"; import type { ModuleApp } from "../../types.ts"; -import type { AoDeps } from "../services/common.ts"; -import { cleanupDownloadCache, userPause, userResume } from "../services/downloads.ts"; -import { reconcileLibrary } from "../services/reconcile.ts"; -import { notFound, toApi, toApiList } from "../util.ts"; +import { type AoDeps, enqueueTracked, timeSlot } from "../services/common.ts"; +import { userPause, userResume } from "../services/downloads.ts"; +import { actorOf, notFound, toApi, toApiList } from "../util.ts"; import { mount, paginate } from "./helpers.ts"; // 下载计划 / 下载任务(含用户暂停与恢复)/ 手动触发缓存清理与资产对账。 @@ -83,18 +82,28 @@ export function registerDownloadRoutes(app: ModuleApp, d: AoDeps) { { tag: TAG }, ); + // 清理/对账是重活(删种删数据、全量 stat):入队返回 {taskId, runId},由任务页 / SSE 追踪。 + // 幂等键按分钟槽:连点只入队一次。 + const accepted = z.object({ taskId: z.uuid(), runId: z.string() }); + const minute = () => timeSlot(d.ports.now(), 60_000); mount( app, { method: "post", path: "/download-cache/cleanup", scope: "admin", - summary: "立即按阈值清理下载缓存(只清已验证入库且满足保留/H&R 的纳管任务)", + summary: "入队一轮缓存清理(只清已验证入库且满足保留/H&R 的纳管任务);返回任务 id,进度见任务页", body: z.object({ downloaderInstanceId: z.uuid().optional() }), - response: z.object({ report: z.record(z.string(), z.unknown()) }), + response: accepted, }, - async (_c, { body }) => ({ report: await cleanupDownloadCache(d, body) }), - { tag: TAG, status: 200 }, + async (c, { body }) => + enqueueTracked(d, { + kind: "cleanupDownloadCache", + idempotencyKey: `manual:${body.downloaderInstanceId ?? "all"}:${minute()}`, + input: body.downloaderInstanceId ? { downloaderInstanceId: body.downloaderInstanceId } : {}, + triggeredBy: actorOf(c), + }), + { tag: TAG, status: 202 }, ); mount( app, @@ -102,14 +111,17 @@ export function registerDownloadRoutes(app: ModuleApp, d: AoDeps) { method: "post", path: "/library/reconcile", scope: "admin", - summary: "立即做一轮资产对账(库根哨兵 + 增量 stat;不删记录、不补下载)", + summary: "入队一轮资产对账(库根哨兵 + 增量 stat;不删记录、不补下载);返回任务 id,进度见任务页", body: z.object({ libraryRootId: z.uuid().optional() }), - response: z.object({ runId: z.uuid(), stats: z.record(z.string(), z.unknown()) }), + response: accepted, }, - async (_c, { body }) => { - const r = await reconcileLibrary(d, body); - return { runId: r.runId, stats: r.stats }; - }, - { tag: TAG, status: 200 }, + async (c, { body }) => + enqueueTracked(d, { + kind: "reconcileLibrary", + idempotencyKey: `manual:${body.libraryRootId ?? "all"}:${minute()}`, + input: body.libraryRootId ? { libraryRootId: body.libraryRootId } : {}, + triggeredBy: actorOf(c), + }), + { tag: TAG, status: 202 }, ); } diff --git a/apps/server/src/modules/acquisition-organize/routes/helpers.ts b/apps/server/src/modules/acquisition-organize/routes/helpers.ts index a45183f5..a758a410 100644 --- a/apps/server/src/modules/acquisition-organize/routes/helpers.ts +++ b/apps/server/src/modules/acquisition-organize/routes/helpers.ts @@ -36,7 +36,7 @@ export function mount( app: ModuleApp, maybe: RouteDef | undefined, handler: Handler, - opts: { tag?: string; status?: 200 | 201 } = {}, + opts: { tag?: string; status?: 200 | 201 | 202 } = {}, ) { if (!maybe) throw new Error("contracts 未定义该端点"); const def = maybe; diff --git a/apps/server/src/modules/acquisition-organize/services/acquisition.ts b/apps/server/src/modules/acquisition-organize/services/acquisition.ts index 993cc832..fa0b0a51 100644 --- a/apps/server/src/modules/acquisition-organize/services/acquisition.ts +++ b/apps/server/src/modules/acquisition-organize/services/acquisition.ts @@ -502,7 +502,7 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon subjectType: "acquisition_plan", subjectId: planId, }).then( - (taskId) => db.updateTable("acquisition_plan").set({ task_id: taskId }).where("id", "=", planId).execute(), + ({ taskId }) => db.updateTable("acquisition_plan").set({ task_id: taskId }).where("id", "=", planId).execute(), (err) => d.ctx.logger.warn({ err: { message: (err as Error).message }, planId }, "submitDownload 入队失败,留给巡检补投"), ); diff --git a/apps/server/src/modules/acquisition-organize/services/common.ts b/apps/server/src/modules/acquisition-organize/services/common.ts index 3ea3c3fa..3363c003 100644 --- a/apps/server/src/modules/acquisition-organize/services/common.ts +++ b/apps/server/src/modules/acquisition-organize/services/common.ts @@ -123,21 +123,56 @@ export async function markTask( .execute(); } -/** 建业务任务行 + 引擎入队(幂等键一致);返回 task.id */ -export async function enqueueTracked(d: AoDeps, t: Omit): Promise { +/** + * 建业务任务行 + 引擎入队。引擎幂等键缺省 `:`(与业务行一致); + * 巡检补投等需要"同一业务行、按时间槽重投"的场景传 engineKey。 + */ +export async function enqueueTracked( + d: AoDeps, + t: Omit, + opts: { engineKey?: string } = {}, +): Promise<{ taskId: string; runId: string }> { const taskId = await ensureTaskRow(d.ctx.db, t); const { runId } = await d.ctx.engine.enqueue( t.kind, { ...t.input, taskId }, { - idempotencyKey: `${t.kind}:${t.idempotencyKey}`, + idempotencyKey: opts.engineKey ?? `${t.kind}:${t.idempotencyKey}`, traceId: taskId, }, ); await d.ctx.db.updateTable("task").set({ engine_run_id: runId }).where("id", "=", taskId).execute(); - return taskId; + return { taskId, runId }; } +const ACTIVE_RUN = new Set(["pending", "running", "waiting"]); + +/** + * 巡检补投:同一业务行(kind + idempotencyKey)上次的引擎 run 仍在进行就跳过; + * 否则按时间槽(slot)生成引擎幂等键重投(引擎幂等键 24h 内有效,同槽重复巡检只入队一次)。 + * 返回 null 表示跳过。 + */ +export async function enqueueSweep( + d: AoDeps, + t: Omit, + slot: string, +): Promise<{ taskId: string; runId: string } | null> { + const prev = await d.ctx.db + .selectFrom("task") + .select(["engine_run_id"]) + .where("kind", "=", t.kind) + .where("idempotency_key", "=", t.idempotencyKey) + .executeTakeFirst(); + if (prev?.engine_run_id) { + const run = await d.ctx.engine.get(prev.engine_run_id).catch(() => null); + if (run && ACTIVE_RUN.has(run.status)) return null; + } + return enqueueTracked(d, t, { engineKey: `${t.kind}:${t.idempotencyKey}:${slot}` }); +} + +/** 时间槽标签:floor(now / periodMs) */ +export const timeSlot = (now: Date, periodMs: number) => String(Math.floor(now.getTime() / Math.max(1, periodMs))); + // ── 实例拓扑:路径映射、保存路径、缓存卷 ── export interface InstanceTopology { diff --git a/apps/server/src/modules/acquisition-organize/services/downloads.ts b/apps/server/src/modules/acquisition-organize/services/downloads.ts index eafbbbc8..4a9cbadd 100644 --- a/apps/server/src/modules/acquisition-organize/services/downloads.ts +++ b/apps/server/src/modules/acquisition-organize/services/downloads.ts @@ -10,10 +10,12 @@ import { automationBlock, downloaderPathOf, type Exec, + enqueueSweep, hostPathOf, type InstanceTopology, loadInstances, loadLibraryRoots, + timeSlot, upsertDecision, } from "./common.ts"; import { hnrSatisfied, retentionElapsed } from "./retention.ts"; @@ -260,6 +262,13 @@ async function failPlan(d: AoDeps, planId: string, slotId: string, reason: strin } /** 巡检:reserved 但没有下载任务的计划(入队失败、被阈值/总开关延后)→ 条件满足时直接补提交 */ +/** 巡检补投的时间槽:同一计划/下载在一个槽内最多重投一次 */ +export const SWEEP_SLOT_MS = 5 * 60_000; + +/** + * 已预留但一直没有下载任务的计划:入队 submitDownload(与自动路径同一业务行 plan:),不在巡检里直接提交。 + * 上次的 run 仍在进行则跳过;返回本轮入队数。 + */ export async function sweepReservedPlans(d: AoDeps): Promise { if (await automationBlock(d)) return 0; const stale = await d.ctx.db @@ -271,13 +280,32 @@ export async function sweepReservedPlans(d: AoDeps): Promise { .where("p.updated_at", "<", new Date(d.ports.now().getTime() - 60_000)) .limit(20) .execute(); + const slot = timeSlot(d.ports.now(), SWEEP_SLOT_MS); let n = 0; for (const p of stale) { - const r = await submitDownload(d, p.id).catch((err) => { - d.ctx.logger.warn({ planId: p.id, err: { message: errMessage(err) } }, "补提交失败"); - return null; - }); - if (r?.status === "submitted") n++; + try { + const r = await enqueueSweep( + d, + { + kind: "submitDownload", + idempotencyKey: `plan:${p.id}`, + input: { acquisitionPlanId: p.id }, + subjectType: "acquisition_plan", + subjectId: p.id, + }, + slot, + ); + if (!r) continue; + n++; + await d.ctx.db + .updateTable("acquisition_plan") + .set({ task_id: r.taskId }) + .where("id", "=", p.id) + .where("task_id", "is", null) + .execute(); + } catch (err) { + d.ctx.logger.warn({ planId: p.id, err: { message: errMessage(err) } }, "补提交入队失败"); + } } return n; } diff --git a/apps/web/app/api/downloads.ts b/apps/web/app/api/downloads.ts index b390e1f7..86da4215 100644 --- a/apps/web/app/api/downloads.ts +++ b/apps/web/app/api/downloads.ts @@ -102,14 +102,14 @@ export function useCacheVolumes() { }); } -/** 手动清理:只清已验证入库且满足保留/H&R 的纳管任务 */ +/** 手动清理:服务端入队 cleanupDownloadCache(只清已验证入库且满足保留/H&R 的纳管任务),返回任务 id */ export function useCacheCleanup() { const qc = useQueryClient(); return useMutation({ mutationFn: (downloaderInstanceId?: string) => apiMutate("/api/v1/download-cache/cleanup", { body: downloaderInstanceId ? { downloaderInstanceId } : {}, - }).then((d) => z.object({ report: z.record(z.string(), z.unknown()) }).parse(d)), + }).then((d) => z.object({ taskId: z.string(), runId: z.string() }).parse(d)), onSuccess: () => qc.invalidateQueries({ queryKey: downloadKeys.all }), }); } diff --git a/apps/web/app/components/downloads/cache-volumes.tsx b/apps/web/app/components/downloads/cache-volumes.tsx index 800986ec..1cddd9f5 100644 --- a/apps/web/app/components/downloads/cache-volumes.tsx +++ b/apps/web/app/components/downloads/cache-volumes.tsx @@ -1,5 +1,6 @@ import type { Dashboard } from "@umt/contracts"; import { Brush, HardDrive } from "lucide-react"; +import { useNavigate } from "react-router"; import { toast } from "sonner"; import { useCacheCleanup, useCacheVolumes } from "~/api/downloads.ts"; import { ConfirmDialog } from "~/components/console/confirm-dialog.tsx"; @@ -79,6 +80,7 @@ function VolumeBar({ v }: { v: Volume }) { export function CacheVolumes() { const vols = useCacheVolumes(); const cleanup = useCacheCleanup(); + const navigate = useNavigate(); return (
@@ -96,10 +98,8 @@ export function CacheVolumes() { onConfirm={() => cleanup.mutateAsync(undefined).then( (r) => { - const vols = (r.report as { volumes?: { removed?: unknown[]; skipped?: unknown[] }[] }).volumes ?? []; - const removed = vols.reduce((n, v) => n + (v.removed?.length ?? 0), 0); - const skipped = vols.reduce((n, v) => n + (v.skipped?.length ?? 0), 0); - toast.success(`清理完成:移除 ${removed} 个任务的缓存${skipped ? `,${skipped} 个因保留/H&R 跳过` : ""}`); + toast.success(`已排队清理任务 ${r.taskId.slice(0, 8)}`, { description: "后台执行,结果见任务详情" }); + navigate(`/console/engine/${r.runId}`); }, (e) => toast.error(errorMessage(e)), ) diff --git a/packages/testkit/src/qbittorrent/mock-qb.ts b/packages/testkit/src/qbittorrent/mock-qb.ts index 82cf292c..055ca26f 100644 --- a/packages/testkit/src/qbittorrent/mock-qb.ts +++ b/packages/testkit/src/qbittorrent/mock-qb.ts @@ -51,6 +51,8 @@ export interface MockQbCall { query: Record; form: Record; status: number; + /** 收到请求的时刻(Date.now()) */ + at: number; } export interface MockQbFault { @@ -533,7 +535,7 @@ export async function createMockQb(opts: MockQbOptions = {}): Promise { const body = await readBody(req); const ct = req.headers["content-type"] ?? ""; const form = ct.includes("urlencoded") ? Object.fromEntries(new URLSearchParams(body.toString())) : {}; - const call: MockQbCall = { method: req.method ?? "GET", endpoint, query, form, status: 0 }; + const call: MockQbCall = { method: req.method ?? "GET", endpoint, query, form, status: 0, at: Date.now() }; calls.push(call); const send = (status: number, text: string, headers: Record = {}) => { call.status = status;