From 887843f7779fd85b412149cfa8b0e01cd673c270 Mon Sep 17 00:00:00 2001 From: root Date: Mon, 28 Sep 2026 00:32:57 +0800 Subject: [PATCH] =?UTF-8?q?fix(server,test):=20W6-A=20stabilize=20CI=20?= =?UTF-8?q?=E2=80=94=20sync=20lost=20update,=20atomic=20recovery=20summary?= =?UTF-8?q?,=20serial=20PG=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - downloads.ts observe: re-read download_task FOR UPDATE inside the tx and merge from the locked row; qB sync could overwrite imported back to importing when verifyAndImport committed between the sync read and write (E8 flake). - agent finalize: accept summaryPrefix so the recovery verdict is written with the terminal status in one update (E6 read a budget_exceeded run before the "未恢复:…" annotation landed); recover-summary step kept, now idempotent no-op. - organize.ts: drop local slotComplete and the once-subscription completion write; completion/history are decided only by applySlotEvent(verify_passed) (W5-C follow-up, removes double write). Manual slot organize uses the shared subscriptions.slotComplete. - vitest: server split into unit (parallel) and pg (file-serial, groupOrder 1) projects; packages/tasks file-serial; explicit worker concurrency in e2e. - docs: W6 stabilization record in 04-v0.1-acceptance.md. Test expectations unchanged. Full `deno task ci` x3 green: 369s / 341s / 337s. Co-Authored-By: Claude Opus 5.5 --- .../services/downloads.ts | 163 ++++++++++-------- .../acquisition-organize/services/organize.ts | 69 +------- apps/server/src/modules/agent/service.ts | 3 + apps/server/src/modules/agent/tasks.ts | 5 +- apps/server/test/e2e/fixtures/worker.ts | 9 +- apps/server/test/e2e/harness.ts | 2 + apps/server/vitest.config.ts | 35 +++- deno.lock | 13 ++ docs/plan/04-v0.1-acceptance.md | 33 ++++ packages/tasks/vitest.config.ts | 9 + 10 files changed, 198 insertions(+), 143 deletions(-) create mode 100644 packages/tasks/vitest.config.ts diff --git a/apps/server/src/modules/acquisition-organize/services/downloads.ts b/apps/server/src/modules/acquisition-organize/services/downloads.ts index 6ac25fd4..f6aa3a0f 100644 --- a/apps/server/src/modules/acquisition-organize/services/downloads.ts +++ b/apps/server/src/modules/acquisition-organize/services/downloads.ts @@ -659,7 +659,29 @@ async function syncInstance(d: AoDeps, topo: InstanceTopology, stats: SyncStats) const sizeOf = (t: QbTorrent) => (t.totalSize ?? t.size) || null; const downloadedOf = (t: QbTorrent) => Math.max(0, Math.round((sizeOf(t) ?? 0) - t.amountLeft)); -async function observe(d: AoDeps, row: DownloadTaskRow, t: QbTorrent, now: Date, stats: SyncStats): Promise { +async function observe(d: AoDeps, seen: DownloadTaskRow, t: QbTorrent, now: Date, stats: SyncStats): Promise { + // 行锁内按最新行合并:同步读到行之后入库事务可能已把 importing 推进到 imported, + // 用同步开始时的快照整行覆写会把状态写回 importing(丢失更新) + return d.ctx.db.transaction().execute(async (trx) => { + const row = await trx + .selectFrom("download_task") + .selectAll() + .where("id", "=", seen.id) + .forUpdate() + .executeTakeFirst(); + if (!row || row.removed_at || row.status === "removed") return false; + return observeLocked(d, trx, row, t, now, stats); + }); +} + +async function observeLocked( + d: AoDeps, + trx: Exec, + row: DownloadTaskRow, + t: QbTorrent, + now: Date, + stats: SyncStats, +): Promise { const cur = { status: row.status, pauseReason: row.pause_reason }; const obs = mapQbState( { state: t.rawState, progress: t.progress, amountLeft: t.amountLeft }, @@ -691,80 +713,71 @@ async function observe(d: AoDeps, row: DownloadTaskRow, t: QbTorrent, now: Date, } const becameComplete = merged.status === "completed" && !["completed", "importing", "imported"].includes(row.status); const changed = merged.status !== row.status || merged.pauseReason !== row.pause_reason; - await d.ctx.db.transaction().execute(async (trx) => { - await trx - .updateTable("download_task") - .set({ - status: merged.status, - pause_reason: merged.status === "paused" ? merged.pauseReason : null, - raw_state: t.rawState, - progress: Math.min(1, Math.max(0, t.progress)), - size_bytes: sizeOf(t), - downloaded_bytes: downloaded, - remaining_bytes: Math.max(0, Math.round(t.amountLeft)), - ratio: t.ratio, - seeding_time_sec: t.seedingTimeSec, - save_path: t.savePath, - content_path: t.contentPath, - name: t.name, - tags: t.tags, - last_progress_at: lastProgressAt, - no_progress_since: noProgressSince, - last_synced_at: now, - ...(becameComplete ? { completed_at: now } : {}), - }) - .where("id", "=", row.id) - .execute(); - if (changed) - await emitStatus( - d, - trx, - row.id, - row.status, - merged.status, - merged.status === "paused" ? merged.pauseReason : null, - ); - if (becameComplete) { - stats.completed.push(row.id); - const plan = row.acquisition_plan_id - ? await trx - .selectFrom("acquisition_plan") - .select(["id", "slot_id"]) - .where("id", "=", row.acquisition_plan_id) - .executeTakeFirst() - : null; - if (plan) await applySlotEvent(d, trx, plan.slot_id, { type: "download_completed" }, { planId: plan.id }); - } - if (stalledNow) { - // 无进展:只记证据 + 待处理,保留原任务,不换源(REQ §8.4.2) - await upsertDecision(d, trx, { - kind: "download_stalled", - title: `下载无进展:${t.name}`, - summary: `已 ${Math.round((now.getTime() - (noProgressSince?.getTime() ?? now.getTime())) / 60000)} 分钟没有新增字节`, - subjectType: "download_task", - subjectId: row.id, - dedupeKey: `download_stalled:${row.id}`, - evidence: { - rawState: t.rawState, - progress: t.progress, - downloadedBytes: downloaded, - noProgressSince: noProgressSince?.toISOString(), - action: "record_evidence", - }, - }); - } - if ((merged.status === "failed" || merged.status === "missing") && changed) { - await upsertDecision(d, trx, { - kind: "download_failed", - title: `下载异常:${t.name}`, - summary: `qB 状态 ${t.rawState}`, - subjectType: "download_task", - subjectId: row.id, - dedupeKey: `download_failed:${row.id}`, - evidence: { rawState: t.rawState, progress: t.progress }, - }); - } - }); + await trx + .updateTable("download_task") + .set({ + status: merged.status, + pause_reason: merged.status === "paused" ? merged.pauseReason : null, + raw_state: t.rawState, + progress: Math.min(1, Math.max(0, t.progress)), + size_bytes: sizeOf(t), + downloaded_bytes: downloaded, + remaining_bytes: Math.max(0, Math.round(t.amountLeft)), + ratio: t.ratio, + seeding_time_sec: t.seedingTimeSec, + save_path: t.savePath, + content_path: t.contentPath, + name: t.name, + tags: t.tags, + last_progress_at: lastProgressAt, + no_progress_since: noProgressSince, + last_synced_at: now, + ...(becameComplete ? { completed_at: now } : {}), + }) + .where("id", "=", row.id) + .execute(); + if (changed) + await emitStatus(d, trx, row.id, row.status, merged.status, merged.status === "paused" ? merged.pauseReason : null); + if (becameComplete) { + stats.completed.push(row.id); + const plan = row.acquisition_plan_id + ? await trx + .selectFrom("acquisition_plan") + .select(["id", "slot_id"]) + .where("id", "=", row.acquisition_plan_id) + .executeTakeFirst() + : null; + if (plan) await applySlotEvent(d, trx, plan.slot_id, { type: "download_completed" }, { planId: plan.id }); + } + if (stalledNow) { + // 无进展:只记证据 + 待处理,保留原任务,不换源(REQ §8.4.2) + await upsertDecision(d, trx, { + kind: "download_stalled", + title: `下载无进展:${t.name}`, + summary: `已 ${Math.round((now.getTime() - (noProgressSince?.getTime() ?? now.getTime())) / 60000)} 分钟没有新增字节`, + subjectType: "download_task", + subjectId: row.id, + dedupeKey: `download_stalled:${row.id}`, + evidence: { + rawState: t.rawState, + progress: t.progress, + downloadedBytes: downloaded, + noProgressSince: noProgressSince?.toISOString(), + action: "record_evidence", + }, + }); + } + if ((merged.status === "failed" || merged.status === "missing") && changed) { + await upsertDecision(d, trx, { + kind: "download_failed", + title: `下载异常:${t.name}`, + summary: `qB 状态 ${t.rawState}`, + subjectType: "download_task", + subjectId: row.id, + dedupeKey: `download_failed:${row.id}`, + evidence: { rawState: t.rawState, progress: t.progress }, + }); + } return changed; } diff --git a/apps/server/src/modules/acquisition-organize/services/organize.ts b/apps/server/src/modules/acquisition-organize/services/organize.ts index cab35328..ebcb9fe8 100644 --- a/apps/server/src/modules/acquisition-organize/services/organize.ts +++ b/apps/server/src/modules/acquisition-organize/services/organize.ts @@ -24,7 +24,6 @@ import { type ExistingPolicy, type ExistingTarget, findOtherVersions, - movieCompletion, type NewBinding, type OrganizePlan, overrideFor, @@ -34,7 +33,6 @@ import { type QualityFacts, resolveTargetDirectory, type SourceFile, - unitCompletion, } from "@umt/domain"; import { organizationStatusFromFiles, organizePlanToFileResults, toDomainPathBinding } from "@umt/mapping"; import { withAttemptToken } from "@umt/tasks"; @@ -45,6 +43,7 @@ import { applySlotEvent } from "./downloads.ts"; import { PathGuardError } from "./organize-guard.ts"; import { type FrozenTransfer, finalizeTransfer, planTransfer, type TransferSpec } from "./organize-transfer.ts"; import { hnrSatisfied } from "./retention.ts"; +import { slotComplete as sharedSlotComplete } from "./subscriptions.ts"; // 入库:冻结整理计划(organization_record + 文件明细)→ 逐文件可恢复执行(操作账本 fencing)→ // 资产/位置/路径绑定同事务登记 → Slot 推进 → Move 后移除 qB 任务记录(不删数据)。 @@ -1455,14 +1454,14 @@ export async function finalizeRecord(d: AoDeps, recordId: string): Promise { - const completion = (slot.completion ?? {}) as { kind?: string; cuts?: string[] }; - if (frozen.mediaType === "movie") { - const assets = await exec - .selectFrom("media_asset") - .select(["id", "content_edition"]) - .where("id", "in", newAssets) - .execute(); - return movieCompletion( - completion.cuts as never, - assets.map((a) => ({ - assetId: a.id, - verified: true, - ...(a.content_edition ? { cut: a.content_edition as never } : {}), - })), - ).complete; - } - // 剧集/综艺:按内容单元覆盖;单元未建档(总集数未知)时继续追踪,不声称完整 - let q = exec - .selectFrom("content_unit") - .select(["id", "status"]) - .where("media_entity_id", "=", slot.media_entity_id) - .where("kind", "=", "episode") - .where("content_type", "=", "main"); - if (slot.content_unit_id) q = q.where("parent_unit_id", "=", slot.content_unit_id); - const units = await q.where("status", "!=", "retired").execute(); - const required = units.map((u) => u.id); - if (required.length === 0) return false; - // 总集数只有在单元都已确认(元数据/人工)时才算已知;由发布标题或文件名临时建出的单元(provisional) - // 只说明"见过这些集",不说明"只有这些集":总集数未知时不判完整,订阅继续追踪(REQ §7.1.1) - const totalKnown = units.every((u) => u.status === "confirmed"); - if (!totalKnown) return false; - const cov = await sql<{ asset_id: string; unit_id: string }>` - select c.media_asset_id as asset_id, c.content_unit_id as unit_id from asset_unit_coverage c - join media_asset a on a.id = c.media_asset_id - where a.media_entity_id = ${slot.media_entity_id} and a.status = 'active'`.execute(exec); - const byAsset = new Map(); - for (const r of cov.rows) byAsset.set(r.asset_id, [...(byAsset.get(r.asset_id) ?? []), r.unit_id]); - return unitCompletion( - required, - [...byAsset].map(([assetId, fileUnits]) => ({ assetId, verified: true, fileUnits })), - { totalKnown }, - ).complete; -} - /** 按季号与集号区间找或建集单元(临时 provisional,挂在季单元下) */ async function ensureEpisodeUnits( exec: Exec, diff --git a/apps/server/src/modules/agent/service.ts b/apps/server/src/modules/agent/service.ts index a0743786..b2b7fa01 100644 --- a/apps/server/src/modules/agent/service.ts +++ b/apps/server/src/modules/agent/service.ts @@ -189,6 +189,7 @@ export async function finalize( runId: string, res: DriveResult, settled: AgentGoalReport | null, + opts: { summaryPrefix?: string } = {}, ): Promise<{ status: AgentRunStatus; summary: string }> { const { ctx } = env; const run = await ctx.db.selectFrom("agent_run").selectAll().where("id", "=", runId).executeTakeFirstOrThrow(); @@ -213,6 +214,8 @@ export async function finalize( } // 模型文本只作参考:附在业务核查结论之后,明确标注未经核查 if (res.finalText && res.status !== "failed") summary += `\n模型汇报(未经核查):${res.finalText}`; + // 系统核实结论(授权内恢复)与终态同一次写入:观察者看到终态时汇总已是最终文本 + if (opts.summaryPrefix) summary = `${opts.summaryPrefix}\n${summary}`; await ctx.db .insertInto("artifact") .values({ diff --git a/apps/server/src/modules/agent/tasks.ts b/apps/server/src/modules/agent/tasks.ts index 602b9382..ebc3a58f 100644 --- a/apps/server/src/modules/agent/tasks.ts +++ b/apps/server/src/modules/agent/tasks.ts @@ -90,8 +90,11 @@ export function registerAgentTasks(registry: TaskRegistry, env: () => AgentEnv) maxRounds: rounds, }); } - const out = await t.step.run("finalize", () => finalize(e, runId, res, settled)); + const out = await t.step.run("finalize", () => + finalize(e, runId, res, settled, recovery ? { summaryPrefix: recovery.summary } : {}), + ); if (recovery) { + // 已在 finalize 内带上核实结论;保留该步骤兼容旧 run 的步骤历史(已带前缀时幂等跳过) const rec = recovery; return t.step.run("recover-summary", () => annotateRunSummary(e.ctx, runId, rec)); } diff --git a/apps/server/test/e2e/fixtures/worker.ts b/apps/server/test/e2e/fixtures/worker.ts index 0f23fb5f..ad3131db 100644 --- a/apps/server/test/e2e/fixtures/worker.ts +++ b/apps/server/test/e2e/fixtures/worker.ts @@ -34,7 +34,14 @@ const worker = await startWorker({ logger, heartbeatMs: 60_000, bus, - runtime: { ctx: booted.ctx, modules, setRelay: booted.outbox.setRelay, outboxIntervalMs: 100, concurrency: 4 }, + runtime: { + ctx: booted.ctx, + modules, + setRelay: booted.outbox.setRelay, + outboxIntervalMs: 100, + concurrency: 4, + agentConcurrency: 2, + }, }); console.log(`worker ready pid=${Deno.pid}`); Deno.addSignalListener("SIGTERM", async () => { diff --git a/apps/server/test/e2e/harness.ts b/apps/server/test/e2e/harness.ts index 6cfbc305..c955d00d 100644 --- a/apps/server/test/e2e/harness.ts +++ b/apps/server/test/e2e/harness.ts @@ -189,7 +189,9 @@ export async function createE2E(name: string, opts: E2EOptions = {}): Promise