feat(worker): IO Worker 分道(WK-1)

io worker 进程内按任务声明的 lane 分文件 / 网络 / 维护三条道,各自并发(缺省 2/4/1)、
各自排队;维护道在文件道或网络道有排队/运行时让路(连续 10 分钟放行一个),调度器内联的
清理/回填/保留期裁剪也走维护道。并发在 设置 → 自动化 → 高级 热生效,env 覆盖;旧
WORKER_CONCURRENCY 映射到网络道。运行状态页按道显示运行中/并发/排队。归属表写进
docs/runbook/dev.md,ci 与注册表逐项核对。

Co-Authored-By: Claude Opus 5.5 <[email protected]>

packages/tasks/src/definitions.ts

@@ -33,6 +33,7 @@ const DB_RETRY = { maximumAttempts: 3, initialInterval: "1s", backoffCoefficient
 export const ingestFeed = defineTask({
   name: "ingestFeed",
   version: 1,
+  lane: "network",
   inputSchema: z.object({ feedSourceId: id }),
   retryPolicy: NETWORK_RETRY,
   handler: stub("ingestFeed"),
@@ -41,6 +42,7 @@ export const ingestFeed = defineTask({
 export const resolveObservation = defineTask({
   name: "resolveObservation",
   version: 1,
+  lane: "network",
   inputSchema: z.object({ observationId: id }),
   retryPolicy: NETWORK_RETRY,
   handler: stub("resolveObservation"),
@@ -49,6 +51,7 @@ export const resolveObservation = defineTask({
 export const evaluateAcquisition = defineTask({
   name: "evaluateAcquisition",
   version: 1,
+  lane: "network",
   inputSchema: z.object({
     subscriptionId: id.optional(),
     acquisitionSlotId: id.optional(),
@@ -61,6 +64,7 @@ export const evaluateAcquisition = defineTask({
 export const submitDownload = defineTask({
   name: "submitDownload",
   version: 1,
+  lane: "network",
   inputSchema: z.object({ acquisitionPlanId: id }),
   retryPolicy: NETWORK_RETRY,
   handler: stub("submitDownload"),
@@ -69,6 +73,7 @@ export const submitDownload = defineTask({
 export const verifyAndImport = defineTask({
   name: "verifyAndImport",
   version: 1,
+  lane: "file",
   inputSchema: z.object({ downloadTaskId: id }),
   retryPolicy: FILE_RETRY,
   handler: stub("verifyAndImport"),
@@ -77,6 +82,7 @@ export const verifyAndImport = defineTask({
 export const refreshMetadata = defineTask({
   name: "refreshMetadata",
   version: 1,
+  lane: "network",
   inputSchema: z.object({ mediaEntityId: id }),
   retryPolicy: NETWORK_RETRY,
   handler: stub("refreshMetadata"),
@@ -85,6 +91,7 @@ export const refreshMetadata = defineTask({
 export const reconcileLibrary = defineTask({
   name: "reconcileLibrary",
   version: 1,
+  lane: "maintenance",
   inputSchema: z.object({ libraryRootId: id.optional(), reconcileRunId: id.optional() }),
   retryPolicy: FILE_RETRY,
   handler: stub("reconcileLibrary"),
@@ -93,6 +100,7 @@ export const reconcileLibrary = defineTask({
 export const cleanupDownloadCache = defineTask({
   name: "cleanupDownloadCache",
   version: 1,
+  lane: "file",
   inputSchema: z.object({ downloaderInstanceId: id.optional() }),
   retryPolicy: FILE_RETRY,
   handler: stub("cleanupDownloadCache"),
@@ -101,6 +109,7 @@ export const cleanupDownloadCache = defineTask({
 export const resolveDecision = defineTask({
   name: "resolveDecision",
   version: 1,
+  lane: "agent",
   inputSchema: z.object({ decisionId: id }),
   retryPolicy: DB_RETRY,
   handler: stub("resolveDecision"),

packages/tasks/src/index.ts

@@ -48,5 +48,13 @@ export {
   type OutboxRoute,
   type OutboxRouteFn,
 } from "./outbox-relay.ts";
-export { defineTask, type TaskDefinition, TaskRegistry } from "./registry.ts";
+export {
+  DEFAULT_TASK_LANE,
+  defineTask,
+  laneOf,
+  TASK_LANES,
+  type TaskDefinition,
+  type TaskLane,
+  TaskRegistry,
+} from "./registry.ts";
 export type * from "./types.ts";

packages/tasks/src/openworkflow-engine.ts

@@ -9,6 +9,7 @@ import type {
   EnqueueOptions,
   JsonValue,
   ListFilter,
+  QueueCounts,
   StartWorkerOptions,
   TaskContext,
   TaskEngine,
@@ -16,6 +17,7 @@ import type {
   TaskRunView,
   TaskStepApi,
   WaitForFactOptions,
+  WorkerHandle,
 } from "./types.ts";
 
 // biome-ignore lint/suspicious/noExplicitAny: 本包只用原生 SQL 访问自有表
@@ -59,19 +61,32 @@ export async function migrateEngineSchema(databaseUrl: string, schema = ENGINE_S
   await backend.stop();
 }
 
+interface ClaimWrapOptions {
+  /** 只认领这些 "名称@引擎版本";null = 不按任务过滤 */
+  keys: string[] | null;
+  /** 本 worker 正在执行的 run 数 */
+  active: () => number;
+  maxActive?: () => number;
+  gate?: () => boolean | Promise<boolean>;
+}
+
 /**
- * 包一层只认领指定 (任务名, 版本) 的 backend:与 BackendPostgres.claimWorkflowRun 同一条 SQL(截止过期处理 +
- * SKIP LOCKED 认领),candidate 多一个 workflow_name/version 过滤。其余方法经原型链委托给原 backend。
+ * 包一层 backend 的认领:
+ * - keys:与 BackendPostgres.claimWorkflowRun 同一条 SQL(截止过期处理 + SKIP LOCKED 认领),candidate 多一个
+ *   workflow_name/version 过滤(OpenWorkflow 0.10.1 认领不看任务名,认领到未注册的 run 会以"未注册"失败并退避);
+ * - maxActive:执行中 + 认领中 ≥ 上限就不认领(同一轮 tick 的多个槽并发认领,用 reserved 计数挡住超额);
+ * - gate:返回 false 不认领。
+ * 其余方法经原型链委托给原 backend。
  */
-function claimFilteredBackend(backend: BackendPostgres, defs: readonly TaskDefinition[]): BackendPostgres {
+function claimWrappedBackend(backend: BackendPostgres, o: ClaimWrapOptions): BackendPostgres {
   // biome-ignore lint/suspicious/noExplicitAny: 访问 BackendPostgres 的私有字段(pg / namespaceId / workflowRunsTable)
   const b = backend as any;
-  const keys = [...new Set(defs.map((d) => `${d.name}@${engineVersion(d.version)}`))];
   const filtered = Object.create(backend) as BackendPostgres;
-  Object.defineProperty(filtered, "claimWorkflowRun", {
-    value: async (params: { workerId: string; leaseDurationMs: number }) => {
-      const table = b.workflowRunsTable();
-      const [claimed] = await b.pg`
+  let reserved = 0;
+  const claim = async (params: { workerId: string; leaseDurationMs: number }) => {
+    if (!o.keys) return backend.claimWorkflowRun(params);
+    const table = b.workflowRunsTable();
+    const [claimed] = await b.pg`
       WITH expired AS (
         UPDATE ${table}
         SET "status" = 'failed', "error" = ${b.pg.json({ message: "Workflow run deadline exceeded" })},
@@ -89,7 +104,7 @@ function claimFilteredBackend(backend: BackendPostgres, defs: readonly TaskDefin
           AND "status" IN ('pending', 'running', 'sleeping')
           AND "available_at" <= NOW()
           AND ("deadline_at" IS NULL OR "deadline_at" > NOW())
-          AND ("workflow_name" || '@' || COALESCE("version", '')) = ANY(${keys}::text[])
+          AND ("workflow_name" || '@' || COALESCE("version", '')) = ANY(${o.keys}::text[])
         ORDER BY CASE WHEN "status" = 'pending' THEN 0 ELSE 1 END, "available_at", "created_at"
         LIMIT 1
         FOR UPDATE SKIP LOCKED
@@ -101,7 +116,20 @@ function claimFilteredBackend(backend: BackendPostgres, defs: readonly TaskDefin
       FROM candidate
       WHERE wr."id" = candidate."id" AND wr."namespace_id" = ${b.namespaceId}
       RETURNING wr.*;`;
-      return claimed ?? null;
+    return claimed ?? null;
+  };
+  Object.defineProperty(filtered, "claimWorkflowRun", {
+    value: async (params: { workerId: string; leaseDurationMs: number }) => {
+      // 同步段先占位:同一轮 tick 的各槽依次进到这里,reserved 保证不超过动态上限
+      if (o.maxActive && o.active() + reserved >= Math.max(0, o.maxActive())) return null;
+      reserved += 1;
+      try {
+        if (o.gate && !(await o.gate())) return null;
+        return await claim(params);
+      } finally {
+        // 认领到的 run 在本函数返回后同步加入 worker 的执行集合,下一轮 tick 前已计入 active
+        reserved -= 1;
+      }
     },
   });
   return filtered;
@@ -189,16 +217,63 @@ export class OpenWorkflowEngine implements TaskEngine {
     return res.data.map((r) => toView(r, metas.get(r.id)));
   }
 
-  async startWorker(opts: StartWorkerOptions = {}) {
+  async startWorker(opts: StartWorkerOptions = {}): Promise<WorkerHandle> {
     const defs = this.opts.registry.select(opts.versions);
     const workflows = defs.map((d) => this.toWorkflow(d)) as Workflow<unknown, unknown, unknown>[];
     // 指定了 versions 的 worker 只认领这些任务:OpenWorkflow 0.10.1 认领不看任务名,认领到未注册的 run 会
-    // 以"未注册"失败并指数退避(5s→5m),分组 worker(io / agent)会互相拖慢对方的任务。
-    const backend = opts.versions?.length ? claimFilteredBackend(this.backend, defs) : this.backend;
-    const worker = new Worker({ backend, workflows, concurrency: opts.concurrency ?? 4 });
+    // 以"未注册"失败并指数退避(5s→5m),分组 worker(各道 / agent)会互相拖慢对方的任务。
+    const keys = opts.versions?.length ? [...new Set(defs.map((d) => `${d.name}@${engineVersion(d.version)}`))] : null;
+    let worker: Worker | null = null;
+    // biome-ignore lint/suspicious/noExplicitAny: OpenWorkflow Worker 的执行集合是私有字段,只读取其大小
+    const active = () => ((worker as any)?.activeExecutions?.size as number | undefined) ?? 0;
+    const backend =
+      keys || opts.maxActive || opts.claimGate
+        ? claimWrappedBackend(this.backend, {
+            keys,
+            active,
+            ...(opts.maxActive ? { maxActive: opts.maxActive } : {}),
+            ...(opts.claimGate ? { gate: opts.claimGate } : {}),
+          })
+        : this.backend;
+    const concurrency = opts.concurrency ?? 4;
+    worker = new Worker({ backend, workflows, concurrency });
     await worker.start();
     this.workers.push(worker);
     this.log.info("worker started", { tasks: defs.map((d) => `${d.name}@${d.version}`) });
+    return { concurrency, active };
+  }
+
+  /**
+   * 按任务名分组统计排队 / 运行数(全库):
+   * queued = 已到 available_at 可被认领(pending、等待后到点、租约过期待接手);
+   * running = worker 持有未过期租约。等信号 / sleep 未到点的不计。
+   */
+  async queueCounts(groups: readonly (readonly string[])[]): Promise<QueueCounts[]> {
+    // biome-ignore lint/suspicious/noExplicitAny: 同 claimWrappedBackend
+    const b = this.backend as any;
+    const table = b.workflowRunsTable();
+    const all = [...new Set(groups.flat())];
+    if (all.length === 0) return groups.map(() => ({ queued: 0, running: 0 }));
+    const rows: { name: string; queued: string | number; running: string | number }[] = await b.pg`
+      SELECT "workflow_name" AS name,
+        COUNT(*) FILTER (WHERE "available_at" <= NOW()) AS queued,
+        COUNT(*) FILTER (WHERE "status" = 'running' AND "worker_id" IS NOT NULL AND "available_at" > NOW()) AS running
+      FROM ${table}
+      WHERE "namespace_id" = ${b.namespaceId}
+        AND "status" IN ('pending', 'running', 'sleeping')
+        AND ("deadline_at" IS NULL OR "deadline_at" > NOW())
+        AND "workflow_name" = ANY(${all}::text[])
+      GROUP BY "workflow_name"`;
+    const by = new Map(rows.map((r) => [r.name, { queued: Number(r.queued), running: Number(r.running) }]));
+    return groups.map((names) =>
+      names.reduce(
+        (acc, n) => {
+          const c = by.get(n);
+          return c ? { queued: acc.queued + c.queued, running: acc.running + c.running } : acc;
+        },
+        { queued: 0, running: 0 },
+      ),
+    );
   }
 
   /** 等进行中的 handler 执行完再断开连接池 */

packages/tasks/src/registry.ts

@@ -2,10 +2,24 @@ 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>;
@@ -15,9 +29,14 @@ export interface TaskDefinition<S extends z.ZodType = z.ZodType, O = unknown> {
 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;

packages/tasks/src/types.ts

@@ -56,7 +56,30 @@ export interface StartWorkerOptions {
    * 省略则执行注册表里的全部版本。升级期间新旧版本要同时列出。
    */
   versions?: string[];
+  /** 并发槽上限(引擎 worker 的槽位数,启动后不变);缺省 4 */
   concurrency?: number;
+  /**
+   * 动态并发:每次认领前读取,本 worker 执行中 + 认领中的 run 数达到它就不再认领(≤ concurrency 才有意义)。
+   * 用于设置页改并发后热生效(槽位按上限开,实际占用按它收)。
+   */
+  maxActive?: () => number;
+  /** 认领闸门:每次认领前调用,返回 false 本次不认领(维护道"空闲时跑"用) */
+  claimGate?: () => boolean | Promise<boolean>;
+}
+
+/** startWorker 返回的句柄:本 worker 的槽位与正在执行的 run 数 */
+export interface WorkerHandle {
+  readonly concurrency: number;
+  /** 本进程内该 worker 正在执行的 run 数 */
+  active(): number;
+}
+
+/** 一组任务在引擎里的排队 / 运行数(全库,跨进程) */
+export interface QueueCounts {
+  /** 已到可执行时间、等待认领的 run(含租约过期待接手的) */
+  queued: number;
+  /** 被某个 worker 持有租约、正在执行的 run */
+  running: number;
 }
 
 export interface TaskEngine {
@@ -70,7 +93,10 @@ export interface TaskEngine {
   cancel(runId: string, reason?: string): Promise<boolean>;
   get(runId: string): Promise<TaskRunView | null>;
   list(filter?: ListFilter): Promise<TaskRunView[]>;
-  startWorker(opts?: StartWorkerOptions): Promise<void>;
+  // biome-ignore lint/suspicious/noConfusingVoidType: 测试替身常写成 async startWorker() {}(返回 void)
+  startWorker(opts?: StartWorkerOptions): Promise<WorkerHandle | void>;
+  /** 按任务名分组统计排队 / 运行数(每组一条);不支持的实现可省略 */
+  queueCounts?(groups: readonly (readonly string[])[]): Promise<QueueCounts[]>;
   stop(): Promise<void>;
 }