merge: fix/d-organize

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei
This commit is contained in:
rootandClaude Opus 5.5 committed 2026-10-03 21:11:22 +08:00
commit fe32520643
12 files changed
+713 -44

No files matched your search

@@ -165,7 +165,13 @@ export function registerAoTasks(registry: TaskRegistry, d: AoDeps) {
...(await executeRecord(d, recordId, { runId: t.runId, abortSignal: t.abortSignal })),
}));
},
(r) => (r.status === "failed" ? "failed" : r.status === "deferred" ? "waiting" : "succeeded"),
// 执行返回 running = 库根离线 / 建目录失败而暂停(记录写明原因并开了待处理,由整理巡检重新入队)
(r) =>
r.status === "failed"
? "failed"
: r.status === "deferred" || r.status === "running"
? "waiting"
: "succeeded",
);
},
}),
@@ -16,6 +16,7 @@ import { registerOrganizationRoutes } from "./routes/organization.ts";
import { registerRuleRoutes } from "./routes/rules.ts";
import { type AoDeps, automationBlock, enqueueSweep, enqueueTracked, timeSlot } from "./services/common.ts";
import { SWEEP_SLOT_MS, sweepReservedPlans, syncDownloads } from "./services/downloads.ts";
import { recoverStalledRecords } from "./services/organize-records.ts";
import { sweepDueSlots, sweepMetadataRefresh, sweepSubscriptionSearch } from "./services/subscriptions.ts";
import { organizeDecisionHook } from "./services/takeover.ts";
import { loadVolumes, usedPercent, volumeThresholds } from "./services/volumes.ts";
@@ -39,7 +40,7 @@ const CLEANUP_SLOT_MS = 5 * 60_000;
* 后台只做编排,start() 把各项周期登记到定时任务注册表(schedules 模块的调度器驱动,不同任务并行、同一任务不重入):
* downloads.sync —— qB 同步 + 缓存阈值暂停/恢复(syncDownloads),只有轻量 HTTP 与 DB 写;
* cache.threshold —— 超硬阈值入队 cleanupDownloadCache;
* downloads.orchestrate —— 巡检补投 submitDownload / verifyAndImport;
* downloads.orchestrate —— 巡检补投 submitDownload / verifyAndImport;整理记录卡在 running 的重投、Move 后补移除 qB 任务;
* slots.reevaluate / subscriptions.search / subscriptions.metadata —— Slot 到期重评、订阅周期搜索、总集数刷新;
* library.reconcile —— 按周期入队 reconcileLibrary。
* 复制/删种/全量 stat 这些重活都进任务引擎(租约 + attemptToken),不在调度里 await。
@@ -149,14 +150,20 @@ export function createAcquisitionOrganizeModule(overrides: AoOverrides = {}): Se
key: "downloads.orchestrate",
defaults: {
name: "补提交与补入库",
description: "补交没提交成功的下载,补整理下载完但没入库的任务",
description: "补交没提交成功的下载,补整理下载完但没入库的任务,重新排队卡住的整理",
group: "download",
schedule: every(planMs),
},
run: guarded(d, async (d) => {
const submitted = await sweepReservedPlans(d);
const imports = await sweepPendingImports(d);
return `补提交 ${submitted},补入库 ${imports}`;
const rec = await recoverStalledRecords(d);
return (
`补提交 ${submitted},补入库 ${imports}` +
(rec.requeued ? `,重排整理 ${rec.requeued}` : "") +
(rec.failed ? `,整理停止 ${rec.failed}` : "") +
(rec.removed ? `,移除下载任务 ${rec.removed}` : "")
);
}),
});
registerScheduledJob(ctx, {
@@ -13,11 +13,13 @@ import {
import type { OrganizationRecord } from "@umt/contracts";
import { sql } from "@umt/db";
import { withAttemptToken } from "@umt/tasks";
import { closeDecisions } from "../../sources-identity/service/common.ts";
import { errMessage, jsonb } from "../util.ts";
import { type AoDeps, upsertDecision } from "./common.ts";
// 入库:冻结整理计划(organization_record + 文件明细)→ 逐文件可恢复执行(操作账本 fencing)→
// 资产/位置/路径绑定同事务登记 → Slot 推进 → Move 后移除 qB 任务记录(不删数据)。
import { finalizeRecord } from "./organize-finalize.ts";
import { finalizeRecord, removeMovedDownload } from "./organize-finalize.ts";
import { emitRecord } from "./organize-plan.ts";
import type { FrozenPlan } from "./organize-types.ts";
import { cleanupReplacement, settleSamePathReplacement, stagedReplacementBackup } from "./replacement.ts";
import { scrapeAfterImport } from "./scrape.ts";
@@ -89,6 +91,17 @@ export async function executeRecord(
enabled: frozen.scrapeAfterImport,
});
}
// Move 成功但移除 qB 任务没成功(qB 暂时连不上 / 收尾后进程退出):重放时补做
if (rec.action === "move" && rec.status === "succeeded" && rec.download_task_id) {
const dt = await db
.selectFrom("download_task")
.select(["id", "instance_id", "torrent_hash"])
.where("id", "=", rec.download_task_id)
.where("removed_at", "is", null)
.where("managed", "=", true)
.executeTakeFirst();
if (dt) await removeMovedDownload(d, dt);
}
return {
recordId,
status: rec.status,
@@ -99,11 +112,51 @@ export async function executeRecord(
const frozen = rec.plan as unknown as FrozenPlan;
const fsr = fsRuntime(d);
// 根检查也走执行子进程:NFS/FUSE 挂死 → offline(timeout),按根离线重试,绝不当成文件缺失
// 离线 / 建目录失败不抛错(抛错会耗尽任务重试、记录卡在 running):记录保持 running 并写明原因、开待处理,
// 由整理巡检(recoverStalledRecords)在库根恢复后重新入队
const health = await fsr.proxy.checkRoot(frozen.rootPath, frozen.sentinel);
if (health.kind === "offline") {
throw new Error(`媒体库目录离线(${health.reason}),稍后重试`);
return holdRecord(d, rec, `${ROOT_OFFLINE}:${health.reason}`, {
kind: "other",
severity: "critical",
title: "媒体库目录离线,整理已暂停",
summary: `${frozen.rootPath} 不可访问或未挂载(${health.reason}),恢复后自动继续整理。文件不会因此被判为缺失`,
subjectType: "library_root",
subjectId: frozen.libraryRootId,
dedupeKey: `library_root_offline:${frozen.libraryRootId}`,
evidence: { reason: health.reason, recordId },
});
}
if (rec.action !== "in_place") {
try {
for (const dir of frozen.targetDirs) await mkdir(dir, { recursive: true });
} catch (err) {
const code = (err as { code?: string } | null)?.code ?? "EIO";
return holdRecord(d, rec, `${TARGET_DIR_UNAVAILABLE}:${code}`, {
kind: "other",
title: "整理目标目录建不出来",
summary: `${errMessage(err)};检查目录权限或挂载后会自动重试`,
subjectType: "organization_record",
subjectId: recordId,
dedupeKey: `organize_target_dir:${recordId}`,
evidence: { code, targetDirs: frozen.targetDirs },
});
}
}
// 之前因库根离线 / 建目录失败暂停过:清掉等待原因,库根离线事项随之关闭
if (isHoldReason(rec.reason)) {
await db.updateTable("organization_record").set({ reason: null }).where("id", "=", recordId).execute();
}
if (rec.reason?.startsWith(ROOT_OFFLINE)) {
await closeDecisions(
d.ctx,
db,
`library_root_offline:${frozen.libraryRootId}`,
"resolved",
{ key: "root_online", note: "媒体库目录已恢复", organizationRecordId: recordId },
"system",
);
}
if (rec.action !== "in_place") for (const dir of frozen.targetDirs) await mkdir(dir, { recursive: true });
// 写前顺带清本目录残留 partial(对照 MP:只清本目录、>24h;失败吞掉)。copy 路径在 copyWithVerify 内还会再清一次
if (rec.action !== "in_place" && fsr.partialMaxAgeMs > 0) {
for (const dir of frozen.targetDirs) {
@@ -392,6 +445,33 @@ export async function executeRecord(
return finalizeRecord(d, recordId);
}
/** 记录暂停的原因前缀:库根离线 / 建目标目录失败(整理巡检据此判断要不要计入重投上限) */
export const ROOT_OFFLINE = "library_root_offline";
export const TARGET_DIR_UNAVAILABLE = "target_dir_unavailable";
export const isHoldReason = (reason: string | null | undefined): boolean =>
!!reason && (reason.startsWith(`${ROOT_OFFLINE}:`) || reason.startsWith(`${TARGET_DIR_UNAVAILABLE}:`));
/** 暂停执行:记录保持 running、写原因并开待处理;返回 running 让任务按"等待"收尾,不耗尽重试 */
async function holdRecord(
d: AoDeps,
rec: { id: string; file_succeeded: number; file_failed: number },
reason: string,
decision: Parameters<typeof upsertDecision>[2],
): Promise<ExecuteResult> {
await d.ctx.db.transaction().execute(async (trx) => {
await trx
.updateTable("organization_record")
.set({ reason })
.where("id", "=", rec.id)
.where("status", "in", ["running", "queued"])
.execute();
await upsertDecision(d, trx, decision);
await emitRecord(d, trx, rec.id);
});
d.ctx.logger.warn({ recordId: rec.id, reason }, "整理暂停,等待巡检重试");
return { recordId: rec.id, status: "running", succeeded: rec.file_succeeded, failed: rec.file_failed };
}
/**
* replace 的旧文件(copy 已同卷移到 <根>/.umt-trash/replaced-*)改放成文件管理回收站的布局
* <根real>/.umt-trash/<yyyymmdd>/<uuid>/<原相对路径> 并写 trash_item(与 file-manager/trash.ts moveToTrash 同构,
@@ -4,7 +4,7 @@ import { statSafe } from "@umt/adapters";
import type { DownloadTaskRow, OrganizationRecordRow } from "@umt/db";
import { sql } from "@umt/db";
import { findOtherVersions } from "@umt/domain";
import { organizationStatusFromFiles } from "@umt/mapping";
import { isUnresolvedExclusion, organizationStatusFromFiles } from "@umt/mapping";
import { errMessage, jsonb } from "../util.ts";
import { type AoDeps, type Exec, loadInstances, upsertDecision } from "./common.ts";
import { applySlotEvent } from "./downloads.ts";
@@ -274,7 +274,9 @@ export async function finalizeRecord(d: AoDeps, recordId: string): Promise<Execu
}
const statuses = files.map((f) => f.status);
const status = organizationStatusFromFiles(statuses);
// 季集待确认 / 没有季目录 / 命名失败而被排除的视频留在原处:记录按 partial,Move 不移除下载任务
const unresolved = files.filter((f) => f.status === "skipped" && isUnresolvedExclusion(f.skip_reason));
const status = organizationStatusFromFiles(statuses, { unresolved: unresolved.length });
const counts = {
file_succeeded: statuses.filter((s) => s === "succeeded").length,
file_failed: statuses.filter((s) => s === "failed").length,
@@ -290,10 +292,31 @@ export async function finalizeRecord(d: AoDeps, recordId: string): Promise<Execu
finished_at: now,
...(status === "failed" && !rec.reason
? { reason: rec.action === "in_place" ? "in_place_failed" : "transfer_failed" }
: {}),
: status === "partial" && unresolved.length > 0 && counts.file_failed === 0 && !rec.reason
? { reason: "excluded_unresolved" }
: {}),
})
.where("id", "=", recordId)
.execute();
if (unresolved.length > 0) {
const shown = unresolved
.slice(0, 5)
.map((f) => `${f.source_path.split("/").pop()}(${f.skip_reason?.replace(/^excluded:/, "")})`);
await upsertDecision(d, trx, {
kind: "other",
title: `有 ${unresolved.length} 个视频没有整理`,
summary: `季集、季目录或命名没能确定,文件留在原处${
rec.action === "move" ? ",下载任务保留" : ""
}:${shown.join("、")}${unresolved.length > shown.length ? " 等" : ""}`,
subjectType: "organization_record",
subjectId: recordId,
dedupeKey: `organize_excluded:${recordId}`,
evidence: {
downloadTaskId: rec.download_task_id,
files: unresolved.map((f) => ({ path: f.source_path, reason: f.skip_reason })),
},
});
}
// 下载任务 / 计划 / Slot
const move = rec.action === "move" && status === "succeeded";
@@ -449,30 +472,7 @@ export async function finalizeRecord(d: AoDeps, recordId: string): Promise<Execu
});
}
// Move 完成:移除 qB 任务记录,结束下载侧管理(不删数据)
if (result.move && result.dt) {
const topo = (
await loadInstances(db, {
enabledOnly: false,
ids: [result.dt.instance_id],
})
)[0];
if (topo) {
try {
const client = await d.ports.downloader(topo.inst);
await client.delete([result.dt.torrent_hash], { deleteFiles: false });
await db
.updateTable("download_task")
.set({ removed_at: d.ports.now(), managed: false })
.where("id", "=", result.dt.id)
.execute();
} catch (err) {
d.ctx.logger.warn(
{ downloadTaskId: result.dt.id, err: { message: errMessage(err) } },
"Move 后移除 qB 任务失败,下次同步重试",
);
}
}
}
if (result.move && result.dt) await removeMovedDownload(d, result.dt);
return {
recordId: result.recordId,
status: result.status,
@@ -481,6 +481,34 @@ export async function finalizeRecord(d: AoDeps, recordId: string): Promise<Execu
};
}
/**
* Move 成功后移除 qB 任务(不删数据)并结束纳管。失败只告警:executeRecord 重放已结束记录、
* 整理巡检(recoverStalledRecords)会对"move 成功但下载任务未移除"的记录再试。
*/
export async function removeMovedDownload(
d: AoDeps,
dt: Pick<DownloadTaskRow, "id" | "instance_id" | "torrent_hash">,
): Promise<boolean> {
const topo = (await loadInstances(d.ctx.db, { enabledOnly: false, ids: [dt.instance_id] }))[0];
if (!topo) return false;
try {
const client = await d.ports.downloader(topo.inst);
await client.delete([dt.torrent_hash], { deleteFiles: false });
await d.ctx.db
.updateTable("download_task")
.set({ removed_at: d.ports.now(), managed: false })
.where("id", "=", dt.id)
.execute();
return true;
} catch (err) {
d.ctx.logger.warn(
{ downloadTaskId: dt.id, err: { message: errMessage(err) } },
"Move 后移除 qB 任务失败,整理巡检会重试",
);
return false;
}
}
/** 关闭下载任务的"整理冲突 / 做种义务阻止移动 / 没识别出作品"事项(resolution.key=manual_organize,记整理记录 id) */
export async function closeOrganizeDecisions(d: AoDeps, trx: Exec, downloadTaskId: string, rec: OrganizationRecordRow) {
const by = rec.actor_id ? `${rec.actor_type}:${rec.actor_id}` : rec.actor_type;
@@ -1,6 +1,10 @@
import { reasonSentence } from "@umt/contracts";
import type { OrganizationRecordRow } from "@umt/db";
import type { Exec } from "./common.ts";
import { type OrganizationRecordRow, sql } from "@umt/db";
import { errMessage, jsonb } from "../util.ts";
import { type AoDeps, type Exec, enqueueSweep, timeSlot, upsertDecision } from "./common.ts";
import { SWEEP_SLOT_MS } from "./downloads.ts";
import { isHoldReason, ROOT_OFFLINE } from "./organize-exec.ts";
import { finalizeRecord, removeMovedDownload } from "./organize-finalize.ts";
// 入库:冻结整理计划(organization_record + 文件明细)→ 逐文件可恢复执行(操作账本 fencing)→
// 资产/位置/路径绑定同事务登记 → Slot 推进 → Move 后移除 qB 任务记录(不删数据)。
@@ -190,3 +194,160 @@ export async function loadRecordsPublic(exec: Exec, rows: OrganizationRecordRow[
};
});
}
// ── 整理巡检:卡在 running 的记录重新入队;Move 成功但下载任务没移除的补移除 ──
/** 记录 running 且没有活跃任务运行超过这么久,视为卡住 */
export const STALLED_RECORD_MS = 30 * 60_000;
/** 库根离线暂停的记录:每隔这么久重投一次查库根(不计入重投上限) */
export const ROOT_WAIT_RECHECK_MS = 5 * 60_000;
/** 巡检重投上限;用完仍卡住 → 未完成的文件记失败、记录收尾、开待处理 */
export const MAX_RECORD_REQUEUES = 3;
/** Move 后移除下载任务的补做窗口:收尾后稍等(避开收尾自己的移除),太久的不再管 */
const MOVE_REMOVE_GRACE_MS = 2 * 60_000;
const MOVE_REMOVE_WINDOW_MS = 7 * 24 * 3600_000;
/** 引擎里排队 / 执行中的 verifyAndImport 涉及的记录 id 与下载任务 id */
async function activeImportSubjects(d: AoDeps): Promise<Set<string>> {
const ids = new Set<string>();
for (const status of ["pending", "running"] as const) {
for (const run of await d.ctx.engine.list({ taskName: "verifyAndImport", status, limit: 500 })) {
const input = (run.input ?? {}) as { recordId?: unknown; downloadTaskId?: unknown };
if (typeof input.recordId === "string") ids.add(input.recordId);
if (typeof input.downloadTaskId === "string") ids.add(input.downloadTaskId);
}
}
return ids;
}
/**
* 巡检整理记录(挂在"补提交与补入库"定时任务上):
* - running / queued、没有活跃的 verifyAndImport 运行、超过 STALLED_RECORD_MS 没动静(任务重试耗尽、
* planImport 提交后入队前进程退出)→ 以 record:<id> 重新入队(executeRecord 可重入:已完成文件跳过,
* 账本未出结果的先对账);重投超过上限 → 未完成文件记失败、记录收尾并开待处理;
* - 库根离线暂停的记录每 ROOT_WAIT_RECHECK_MS 重投一次,不计上限(库根恢复即继续);
* - Move 成功但下载任务未移除(移除 qB 任务失败)→ 再试移除。
*/
export async function recoverStalledRecords(d: AoDeps): Promise<{ requeued: number; failed: number; removed: number }> {
const db = d.ctx.db;
const now = d.ports.now();
const out = { requeued: 0, failed: 0, removed: 0 };
const rows = await db
.selectFrom("organization_record")
.select(["id", "download_task_id", "reason", "requeue_count", "requeued_at", "started_at", "created_at"])
.where("status", "in", ["running", "queued"])
.where(
sql<Date>`coalesce(requeued_at, started_at, created_at)`,
"<",
new Date(now.getTime() - Math.min(STALLED_RECORD_MS, ROOT_WAIT_RECHECK_MS)),
)
.orderBy("created_at")
.limit(20)
.execute();
const due = rows.filter((r) => {
const since = (r.requeued_at ?? r.started_at ?? r.created_at).getTime();
const wait = r.reason?.startsWith(`${ROOT_OFFLINE}:`) ? ROOT_WAIT_RECHECK_MS : STALLED_RECORD_MS;
return now.getTime() - since >= wait;
});
const active = due.length ? await activeImportSubjects(d) : new Set<string>();
const slot = timeSlot(now, SWEEP_SLOT_MS);
for (const r of due) {
if (active.has(r.id) || (r.download_task_id && active.has(r.download_task_id))) continue;
const rootWait = r.reason?.startsWith(`${ROOT_OFFLINE}:`) ?? false;
try {
if (!rootWait && r.requeue_count >= MAX_RECORD_REQUEUES) {
await failStalledRecord(d, r.id, r.requeue_count);
out.failed++;
continue;
}
const q = await enqueueSweep(
d,
{
kind: "verifyAndImport",
idempotencyKey: `record:${r.id}`,
input: { recordId: r.id },
subjectType: "organization_record",
subjectId: r.id,
},
slot,
);
if (!q) continue;
await db
.updateTable("organization_record")
.set({ requeued_at: now, ...(rootWait ? {} : { requeue_count: r.requeue_count + 1 }) })
.where("id", "=", r.id)
.execute();
out.requeued++;
} catch (err) {
d.ctx.logger.warn({ recordId: r.id, err: { message: errMessage(err) } }, "整理记录巡检处理失败");
}
}
// Move 成功、下载任务仍纳管未移除:补移除 qB 任务(同一下载任务只看最新一条记录)
const moved = await db
.selectFrom("organization_record as r")
.innerJoin("download_task as dt", "dt.id", "r.download_task_id")
.select(["dt.id", "dt.instance_id", "dt.torrent_hash"])
.where("r.action", "=", "move")
.where("r.status", "=", "succeeded")
.where("r.finished_at", "<", new Date(now.getTime() - MOVE_REMOVE_GRACE_MS))
.where("r.finished_at", ">", new Date(now.getTime() - MOVE_REMOVE_WINDOW_MS))
.where("dt.removed_at", "is", null)
.where("dt.managed", "=", true)
.where((eb) =>
eb.not(
eb.exists(
eb
.selectFrom("organization_record as r2")
.select("r2.id")
.whereRef("r2.download_task_id", "=", "r.download_task_id")
.whereRef("r2.created_at", ">", "r.created_at"),
),
),
)
.limit(20)
.execute();
for (const dt of moved) if (await removeMovedDownload(d, dt)) out.removed++;
return out;
}
/** 重投用完仍卡住:未完成文件记失败,按已完成的文件收尾(登记资产、推进下载任务 / Slot),开待处理 */
async function failStalledRecord(d: AoDeps, recordId: string, requeues: number): Promise<void> {
const db = d.ctx.db;
await db.transaction().execute(async (trx) => {
const rec = await trx
.selectFrom("organization_record")
.select(["id", "status", "reason", "download_task_id"])
.where("id", "=", recordId)
.forUpdate()
.executeTakeFirst();
if (!rec || (rec.status !== "running" && rec.status !== "queued")) return;
await trx
.updateTable("organization_file_result")
.set({
status: "failed",
error: jsonb({ code: "STALLED", message: "整理多次重试仍未完成" }),
finished_at: d.ports.now(),
})
.where("organization_record_id", "=", recordId)
.where("status", "in", ["queued", "running"])
.execute();
await trx
.updateTable("organization_record")
.set({ reason: "organize_stalled" })
.where("id", "=", recordId)
.execute();
await upsertDecision(d, trx, {
kind: "other",
title: "整理一直没完成,已停止",
summary: `重新排队 ${requeues} 次仍没有完成${
isHoldReason(rec.reason) ? `(${rec.reason})` : ""
};已完成的文件照常登记,其余请检查后重新整理`,
subjectType: "organization_record",
subjectId: recordId,
dedupeKey: `organize_stalled:${recordId}`,
evidence: { requeues, previousReason: rec.reason, downloadTaskId: rec.download_task_id },
});
});
await finalizeRecord(d, recordId);
}