rootandClaude Opus 5.5
da33cad81d
io worker 进程内按任务声明的 lane 分文件 / 网络 / 维护三条道,各自并发(缺省 2/4/1)、 各自排队;维护道在文件道或网络道有排队/运行时让路(连续 10 分钟放行一个),调度器内联的 清理/回填/保留期裁剪也走维护道。并发在 设置 → 自动化 → 高级 热生效,env 覆盖;旧 WORKER_CONCURRENCY 映射到网络道。运行状态页按道显示运行中/并发/排队。归属表写进 docs/runbook/dev.md,ci 与注册表逐项核对。 Co-Authored-By: Claude Opus 5.5 <[email protected]>
102 lines
4.1 KiB
TypeScript
102 lines
4.1 KiB
TypeScript
import type { z } from "zod";
|
|
import { UnknownTaskError } from "./errors.ts";
|
|
import type { RetryPolicy, TaskContext } from "./types.ts";
|
|
|
|
/**
|
|
* 执行道(WK-1):io worker 进程内按道分池,各道独立并发。
|
|
* file — 整理、硬链接/移动、ffprobe、校验等占磁盘的任务
|
|
* network — 站点/RSS/TMDB/下载器/搜索等外部请求
|
|
* maintenance — 清理、回填、对账等可以等空闲再跑的任务(文件道/网络道有活时让路)
|
|
* agent — Agent 会话(worker:agent 进程,不在 io 进程的三条道里)
|
|
*/
|
|
export const TASK_LANES = ["file", "network", "maintenance", "agent"] as const;
|
|
export type TaskLane = (typeof TASK_LANES)[number];
|
|
/** 没声明 lane 的任务运行时归入网络道(启动时打 warn;ci 的归属表测试会直接失败) */
|
|
export const DEFAULT_TASK_LANE: TaskLane = "network";
|
|
|
|
export interface TaskDefinition<S extends z.ZodType = z.ZodType, O = unknown> {
|
|
name: string;
|
|
/** 正整数;不兼容改动 +1,并在 worker 上同时注册新旧版本直到旧 run 清空 */
|
|
version: number;
|
|
/** 执行道;新任务必须显式声明(见 TASK_LANES) */
|
|
lane?: TaskLane;
|
|
inputSchema: S;
|
|
/** 步骤默认重试策略(步骤可单独覆盖)。外部网络类建议 5 次 1s→60s,文件类 3 次 */
|
|
retryPolicy?: Partial<RetryPolicy>;
|
|
handler: (ctx: TaskContext<z.output<S>>) => Promise<O>;
|
|
}
|
|
|
|
export function defineTask<S extends z.ZodType, O>(def: TaskDefinition<S, O>): TaskDefinition<S, O> {
|
|
if (!/^[A-Za-z][A-Za-z0-9_.-]*$/.test(def.name)) throw new Error(`invalid task name: ${def.name}`);
|
|
if (!Number.isInteger(def.version) || def.version < 1) throw new Error(`invalid task version: ${def.version}`);
|
|
if (def.lane !== undefined && !(TASK_LANES as readonly string[]).includes(def.lane))
|
|
throw new Error(`invalid task lane: ${def.lane}`);
|
|
return def;
|
|
}
|
|
|
|
/** 任务所属执行道;未声明时回退 DEFAULT_TASK_LANE */
|
|
export const laneOf = (def: Pick<TaskDefinition, "lane">): TaskLane => def.lane ?? DEFAULT_TASK_LANE;
|
|
|
|
export const engineVersion = (version: number) => `v${version}`;
|
|
export const parseEngineVersion = (v: string | null): number | null => {
|
|
const m = v ? /^v(\d+)$/.exec(v) : null;
|
|
return m ? Number(m[1]) : null;
|
|
};
|
|
|
|
export class TaskRegistry {
|
|
private readonly defs = new Map<string, Map<number, TaskDefinition>>();
|
|
|
|
// biome-ignore lint/suspicious/noExplicitAny: 注册表按名存放异构的任务定义
|
|
register(...defs: TaskDefinition<any, any>[]): this {
|
|
for (const def of defs) {
|
|
let byVersion = this.defs.get(def.name);
|
|
if (!byVersion) {
|
|
byVersion = new Map();
|
|
this.defs.set(def.name, byVersion);
|
|
}
|
|
if (byVersion.has(def.version)) throw new Error(`task ${def.name}@${def.version} already registered`);
|
|
byVersion.set(def.version, def as TaskDefinition);
|
|
}
|
|
return this;
|
|
}
|
|
|
|
get(name: string, version: number): TaskDefinition {
|
|
const def = this.defs.get(name)?.get(version);
|
|
if (!def) throw new UnknownTaskError(name, version);
|
|
return def;
|
|
}
|
|
|
|
/** 新 run 一律用最高版本 */
|
|
latest(name: string): TaskDefinition {
|
|
const byVersion = this.defs.get(name);
|
|
if (!byVersion || byVersion.size === 0) throw new UnknownTaskError(name);
|
|
return byVersion.get(Math.max(...byVersion.keys()))!;
|
|
}
|
|
|
|
/** 是否已注册该任务(任一版本) */
|
|
has(name: string): boolean {
|
|
return (this.defs.get(name)?.size ?? 0) > 0;
|
|
}
|
|
|
|
names(): string[] {
|
|
return [...this.defs.keys()].sort();
|
|
}
|
|
|
|
all(): TaskDefinition[] {
|
|
return [...this.defs.values()].flatMap((m) => [...m.values()]);
|
|
}
|
|
|
|
/** 按 StartWorkerOptions.versions 选择:"name@2" / "@1" / "name"(该任务全部版本) */
|
|
select(selectors?: string[]): TaskDefinition[] {
|
|
if (!selectors || selectors.length === 0) return this.all();
|
|
const picked = this.all().filter((d) =>
|
|
selectors.some((s) => {
|
|
const [name, ver] = s.split("@");
|
|
return (!name || name === d.name) && (ver === undefined || ver === "" || Number(ver) === d.version);
|
|
}),
|
|
);
|
|
if (picked.length === 0) throw new Error(`no task definitions match ${selectors.join(",")}`);
|
|
return picked;
|
|
}
|
|
}
|