diff --git a/packages/tasks/README.md b/packages/tasks/README.md index 14e0577e..b261968e 100644 --- a/packages/tasks/README.md +++ b/packages/tasks/README.md @@ -1,3 +1,57 @@ # @umt/tasks -持久任务定义与执行器(OpenWorkflow 或自研 PG 租约表,结论见 ADR-002)与 outbox 投递。负责幂等键、重试、取消传播与重启恢复。 +持久任务引擎封装。依据 ADR-002:OpenWorkflow `0.10.1`(精确锁定)+ postgres.js,引擎表固定在 `openworkflow` schema。业务代码只依赖本包导出的 `TaskEngine` 端口,不直接 import `openworkflow`。 + +## 引擎边界 + +- **引擎只管过程**:推进步骤、定时唤醒、重试、租约。**业务状态不放引擎**,download_task / approval / asset_operation 等都写业务表。 +- **UI 读业务表**。`TaskEngine.get/list` 只是只读投影,给 task.inspect 和管理页看状态、attempts、错误用。不要读 `openworkflow.step_attempts` 来拼时间线:被中断的 attempt 会一直停在 `running`。 +- 引擎用自己的 postgres.js 连接池,不能和业务写入放在同一个事务里入队。入队要走 outbox。 + +## 组成 + +| 导出 | 作用 | +| --- | --- | +| `OpenWorkflowEngine.connect({ databaseUrl, db, registry })` | `TaskEngine` 实现:`enqueue / signal / cancel / get / list / startWorker / stop` | +| `defineTask` / `TaskRegistry` / `createDefaultRegistry()` | 任务定义与注册表;默认注册表里有 §3.1-8 的 9 个任务(目前是空壳) | +| `OutboxRelay` | 用 `FOR UPDATE SKIP LOCKED` 认领 `outbox_event` 里未投递的行,按 type 路由后 enqueue(幂等键 `outbox:`),再标记 `published_at`;多进程并发时不会重复 | +| `withAttemptToken(db, opKey, fn, { commit })` | 操作账本 fencing,表是 `tasks_op_ledger` | +| `migrateTasksTables(db)` | 本包迁移(`migrations/`):`tasks_run_meta`、`tasks_fact`、`tasks_op_ledger`。W2 并入 `packages/db` | + +`TaskContext` 包含 `input`(已经过 zod 校验)、`runId`、`traceId`、`priority`、`attemptToken`(每次执行 handler 都会重新生成)、`abortSignal`、`logger`(自动脱敏)、`step.run / step.sleep / step.waitForFact`、`throwIfCanceled()`。 + +- **取消**:`cancel(runId, reason)` 先写 `tasks_run_meta.cancel_requested_at`,再取消引擎里的 run。handler 执行期间会轮询这个标志(默认 1s)并触发 `abortSignal`;每个 `step.run/sleep` 开始前也会检查一次。进行中的步骤只能靠自己检查 `abortSignal` 来中止。已完成的外部动作不会撤销,补偿要写成显式步骤。 +- **等待外部条件**:`signal(runId, name, payload)` 先把事实写进 `tasks_fact`,再发送引擎信号。`step.waitForFact(name, { check?, timeout })` 每一轮都先查库,查不到才带超时等待信号。所以事实比等待方先到也不会丢。审批、下载完成这类业务事实,先写业务表,再用 `check` 去读业务表。 + +## 新增/填充任务(W2 规则) + +1. 在 `src/definitions.ts` 里用 `defineTask({ name, version, inputSchema, retryPolicy, handler })` 定义任务。输入只放业务 id,handler 自己去读业务表。重试策略要显式写:外部网络类 5 次(1s→60s),文件类 3 次。 +2. **步骤体必须幂等**。worker 被 kill 后,正在执行的步骤会在新 worker 上**从头重跑**(最长要等约 30s 租约到期)。已完成的步骤不会重跑。 +3. **有外部副作用的步骤都走 `withAttemptToken`**,opKey 按语义生成,例如 `submit:`: + - 已经有结果时直接返回; + - `info.resumed === true` 表示上一次 attempt 没留下结果。这时先对账外部状态(qB 按 hash 或管理标签查,文件按目标路径 + 大小/hash 查),再决定是收尾还是重做; + - 业务状态放在 `commit(trx, result)` 里提交,和 fencing 校验在同一个事务。租约丢失的旧 attempt 提交时会抛 `FencingError`。 +4. 等授权、等下载完成,用 `step.waitForFact`,不要直接等信号。同一个 run 里,fact 名不要重复使用。 +5. 长步骤(比如复制大文件)要分块检查 `ctx.abortSignal`,被中止时 `throw ctx.abortSignal.reason`。 +6. 步骤名在同一版本内保持稳定。重放是按步骤名匹配的。 +7. 日志只用 `ctx.logger`。passkey、token 这类字段和 URL 参数会被脱敏,但也不要主动把凭证传给 logger。 + +## 版本升级规则 + +- 不兼容的改动(改步骤名或顺序、改输入结构、改已有步骤的语义)必须把 `version` +1,同时**保留旧版本定义并注册**。worker 用 `startWorker({ versions: ["submitDownload@1", "submitDownload@2"] })` 或者省略参数(表示全部版本)。等 `list({ taskName, status })` 查不到旧版本的 run 了,再删除旧定义。 +- 新 run 一律用最高版本。旧版本 run 如果找不到对应的 worker,会无限退避等待,不会失败。 +- 同一版本里,只允许在末尾追加步骤,或插入幂等步骤。 + +## 已知缺口(OpenWorkflow 0.10.1) + +- 引擎信号不持久,同名信号同一时间只能有一个等待者。已经用 `tasks_fact` + `waitForFact` 补上。 +- 进行中的步骤拿不到引擎的 AbortSignal 和租约状态。已经用 PG 取消标志和操作账本 fencing 补上。 +- 租约固定 30s,不可配置。崩溃后最长约 30s 才会续跑。 +- 引擎幂等键只在 24h 内有效,而且不比较输入。长期去重靠业务表的唯一约束。 +- `priority` 只记录在 `tasks_run_meta` 和 run 输入里,引擎仍按 `available_at` 先后认领,不支持真正的优先级调度。 +- 取消只阻止后续步骤,不撤销已完成的外部动作。在已取消的 run 上,handler 抛出的错误会被引擎拒收,只在 worker 日志里出现。 +- `runMigrations` 默认 true(幂等)。等并入统一迁移后,启动时改成 false。 + +## 测试 + +`pnpm test` 会连接开发库(`DATABASE_URL`,默认 `postgres://umt:umt@localhost:55433/umt`)。每个测试文件使用独立的 `tasks_test_*` schema,测试前后都会清理;库不可达时跳过集成测试。覆盖范围:kill -9 后续跑(子进程 worker)、取消、幂等 enqueue、两个 relay 并发、事实先到达、fencing。 diff --git a/packages/tasks/migrations/0001_tasks_tables.ts b/packages/tasks/migrations/0001_tasks_tables.ts new file mode 100644 index 00000000..e7ca169b --- /dev/null +++ b/packages/tasks/migrations/0001_tasks_tables.ts @@ -0,0 +1,57 @@ +import { type Kysely, sql } from "kysely"; + +// @umt/tasks 自带的三张表(W2 合并进 packages/db 迁移流程): +// - tasks_run_meta:引擎 run 的业务侧元数据(traceId/优先级/取消标志),取消标志是 abortSignal 的来源 +// - tasks_fact:signal 之前先落库的事实;waitForFact 以它为准,信号只用来加速唤醒 +// - tasks_op_ledger:操作账本(fencing),记录 attempt_token,旧 attempt 的提交会被拒绝 + +// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型 +export async function up(db: Kysely): Promise { + await db.schema + .createTable("tasks_run_meta") + .addColumn("run_id", "text", (c) => c.primaryKey()) + .addColumn("task_name", "text", (c) => c.notNull()) + .addColumn("task_version", "integer", (c) => c.notNull()) + .addColumn("idempotency_key", "text", (c) => c.notNull()) + .addColumn("trace_id", "text") + .addColumn("priority", "integer", (c) => c.notNull().defaultTo(0)) + .addColumn("cancel_requested_at", "timestamptz") + .addColumn("cancel_reason", "text") + .addColumn("created_at", "timestamptz", (c) => c.notNull().defaultTo(sql`now()`)) + .execute(); + await db.schema + .createIndex("tasks_run_meta_task_idx") + .on("tasks_run_meta") + .columns(["task_name", "created_at"]) + .execute(); + + await db.schema + .createTable("tasks_fact") + .addColumn("id", "bigserial", (c) => c.primaryKey()) + .addColumn("run_id", "text", (c) => c.notNull()) + .addColumn("name", "text", (c) => c.notNull()) + .addColumn("payload", "jsonb") + .addColumn("created_at", "timestamptz", (c) => c.notNull().defaultTo(sql`now()`)) + .execute(); + await db.schema.createIndex("tasks_fact_run_name_idx").on("tasks_fact").columns(["run_id", "name", "id"]).execute(); + + await db.schema + .createTable("tasks_op_ledger") + .addColumn("op_key", "text", (c) => c.primaryKey()) + .addColumn("run_id", "text") + .addColumn("attempt_token", "uuid", (c) => c.notNull()) + .addColumn("attempts", "integer", (c) => c.notNull().defaultTo(1)) + .addColumn("status", "text", (c) => c.notNull().check(sql`status in ('started','succeeded')`)) + .addColumn("result", "jsonb") + .addColumn("last_error", "text") + .addColumn("started_at", "timestamptz", (c) => c.notNull().defaultTo(sql`now()`)) + .addColumn("completed_at", "timestamptz") + .execute(); +} + +// biome-ignore lint/suspicious/noExplicitAny: 同上 +export async function down(db: Kysely): Promise { + await db.schema.dropTable("tasks_op_ledger").ifExists().execute(); + await db.schema.dropTable("tasks_fact").ifExists().execute(); + await db.schema.dropTable("tasks_run_meta").ifExists().execute(); +} diff --git a/packages/tasks/package.json b/packages/tasks/package.json index d5a5c4ab..5cd50e85 100644 --- a/packages/tasks/package.json +++ b/packages/tasks/package.json @@ -9,5 +9,16 @@ "scripts": { "typecheck": "tsc -p tsconfig.json", "test": "vitest run" + }, + "dependencies": { + "@umt/db": "workspace:*", + "kysely": "^0.29.6", + "openworkflow": "0.10.1", + "pg": "^8.23.0", + "postgres": "^3.4.9", + "zod": "^4.6.5" + }, + "devDependencies": { + "@types/pg": "^8.23.1" } } diff --git a/packages/tasks/src/definitions.ts b/packages/tasks/src/definitions.ts new file mode 100644 index 00000000..94762ede --- /dev/null +++ b/packages/tasks/src/definitions.ts @@ -0,0 +1,135 @@ +import { z } from "zod"; +import { defineTask, TaskRegistry } from "./registry"; + +// 任务书 §3.1-8 的九类任务。W1 只落空壳:输入 schema + 回显 handler,W2 填充处理器。 +// 填充规则见 README「新增/填充任务」:副作用走 withAttemptToken,外部条件用 step.waitForFact, +// 业务状态写业务表,不要放进 handler 返回值。 + +const id = z.string().min(1); + +/** 空壳 handler:记录一条日志并回显输入 */ +function stub(name: string) { + return async (ctx: { input: I; logger: { info(msg: string, f?: Record): void } }) => { + // TODO(W2): 实现真实处理器 + ctx.logger.info(`${name}: stub handler (TODO W2)`); + return { todo: true as const, task: name, input: ctx.input }; + }; +} + +const NETWORK_RETRY = { + maximumAttempts: 5, + initialInterval: "1s", + backoffCoefficient: 2, + maximumInterval: "60s", +} as const; +const FILE_RETRY = { + maximumAttempts: 3, + initialInterval: "2s", + backoffCoefficient: 2, + maximumInterval: "60s", +} as const; +const DB_RETRY = { maximumAttempts: 3, initialInterval: "1s", backoffCoefficient: 2, maximumInterval: "30s" } as const; + +export const ingestFeed = defineTask({ + name: "ingestFeed", + version: 1, + inputSchema: z.object({ feedSourceId: id }), + retryPolicy: NETWORK_RETRY, + handler: stub("ingestFeed"), +}); + +export const resolveObservation = defineTask({ + name: "resolveObservation", + version: 1, + inputSchema: z.object({ observationId: id }), + retryPolicy: NETWORK_RETRY, + handler: stub("resolveObservation"), +}); + +export const evaluateAcquisition = defineTask({ + name: "evaluateAcquisition", + version: 1, + inputSchema: z.object({ + subscriptionId: id.optional(), + acquisitionSlotId: id.optional(), + sourceReleaseId: id.optional(), + }), + retryPolicy: DB_RETRY, + handler: stub("evaluateAcquisition"), +}); + +export const submitDownload = defineTask({ + name: "submitDownload", + version: 1, + inputSchema: z.object({ acquisitionPlanId: id }), + retryPolicy: NETWORK_RETRY, + handler: stub("submitDownload"), +}); + +export const verifyAndImport = defineTask({ + name: "verifyAndImport", + version: 1, + inputSchema: z.object({ downloadTaskId: id }), + retryPolicy: FILE_RETRY, + handler: stub("verifyAndImport"), +}); + +export const refreshMetadata = defineTask({ + name: "refreshMetadata", + version: 1, + inputSchema: z.object({ mediaEntityId: id }), + retryPolicy: NETWORK_RETRY, + handler: stub("refreshMetadata"), +}); + +export const reconcileLibrary = defineTask({ + name: "reconcileLibrary", + version: 1, + inputSchema: z.object({ libraryRootId: id.optional(), reconcileRunId: id.optional() }), + retryPolicy: FILE_RETRY, + handler: stub("reconcileLibrary"), +}); + +export const cleanupDownloadCache = defineTask({ + name: "cleanupDownloadCache", + version: 1, + inputSchema: z.object({ downloaderInstanceId: id.optional() }), + retryPolicy: FILE_RETRY, + handler: stub("cleanupDownloadCache"), +}); + +export const resolveDecision = defineTask({ + name: "resolveDecision", + version: 1, + inputSchema: z.object({ decisionId: id }), + retryPolicy: DB_RETRY, + handler: stub("resolveDecision"), +}); + +export const TASK_NAMES = [ + "ingestFeed", + "resolveObservation", + "evaluateAcquisition", + "submitDownload", + "verifyAndImport", + "refreshMetadata", + "reconcileLibrary", + "cleanupDownloadCache", + "resolveDecision", +] as const; +export type TaskName = (typeof TASK_NAMES)[number]; + +/** v0.1 的默认注册表;W2 把真实 handler 换进来(同名同版本替换,或升版本并保留旧版本) */ +export function createDefaultRegistry(): TaskRegistry { + return new TaskRegistry().register( + ingestFeed, + resolveObservation, + evaluateAcquisition, + submitDownload, + verifyAndImport, + refreshMetadata, + reconcileLibrary, + cleanupDownloadCache, + resolveDecision, + ); +} diff --git a/packages/tasks/src/errors.ts b/packages/tasks/src/errors.ts new file mode 100644 index 00000000..36db108d --- /dev/null +++ b/packages/tasks/src/errors.ts @@ -0,0 +1,21 @@ +export class TaskCanceledError extends Error { + override name = "TaskCanceledError"; + constructor(runId: string, reason?: string | null) { + super(`task run ${runId} canceled${reason ? `: ${reason}` : ""}`); + } +} + +/** 租约已丢:该 opKey 已有更新的 attempt,当前 attempt 的提交被拒绝 */ +export class FencingError extends Error { + override name = "FencingError"; + constructor(opKey: string) { + super(`operation ${opKey}: attempt token superseded, commit rejected`); + } +} + +export class UnknownTaskError extends Error { + override name = "UnknownTaskError"; + constructor(name: string, version?: number) { + super(`task ${name}${version === undefined ? "" : `@${version}`} is not registered`); + } +} diff --git a/packages/tasks/src/index.ts b/packages/tasks/src/index.ts index 582e6f85..8c22e685 100644 --- a/packages/tasks/src/index.ts +++ b/packages/tasks/src/index.ts @@ -1,2 +1,47 @@ -// @umt/tasks 公开入口;W1 起在此导出实现 +// @umt/tasks 公开入口:TaskEngine 端口 + OpenWorkflow 实现、任务注册表、outbox 中继、操作账本 fencing export const PACKAGE_NAME = "@umt/tasks"; + +export { + cleanupDownloadCache, + createDefaultRegistry, + evaluateAcquisition, + ingestFeed, + reconcileLibrary, + refreshMetadata, + resolveDecision, + resolveObservation, + submitDownload, + TASK_NAMES, + type TaskName, + verifyAndImport, +} from "./definitions"; +export { FencingError, TaskCanceledError, UnknownTaskError } from "./errors"; +export { + consoleSink, + createLogger, + type LogLevel, + type LogRecord, + type LogSink, + redact, + silentSink, + type TaskLogger, +} from "./logger"; +export { migrateTasksTables, TASKS_MIGRATIONS_DIR } from "./migrate"; +export { + type AttemptInfo, + assertAttemptToken, + getOperation, + type OpLedgerRow, + type WithAttemptTokenOptions, + withAttemptToken, +} from "./op-ledger"; +export { ENGINE_SCHEMA, OpenWorkflowEngine, type OpenWorkflowEngineOptions, signalName } from "./openworkflow-engine"; +export { + type OutboxEvent, + OutboxRelay, + type OutboxRelayOptions, + type OutboxRoute, + type OutboxRouteFn, +} from "./outbox-relay"; +export { defineTask, type TaskDefinition, TaskRegistry } from "./registry"; +export type * from "./types"; diff --git a/packages/tasks/src/logger.ts b/packages/tasks/src/logger.ts new file mode 100644 index 00000000..8ce6ac18 --- /dev/null +++ b/packages/tasks/src/logger.ts @@ -0,0 +1,64 @@ +export type LogLevel = "debug" | "info" | "warn" | "error"; + +export interface LogRecord { + level: LogLevel; + msg: string; + at: string; + fields: Record; +} + +export type LogSink = (record: LogRecord) => void; + +export interface TaskLogger { + debug(msg: string, fields?: Record): void; + info(msg: string, fields?: Record): void; + warn(msg: string, fields?: Record): void; + error(msg: string, fields?: Record): void; + child(fields: Record): TaskLogger; +} + +const SENSITIVE_KEY = /pass(word|key)?|secret|token|cookie|authorization|api[-_]?key|credential|session/i; +// URL 查询参数里的凭证(passkey、token、apikey、sign 等) +const SENSITIVE_QUERY = /([?&](?:passkey|token|apikey|api_key|key|sign|secret|auth)=)[^&#\s]+/gi; +const REDACTED = "***"; + +/** 深拷贝并脱敏:敏感键名整体替换,字符串里的凭证型 URL 参数替换 */ +export function redact(value: unknown, depth = 0): unknown { + if (depth > 8) return "[depth]"; + if (typeof value === "string") return value.replace(SENSITIVE_QUERY, `$1${REDACTED}`); + if (Array.isArray(value)) return value.map((v) => redact(v, depth + 1)); + if (value instanceof Error) return { name: value.name, message: redact(value.message, depth + 1) }; + if (value && typeof value === "object") { + const out: Record = {}; + for (const [k, v] of Object.entries(value)) out[k] = SENSITIVE_KEY.test(k) ? REDACTED : redact(v, depth + 1); + return out; + } + return value; +} + +export const consoleSink: LogSink = (r) => { + const line = JSON.stringify(r); + if (r.level === "error" || r.level === "warn") console.error(line); + else console.log(line); +}; + +export const silentSink: LogSink = () => {}; + +export function createLogger(sink: LogSink = consoleSink, base: Record = {}): TaskLogger { + const emit = + (level: LogLevel) => + (msg: string, fields: Record = {}) => + sink({ + level, + msg: redact(msg) as string, + at: new Date().toISOString(), + fields: redact({ ...base, ...fields }) as Record, + }); + return { + debug: emit("debug"), + info: emit("info"), + warn: emit("warn"), + error: emit("error"), + child: (fields) => createLogger(sink, { ...base, ...fields }), + }; +} diff --git a/packages/tasks/src/migrate.ts b/packages/tasks/src/migrate.ts new file mode 100644 index 00000000..76b439ed --- /dev/null +++ b/packages/tasks/src/migrate.ts @@ -0,0 +1,25 @@ +import { promises as fs } from "node:fs"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import { type Kysely, sql } from "kysely"; +import { FileMigrationProvider, type MigrationResultSet, Migrator } from "kysely/migration"; + +/** 本包自带迁移(tasks_run_meta / tasks_fact / tasks_op_ledger),W2 合并进 packages/db */ +export const TASKS_MIGRATIONS_DIR = fileURLToPath(new URL("../migrations", import.meta.url)); + +/** 迁移到最新;表建在连接的默认 schema(search_path)里,迁移记录表独立命名避免与 @umt/db 冲突 */ +// biome-ignore lint/suspicious/noExplicitAny: 迁移器不关心具体表类型 +export async function migrateTasksTables(db: Kysely): Promise { + // 显式指定迁移记录表的 schema:Kysely 默认按表名跨 schema 判断存在性,多个 schema 并存时会误判 + const { rows } = await sql<{ schema: string }>`SELECT current_schema() AS schema`.execute(db); + const migrator = new Migrator({ + db, + provider: new FileMigrationProvider({ fs, path, migrationFolder: TASKS_MIGRATIONS_DIR }), + migrationTableName: "tasks_migration", + migrationLockTableName: "tasks_migration_lock", + migrationTableSchema: rows[0]!.schema, + }); + const result = await migrator.migrateToLatest(); + if (result.error) throw result.error; + return result; +} diff --git a/packages/tasks/src/op-ledger.ts b/packages/tasks/src/op-ledger.ts new file mode 100644 index 00000000..492e9b8c --- /dev/null +++ b/packages/tasks/src/op-ledger.ts @@ -0,0 +1,116 @@ +import { randomUUID } from "node:crypto"; +import { type Kysely, sql, type Transaction } from "kysely"; +import { FencingError } from "./errors"; +import type { JsonValue } from "./types"; + +// biome-ignore lint/suspicious/noExplicitAny: 本包只用原生 SQL 访问自有表,不依赖业务库类型 +type AnyDb = Kysely; +// biome-ignore lint/suspicious/noExplicitAny: 同上 +type AnyTrx = Transaction; + +export interface AttemptInfo { + opKey: string; + attemptToken: string; + /** 第几次尝试(1 起) */ + attempt: number; + /** 之前有 attempt 但没有结果:先对账外部状态(qB 按 hash、文件按路径+大小/hash),再决定收尾还是重做 */ + resumed: boolean; +} + +export interface WithAttemptTokenOptions { + runId?: string; + /** 缺省新生成;每次调用都是一个新 attempt,会覆盖旧 token */ + attemptToken?: string; + /** 在同一事务里、fencing 校验通过后提交业务状态;抛错则整体回滚 */ + commit?: (trx: AnyTrx, result: T, info: AttemptInfo) => Promise; +} + +/** + * 操作账本 fencing(ADR-002 提醒 3): + * 1. 已有 succeeded 结果 → 直接返回,不重做; + * 2. 否则写入新 attempt_token(旧 attempt 自此失效)并执行 fn; + * 3. 提交时 `SELECT … FOR UPDATE` 校验 token 仍是自己,不是则抛 FencingError(租约已丢,拒绝提交)。 + */ +export async function withAttemptToken( + db: AnyDb, + opKey: string, + fn: (info: AttemptInfo) => Promise, + opts: WithAttemptTokenOptions = {}, +): Promise { + const token = opts.attemptToken ?? randomUUID(); + const claimed = await sql<{ attempts: number }>` + INSERT INTO tasks_op_ledger (op_key, run_id, attempt_token, attempts, status) + VALUES (${opKey}, ${opts.runId ?? null}, ${token}, 1, 'started') + ON CONFLICT (op_key) DO UPDATE + SET attempt_token = EXCLUDED.attempt_token, attempts = tasks_op_ledger.attempts + 1, + started_at = now(), run_id = COALESCE(EXCLUDED.run_id, tasks_op_ledger.run_id) + WHERE tasks_op_ledger.status <> 'succeeded' + RETURNING attempts`.execute(db); + + const row = claimed.rows[0]; + if (!row) { + const done = await sql<{ result: T }>`SELECT result FROM tasks_op_ledger WHERE op_key = ${opKey}`.execute(db); + return done.rows[0]!.result; + } + + const info: AttemptInfo = { opKey, attemptToken: token, attempt: row.attempts, resumed: row.attempts > 1 }; + let result: T; + try { + result = await fn(info); + } catch (err) { + await sql`UPDATE tasks_op_ledger SET last_error = ${String((err as Error)?.message ?? err)} + WHERE op_key = ${opKey} AND attempt_token = ${token}`.execute(db); + throw err; + } + + await db.transaction().execute(async (trx) => { + await assertAttemptToken(trx, opKey, token); + if (opts.commit) await opts.commit(trx, result, info); + await sql`UPDATE tasks_op_ledger + SET status = 'succeeded', result = ${JSON.stringify(result)}::jsonb, completed_at = now(), last_error = NULL + WHERE op_key = ${opKey}`.execute(trx); + }); + return result; +} + +/** 在业务事务里校验 token 并锁住账本行;用于 withAttemptToken 之外的业务提交 */ +export async function assertAttemptToken(trx: AnyTrx, opKey: string, attemptToken: string): Promise { + const cur = await sql<{ attempt_token: string; status: string }>` + SELECT attempt_token, status FROM tasks_op_ledger WHERE op_key = ${opKey} FOR UPDATE`.execute(trx); + const r = cur.rows[0]; + if (!r || r.attempt_token !== attemptToken || r.status === "succeeded") throw new FencingError(opKey); +} + +export interface OpLedgerRow { + opKey: string; + runId: string | null; + attemptToken: string; + attempts: number; + status: "started" | "succeeded"; + result: unknown; + lastError: string | null; +} + +export async function getOperation(db: AnyDb, opKey: string): Promise { + const r = await sql<{ + op_key: string; + run_id: string | null; + attempt_token: string; + attempts: number; + status: "started" | "succeeded"; + result: unknown; + last_error: string | null; + }>`SELECT * FROM tasks_op_ledger WHERE op_key = ${opKey}`.execute(db); + const row = r.rows[0]; + return row + ? { + opKey: row.op_key, + runId: row.run_id, + attemptToken: row.attempt_token, + attempts: row.attempts, + status: row.status, + result: row.result, + lastError: row.last_error, + } + : null; +} diff --git a/packages/tasks/src/openworkflow-engine.ts b/packages/tasks/src/openworkflow-engine.ts new file mode 100644 index 00000000..dd2d8170 --- /dev/null +++ b/packages/tasks/src/openworkflow-engine.ts @@ -0,0 +1,321 @@ +import { randomUUID } from "node:crypto"; +import { type Kysely, sql } from "kysely"; +import { defineWorkflow, OpenWorkflow, Worker, type Workflow } from "openworkflow"; +import { BackendPostgres } from "openworkflow/postgres"; +import { TaskCanceledError } from "./errors"; +import { createLogger, type LogSink } from "./logger"; +import { engineVersion, parseEngineVersion, type TaskDefinition, type TaskRegistry } from "./registry"; +import type { + EnqueueOptions, + JsonValue, + ListFilter, + StartWorkerOptions, + TaskContext, + TaskEngine, + TaskRunStatus, + TaskRunView, + TaskStepApi, + WaitForFactOptions, +} from "./types"; + +// biome-ignore lint/suspicious/noExplicitAny: 本包只用原生 SQL 访问自有表 +type AnyDb = Kysely; +type OwWorkflowRun = NonNullable>>; +type OwStep = Parameters[1]>[0]["step"]; + +export const ENGINE_SCHEMA = "openworkflow"; +const DEFAULT_STEP_RETRY = { maximumAttempts: 3, initialInterval: "1s", backoffCoefficient: 2, maximumInterval: "60s" }; + +/** 写入引擎的 run 输入:业务输入 + 追踪元数据 */ +interface Envelope { + input: JsonValue; + traceId: string; + priority: number; +} + +export interface OpenWorkflowEngineOptions { + /** 引擎自己的 postgres.js 连接池使用(与 Kysely 同库) */ + databaseUrl: string; + /** 业务库(Kysely/pg);tasks_* 表所在 */ + db: AnyDb; + registry: TaskRegistry; + /** 固定 "openworkflow";只允许测试覆盖 */ + schema?: string; + /** 默认 true(幂等)。W2 并入统一迁移后启动时传 false */ + runMigrations?: boolean; + logSink?: LogSink; + /** handler 执行期间轮询取消标志的间隔,默认 1000ms */ + cancelPollMs?: number; +} + +export const signalName = (runId: string, name: string) => `umt:${runId}:${name}`; + +export class OpenWorkflowEngine implements TaskEngine { + private readonly ow: OpenWorkflow; + private readonly workers: Worker[] = []; + private readonly log; + + private constructor( + private readonly backend: BackendPostgres, + private readonly opts: OpenWorkflowEngineOptions, + ) { + this.ow = new OpenWorkflow({ backend }); + this.log = createLogger(opts.logSink, { component: "tasks.engine" }); + } + + static async connect(opts: OpenWorkflowEngineOptions): Promise { + const backend = await BackendPostgres.connect(opts.databaseUrl, { + schema: opts.schema ?? ENGINE_SCHEMA, + runMigrations: opts.runMigrations ?? true, + }); + return new OpenWorkflowEngine(backend, opts); + } + + async enqueue(taskName: string, input: unknown, opts: EnqueueOptions) { + const def = this.opts.registry.latest(taskName); + const parsed = def.inputSchema.parse(input) as JsonValue; + const envelope: Envelope = { input: parsed, traceId: opts.traceId, priority: opts.priority ?? 0 }; + const handle = await this.ow.runWorkflow( + { name: def.name, version: engineVersion(def.version) }, + envelope as unknown as JsonValue, + { idempotencyKey: opts.idempotencyKey, ...(opts.availableAt ? { availableAt: opts.availableAt } : {}) }, + ); + const run = handle.workflowRun; + const inserted = await sql` + INSERT INTO tasks_run_meta (run_id, task_name, task_version, idempotency_key, trace_id, priority) + VALUES (${run.id}, ${def.name}, ${def.version}, ${opts.idempotencyKey}, ${opts.traceId}, ${envelope.priority}) + ON CONFLICT (run_id) DO NOTHING`.execute(this.opts.db); + // 同一幂等键返回已有 run 时 meta 行已存在 + return { runId: run.id, deduplicated: Number(inserted.numAffectedRows ?? 0n) === 0 }; + } + + async signal(runId: string, name: string, payload: JsonValue = null) { + await sql`INSERT INTO tasks_fact (run_id, name, payload) VALUES (${runId}, ${name}, ${JSON.stringify(payload)}::jsonb)`.execute( + this.opts.db, + ); + const r = await this.ow.sendSignal({ signal: signalName(runId, name), data: payload }); + return { delivered: r.workflowRunIds.length }; + } + + async cancel(runId: string, reason?: string) { + // 先写业务侧标志:运行中的 handler 通过它触发 abortSignal,步骤间检查也读它 + await sql` + INSERT INTO tasks_run_meta (run_id, task_name, task_version, idempotency_key, cancel_requested_at, cancel_reason) + VALUES (${runId}, '', 0, '', now(), ${reason ?? null}) + ON CONFLICT (run_id) DO UPDATE SET cancel_requested_at = COALESCE(tasks_run_meta.cancel_requested_at, now()), + cancel_reason = COALESCE(tasks_run_meta.cancel_reason, EXCLUDED.cancel_reason)`.execute(this.opts.db); + try { + await this.ow.cancelWorkflowRun(runId); + return true; + } catch (err) { + const run = await this.backend.getWorkflowRun({ workflowRunId: runId }); + if (run?.status === "canceled") return true; + if (run) return false; + throw err; + } + } + + async get(runId: string): Promise { + const run = await this.backend.getWorkflowRun({ workflowRunId: runId }); + if (!run) return null; + const metas = await this.loadMeta([run.id]); + return toView(run, metas.get(run.id)); + } + + async list(filter: ListFilter = {}): Promise { + const res = await this.backend.listWorkflowRuns({ + limit: filter.limit ?? 50, + ...(filter.taskName ? { workflowName: filter.taskName } : {}), + ...(filter.status ? { status: filter.status } : {}), + }); + const metas = await this.loadMeta(res.data.map((r) => r.id)); + return res.data.map((r) => toView(r, metas.get(r.id))); + } + + async startWorker(opts: StartWorkerOptions = {}) { + const defs = this.opts.registry.select(opts.versions); + const workflows = defs.map((d) => this.toWorkflow(d)) as Workflow[]; + const worker = new Worker({ backend: this.backend, workflows, concurrency: opts.concurrency ?? 4 }); + await worker.start(); + this.workers.push(worker); + this.log.info("worker started", { tasks: defs.map((d) => `${d.name}@${d.version}`) }); + } + + /** 等进行中的 handler 执行完再断开连接池 */ + async stop() { + await Promise.all(this.workers.splice(0).map((w) => w.stop())); + await this.backend.stop(); + } + + private async loadMeta(runIds: string[]) { + const out = new Map(); + if (runIds.length === 0) return out; + const r = await sql`SELECT * FROM tasks_run_meta WHERE run_id = ANY(${runIds}::text[])`.execute( + this.opts.db, + ); + for (const row of r.rows) out.set(row.run_id, row); + return out; + } + + private async cancelState(runId: string): Promise<{ at: Date | null; reason: string | null }> { + const r = await sql<{ cancel_requested_at: Date | null; cancel_reason: string | null }>` + SELECT cancel_requested_at, cancel_reason FROM tasks_run_meta WHERE run_id = ${runId}`.execute(this.opts.db); + return { at: r.rows[0]?.cancel_requested_at ?? null, reason: r.rows[0]?.cancel_reason ?? null }; + } + + private toWorkflow(def: TaskDefinition) { + const { db } = this.opts; + const pollMs = this.opts.cancelPollMs ?? 1000; + const stepRetry = { ...DEFAULT_STEP_RETRY, ...def.retryPolicy }; + return defineWorkflow( + { name: def.name, version: engineVersion(def.version), retryPolicy: { maximumAttempts: 1 } }, + async ({ input, step, run }) => { + const env = input as unknown as Envelope; + const parsed = def.inputSchema.parse(env.input); + const attemptToken = randomUUID(); + const logger = createLogger(this.opts.logSink, { + task: def.name, + version: def.version, + runId: run.id, + traceId: env.traceId, + }); + const abort = new AbortController(); + let cancelReason: string | null = null; + const poll = async () => { + const s = await this.cancelState(run.id); + if (s.at && !abort.signal.aborted) { + cancelReason = s.reason; + abort.abort(new TaskCanceledError(run.id, s.reason)); + } + return abort.signal.aborted; + }; + const timer = setInterval(() => { + poll().catch((err) => logger.warn("cancel poll failed", { err })); + }, pollMs); + const throwIfCanceled = async () => { + if (await poll()) throw new TaskCanceledError(run.id, cancelReason); + }; + + const stepApi = makeStepApi(step as OwStep, run.id, db, stepRetry, throwIfCanceled); + const ctx: TaskContext = { + input: parsed, + runId: run.id, + taskName: def.name, + version: def.version, + traceId: env.traceId, + priority: env.priority, + attemptToken, + abortSignal: abort.signal, + logger, + step: stepApi, + throwIfCanceled, + }; + try { + await throwIfCanceled(); + const out = await def.handler(ctx); + return (out ?? null) as JsonValue; + } finally { + clearInterval(timer); + } + }, + ); + } +} + +function makeStepApi( + step: OwStep, + runId: string, + db: AnyDb, + stepRetry: typeof DEFAULT_STEP_RETRY, + throwIfCanceled: () => Promise, +): TaskStepApi { + const api: TaskStepApi = { + async run(name, fn, opts) { + await throwIfCanceled(); + // biome-ignore lint/suspicious/noExplicitAny: 引擎的 DurationString 是模板字面量类型 + const retryPolicy = { ...stepRetry, ...opts?.retryPolicy } as any; + return (await step.run({ name, retryPolicy }, async () => (await fn()) as never)) as never; + }, + async sleep(name, duration) { + await throwIfCanceled(); + // biome-ignore lint/suspicious/noExplicitAny: 同上 + await step.sleep(name, duration as any); + }, + async waitForFact(name: string, opts: WaitForFactOptions = {}) { + const check = + opts.check ?? + (async () => { + const r = await sql<{ payload: T }>` + SELECT payload FROM tasks_fact WHERE run_id = ${runId} AND name = ${name} ORDER BY id DESC LIMIT 1`.execute( + db, + ); + return r.rows[0] ? r.rows[0].payload : undefined; + }); + const maxRounds = opts.maxRounds ?? 100; + for (let i = 0; ; i++) { + const got = await api.run(`fact:${name}:check:${i}`, async () => { + const v = await check(); + return v === undefined || v === null ? { found: false } : { found: true, value: v as unknown }; + }); + if (got.found) return (got as { value: T }).value; + if (i >= maxRounds) return null; + await throwIfCanceled(); + await step.waitForSignal({ + name: `fact:${name}:wait:${i}`, + signal: signalName(runId, name), + timeout: opts.timeout ?? "5m", + }); + } + }, + }; + return api; +} + +interface MetaRow { + run_id: string; + task_name: string; + task_version: number; + idempotency_key: string; + trace_id: string | null; + priority: number; + cancel_requested_at: Date | null; + cancel_reason: string | null; +} + +function toStatus(run: OwWorkflowRun): TaskRunStatus { + switch (run.status) { + case "succeeded": + case "completed": + return "completed"; + case "sleeping": + return "waiting"; + case "running": + // 等信号 / sleep 期间 status 仍是 running,但没有 worker 持有 + return run.workerId ? "running" : "waiting"; + default: + return run.status; + } +} + +function toView(run: OwWorkflowRun, meta: MetaRow | undefined): TaskRunView { + const env = (run.input ?? {}) as Partial; + return { + runId: run.id, + taskName: run.workflowName, + version: parseEngineVersion(run.version), + status: toStatus(run), + attempts: run.attempts, + idempotencyKey: run.idempotencyKey, + traceId: meta?.trace_id ?? env.traceId ?? null, + priority: meta?.priority ?? env.priority ?? 0, + input: env.input ?? null, + output: run.output, + error: run.error ? { name: run.error.name, message: run.error.message } : null, + cancelRequestedAt: meta?.cancel_requested_at ?? null, + cancelReason: meta?.cancel_reason ?? null, + availableAt: run.availableAt, + createdAt: run.createdAt, + startedAt: run.startedAt, + finishedAt: run.finishedAt, + }; +} diff --git a/packages/tasks/src/outbox-relay.ts b/packages/tasks/src/outbox-relay.ts new file mode 100644 index 00000000..cfdc840d --- /dev/null +++ b/packages/tasks/src/outbox-relay.ts @@ -0,0 +1,112 @@ +import { randomUUID } from "node:crypto"; +import { type Kysely, sql } from "kysely"; +import { createLogger, type LogSink } from "./logger"; +import type { TaskEngine } from "./types"; + +// biome-ignore lint/suspicious/noExplicitAny: 只用原生 SQL 读写 outbox_event +type AnyDb = Kysely; + +export interface OutboxEvent { + id: string; + type: string; + payload: unknown; + at: Date; +} + +export interface OutboxRoute { + task: string; + input: unknown; + /** 缺省用事件 payload.traceId,再缺省生成新的 */ + traceId?: string; + priority?: number; +} + +/** 返回 null 表示该事件无需建任务,直接标记已投递 */ +export type OutboxRouteFn = (event: OutboxEvent) => OutboxRoute | null; + +export interface OutboxRelayOptions { + db: AnyDb; + engine: Pick; + /** 事件 type → 任务映射;只认领 routes 里列出的 type */ + routes: Record; + batchSize?: number; + intervalMs?: number; + logSink?: LogSink; +} + +/** + * Outbox 中继:业务事务写 outbox_event,中继在自己的事务里 + * `SELECT … FOR UPDATE SKIP LOCKED` 认领未投递行 → enqueue(幂等键 = outbox:<行 id>)→ 标记 published_at。 + * 多个中继进程并发时互不重复认领;enqueue 成功但标记前崩溃,重投时由引擎幂等键去重(24h), + * 更长的去重由业务唯一约束兜底。 + */ +export class OutboxRelay { + private timer: NodeJS.Timeout | null = null; + private running: Promise | null = null; + private readonly log; + readonly id = randomUUID().slice(0, 8); + + constructor(private readonly opts: OutboxRelayOptions) { + this.log = createLogger(opts.logSink, { component: "tasks.outbox", relay: this.id }); + } + + static idempotencyKey(eventId: string) { + return `outbox:${eventId}`; + } + + /** 处理一批;返回本批已投递与失败数 */ + async pollOnce(): Promise<{ published: number; failed: number }> { + const types = Object.keys(this.opts.routes); + if (types.length === 0) return { published: 0, failed: 0 }; + let published = 0; + let failed = 0; + await this.opts.db.transaction().execute(async (trx) => { + const rows = await sql` + SELECT id, type, payload, at FROM outbox_event + WHERE published_at IS NULL AND type = ANY(${types}::text[]) + ORDER BY at, id + LIMIT ${this.opts.batchSize ?? 50} + FOR UPDATE SKIP LOCKED`.execute(trx); + for (const ev of rows.rows) { + try { + const route = this.opts.routes[ev.type]!(ev); + if (route) { + const payloadTrace = (ev.payload as { traceId?: unknown } | null)?.traceId; + await this.opts.engine.enqueue(route.task, route.input, { + idempotencyKey: OutboxRelay.idempotencyKey(ev.id), + traceId: route.traceId ?? (typeof payloadTrace === "string" ? payloadTrace : randomUUID()), + ...(route.priority === undefined ? {} : { priority: route.priority }), + }); + } + await sql`UPDATE outbox_event SET published_at = now() WHERE id = ${ev.id}`.execute(trx); + published++; + } catch (err) { + failed++; + // 留在未投递状态,下一轮重试 + this.log.error("outbox dispatch failed", { eventId: ev.id, type: ev.type, err }); + } + } + }); + return { published, failed }; + } + + start() { + if (this.timer) return; + const tick = () => { + if (this.running) return; + this.running = this.pollOnce() + .catch((err) => this.log.error("outbox poll failed", { err })) + .finally(() => { + this.running = null; + }); + }; + this.timer = setInterval(tick, this.opts.intervalMs ?? 1000); + tick(); + } + + async stop() { + if (this.timer) clearInterval(this.timer); + this.timer = null; + await this.running; + } +} diff --git a/packages/tasks/src/registry.ts b/packages/tasks/src/registry.ts new file mode 100644 index 00000000..377be77f --- /dev/null +++ b/packages/tasks/src/registry.ts @@ -0,0 +1,77 @@ +import type { z } from "zod"; +import { UnknownTaskError } from "./errors"; +import type { RetryPolicy, TaskContext } from "./types"; + +export interface TaskDefinition { + name: string; + /** 正整数;不兼容改动 +1,并在 worker 上同时注册新旧版本直到旧 run 清空 */ + version: number; + inputSchema: S; + /** 步骤默认重试策略(步骤可单独覆盖)。外部网络类建议 5 次 1s→60s,文件类 3 次 */ + retryPolicy?: Partial; + handler: (ctx: TaskContext>) => Promise; +} + +export function defineTask(def: TaskDefinition): TaskDefinition { + 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}`); + return def; +} + +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>(); + + // biome-ignore lint/suspicious/noExplicitAny: 注册表按名存放异构的任务定义 + register(...defs: TaskDefinition[]): 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()))!; + } + + 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; + } +} diff --git a/packages/tasks/src/types.ts b/packages/tasks/src/types.ts new file mode 100644 index 00000000..ee0324e3 --- /dev/null +++ b/packages/tasks/src/types.ts @@ -0,0 +1,113 @@ +import type { TaskLogger } from "./logger"; + +export type JsonValue = string | number | boolean | null | JsonValue[] | { [key: string]: JsonValue }; + +/** 与 OpenWorkflow 的 DurationString 一致:"500ms" / "5s" / "10m" / "2 hours" */ +export type Duration = `${number}` | `${number}${string}` | `${number} ${string}`; + +export interface RetryPolicy { + maximumAttempts: number; + initialInterval: Duration; + backoffCoefficient: number; + maximumInterval: Duration; +} + +export type TaskRunStatus = "pending" | "running" | "waiting" | "completed" | "failed" | "canceled"; + +/** 只读投影,供 task.inspect / 管理页用;业务时间线仍读业务表 */ +export interface TaskRunView { + runId: string; + taskName: string; + version: number | null; + status: TaskRunStatus; + attempts: number; + idempotencyKey: string | null; + traceId: string | null; + priority: number; + input: unknown; + output: unknown; + error: { name?: string; message: string } | null; + cancelRequestedAt: Date | null; + cancelReason: string | null; + availableAt: Date | null; + createdAt: Date; + startedAt: Date | null; + finishedAt: Date | null; +} + +export interface EnqueueOptions { + /** 由服务端按语义生成(如 outbox 行 id);引擎只保证 24h 窗口内去重,长期去重靠业务唯一约束 */ + idempotencyKey: string; + traceId: string; + /** 记录在元数据里;OpenWorkflow 0.10.1 按 available_at 先后认领,暂不支持真正的优先级调度 */ + priority?: number; + availableAt?: Date; +} + +export interface ListFilter { + taskName?: string; + status?: Exclude; + limit?: number; +} + +export interface StartWorkerOptions { + /** + * 本 worker 执行哪些版本:"ingestFeed@2" 精确指定,"@1" 表示所有任务的 v1, + * 省略则执行注册表里的全部版本。升级期间新旧版本要同时列出。 + */ + versions?: string[]; + concurrency?: number; +} + +export interface TaskEngine { + enqueue(taskName: string, input: unknown, opts: EnqueueOptions): Promise<{ runId: string; deduplicated: boolean }>; + /** + * 先把事实写入 tasks_fact,再发引擎信号唤醒等待方。信号丢了不影响正确性: + * waitForFact 每轮都先查库。业务事实(审批、下载完成)应先写业务表,再调用这里。 + */ + signal(runId: string, name: string, payload?: JsonValue): Promise<{ delivered: number }>; + /** 写取消标志后取消引擎 run。不撤销已完成的外部动作;返回 false 表示 run 已结束无法取消 */ + cancel(runId: string, reason?: string): Promise; + get(runId: string): Promise; + list(filter?: ListFilter): Promise; + startWorker(opts?: StartWorkerOptions): Promise; + stop(): Promise; +} + +export interface StepRunOptions { + retryPolicy?: Partial; +} + +export interface WaitForFactOptions { + /** 业务库检查;省略时读 tasks_fact(engine.signal 写入的事实)。返回 null/undefined 表示尚未满足 */ + check?: () => Promise; + /** 每轮等待信号的超时,超时后重新查库;默认 "5m" */ + timeout?: Duration; + /** 最多等待几轮,用尽返回 null;默认 100(单个 run 上限 1000 个 step attempt) */ + maxRounds?: number; +} + +export interface TaskStepApi { + /** 结果持久化;重放时直接返回。步骤体必须幂等(可能在新 worker 上重跑) */ + run(name: string, fn: () => Promise | T, opts?: StepRunOptions): Promise; + sleep(name: string, duration: Duration): Promise; + /** 先查库、再带超时等信号的循环。事实先于等待到达也不会丢 */ + waitForFact(name: string, opts?: WaitForFactOptions): Promise; +} + +export interface TaskContext { + input: I; + runId: string; + taskName: string; + version: number; + traceId: string; + priority: number; + /** 每次 handler 执行(含重放)新生成;配合 withAttemptToken 做 fencing */ + attemptToken: string; + /** cancel 写 PG 标志后触发;长步骤自行轮询并中止 */ + abortSignal: AbortSignal; + logger: TaskLogger; + step: TaskStepApi; + /** 已请求取消时抛出 TaskCanceledError */ + throwIfCanceled(): Promise; +} diff --git a/packages/tasks/test/crash-recovery.test.ts b/packages/tasks/test/crash-recovery.test.ts new file mode 100644 index 00000000..c8fa5215 --- /dev/null +++ b/packages/tasks/test/crash-recovery.test.ts @@ -0,0 +1,96 @@ +import { type ChildProcess, spawn } from "node:child_process"; +import { fileURLToPath } from "node:url"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { createTestRegistry } from "./fixtures/tasks"; +import { + connectDb, + createSideEffectTable, + dropSchemas, + pgAvailable, + resetSchemas, + setupEngine, + sideEffects, + type TestDb, + testSchemas, + waitFor, +} from "./helpers"; + +const schemas = testSchemas("crash"); +const hasPg = await pgAvailable(); +const WORKER = fileURLToPath(new URL("./fixtures/worker.ts", import.meta.url)); +const PKG_DIR = fileURLToPath(new URL("..", import.meta.url)); + +function startWorker(env: Record): Promise { + const child = spawn(process.execPath, ["--import", "tsx", WORKER], { + cwd: PKG_DIR, + env: { ...process.env, SCHEMA: schemas.biz, OW_SCHEMA: schemas.ow, ...env }, + stdio: ["ignore", "pipe", "pipe"], + }); + let err = ""; + child.stderr!.on("data", (d) => { + err += d; + }); + return new Promise((resolve, reject) => { + child.stdout!.on("data", (d) => { + if (String(d).includes("worker ready")) resolve(child); + }); + child.on("exit", (code) => reject(new Error(`worker exited early code=${code}: ${err}`))); + }); +} + +describe.skipIf(!hasPg)("崩溃恢复(worker 子进程 kill -9)", () => { + let db: TestDb; + let engine: Awaited>; + const children: ChildProcess[] = []; + + beforeAll(async () => { + await resetSchemas(schemas); + db = connectDb(schemas.biz); + await createSideEffectTable(db); + engine = await setupEngine(db, schemas, createTestRegistry(db)); // 只做客户端,不起 worker + }, 30_000); + + afterAll(async () => { + for (const c of children) if (c.exitCode === null && c.signalCode === null) c.kill("SIGKILL"); + await engine?.stop(); + await db?.destroy(); + await dropSchemas(schemas); + }, 30_000); + + it("第 2 步完成后 kill -9,新 worker 续跑:第 1、2 步不重跑,第 3 步完成", async () => { + const w1 = await startWorker({ HANG_STEP3: "1" }); + children.push(w1); + const { runId } = await engine.enqueue("threeSteps", { label: "x" }, { idempotencyKey: "crash-1", traceId: "t" }); + + // 第 3 步开始 = 第 1、2 步已持久化 + await waitFor(async () => (await sideEffects(db, runId)).some((s) => s.step === "step3"), 20_000, "step3 start"); + w1.kill("SIGKILL"); + await new Promise((r) => w1.once("exit", r)); + + const w2 = await startWorker({}); + children.push(w2); + // 租约 30s 到期后被新 worker 认领 + const done = await waitFor( + async () => { + const v = await engine.get(runId); + return v?.status === "completed" ? v : null; + }, + 75_000, + "run completed", + ); + expect(done.output).toEqual({ label: "x", sum: 6 }); + + const effects = await sideEffects(db, runId); + const count = (step: string, note: string | null = null) => + effects.filter((e) => e.step === step && e.note === note).length; + expect(count("step1")).toBe(1); + expect(count("step2")).toBe(1); + expect(count("step3", "start")).toBe(2); // 被中断的步骤在新 worker 上重跑(步骤体须幂等) + expect(count("step3", "end")).toBe(1); + const step3Pids = effects.filter((e) => e.step === "step3").map((e) => e.pid); + expect(step3Pids).toEqual([w1.pid, w2.pid, w2.pid]); + + w2.kill("SIGTERM"); + await new Promise((r) => w2.once("exit", r)); + }, 120_000); +}); diff --git a/packages/tasks/test/engine.test.ts b/packages/tasks/test/engine.test.ts new file mode 100644 index 00000000..6a85e006 --- /dev/null +++ b/packages/tasks/test/engine.test.ts @@ -0,0 +1,109 @@ +import { sql } from "kysely"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import type { OpenWorkflowEngine } from "../src"; +import { createTestRegistry } from "./fixtures/tasks"; +import { + connectDb, + createSideEffectTable, + dropSchemas, + pgAvailable, + resetSchemas, + setupEngine, + sideEffects, + type TestDb, + testSchemas, + waitFor, +} from "./helpers"; + +const schemas = testSchemas("engine"); +const hasPg = await pgAvailable(); + +describe.skipIf(!hasPg)("OpenWorkflowEngine(进程内 worker)", () => { + let db: TestDb; + let engine: OpenWorkflowEngine; + + beforeAll(async () => { + await resetSchemas(schemas); + db = connectDb(schemas.biz); + await createSideEffectTable(db); + engine = await setupEngine(db, schemas, createTestRegistry(db)); + await engine.startWorker({ concurrency: 4 }); + }, 30_000); + + afterAll(async () => { + await engine?.stop(); + await db?.destroy(); + await dropSchemas(schemas); + }, 60_000); + + it("enqueue 幂等:同 key 返回同一 run,不同 key 新建;traceId 进入 handler", async () => { + const a = await engine.enqueue("echo", { n: 1 }, { idempotencyKey: "k-1", traceId: "t-1", priority: 5 }); + const b = await engine.enqueue("echo", { n: 2 }, { idempotencyKey: "k-1", traceId: "t-2" }); + const c = await engine.enqueue("echo", { n: 3 }, { idempotencyKey: "k-2", traceId: "t-3" }); + expect(a.deduplicated).toBe(false); + expect(b).toEqual({ runId: a.runId, deduplicated: true }); + expect(c.runId).not.toBe(a.runId); + + const done = await waitFor(async () => { + const v = await engine.get(a.runId); + return v?.status === "completed" ? v : null; + }, 15_000); + expect(done.output).toEqual({ n: 1, trace: "t-1" }); + expect(done).toMatchObject({ taskName: "echo", version: 1, traceId: "t-1", priority: 5, idempotencyKey: "k-1" }); + + const listed = await engine.list({ taskName: "echo" }); + expect(listed.map((r) => r.runId).sort()).toEqual([a.runId, c.runId].sort()); + }, 30_000); + + it("enqueue 拒绝不合 schema 的输入与未注册任务", async () => { + await expect(engine.enqueue("echo", { n: "x" }, { idempotencyKey: "bad", traceId: "t" })).rejects.toThrow(); + await expect(engine.enqueue("nope", {}, { idempotencyKey: "bad2", traceId: "t" })).rejects.toThrow( + /not registered/, + ); + }); + + it("取消:进行中的长步骤经 abortSignal 中止,后续步骤不执行", async () => { + const { runId } = await engine.enqueue("cancellable", {}, { idempotencyKey: "c-1", traceId: "t" }); + await waitFor(async () => (await sideEffects(db, runId)).some((s) => s.step === "long"), 15_000, "long start"); + expect(await engine.cancel(runId, "user requested")).toBe(true); + + await waitFor( + async () => (await sideEffects(db, runId)).some((s) => s.note === "aborted"), + 5_000, + "abort propagated", + ); + const v = await engine.get(runId); + expect(v?.status).toBe("canceled"); + expect(v?.cancelReason).toBe("user requested"); + await new Promise((r) => setTimeout(r, 1500)); + expect((await sideEffects(db, runId)).map((s) => s.step)).not.toContain("after"); + // 已取消再取消:幂等 + expect(await engine.cancel(runId)).toBe(true); + }, 30_000); + + it("取消已完成的 run 返回 false", async () => { + const { runId } = await engine.enqueue("echo", { n: 9 }, { idempotencyKey: "c-done", traceId: "t" }); + await waitFor(async () => (await engine.get(runId))?.status === "completed", 15_000); + expect(await engine.cancel(runId)).toBe(false); + }, 30_000); + + it("waitForFact:事实先于等待到达不丢失;等待中到达的事实经信号唤醒", async () => { + const { runId } = await engine.enqueue("waitFacts", {}, { idempotencyKey: "f-1", traceId: "t" }); + // 此刻 handler 还在 before 步骤(1.5s),没有等待者,引擎信号会丢,但事实已落库 + const early = await engine.signal(runId, "a", { approved: true }); + expect(early.delivered).toBe(0); + + await waitFor(async () => (await sideEffects(db, runId)).some((s) => s.step === "got-a"), 15_000, "got-a"); + await waitFor(async () => (await engine.get(runId))?.status === "waiting", 10_000, "waiting on b"); + const late = await engine.signal(runId, "b", 42); + expect(late.delivered).toBe(1); + + const v = await waitFor(async () => { + const r = await engine.get(runId); + return r?.status === "completed" ? r : null; + }, 15_000); + expect(v.output).toEqual({ a: { approved: true }, b: 42 }); + const facts = await sql<{ n: string }>`SELECT count(*) n FROM tasks_fact WHERE run_id = ${runId}`.execute(db); + expect(Number(facts.rows[0]!.n)).toBe(2); + }, 45_000); +}); diff --git a/packages/tasks/test/fixtures/tasks.ts b/packages/tasks/test/fixtures/tasks.ts new file mode 100644 index 00000000..ee1e0ebf --- /dev/null +++ b/packages/tasks/test/fixtures/tasks.ts @@ -0,0 +1,86 @@ +import { sql } from "kysely"; +import { z } from "zod"; +import { defineTask, TaskRegistry } from "../../src"; +import type { TestDb } from "../helpers"; + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +/** 测试任务;副作用写 side_effect 表,用来判断步骤是否重跑 */ +export function createTestRegistry(db: TestDb) { + const mark = (runId: string, step: string, note: string | null = null) => + sql`INSERT INTO side_effect (run_id, step, note, pid) VALUES (${runId}, ${step}, ${note}, ${process.pid})`.execute( + db, + ); + + const threeSteps = defineTask({ + name: "threeSteps", + version: 1, + inputSchema: z.object({ label: z.string() }), + handler: async (ctx) => { + const a = await ctx.step.run("step1", async () => { + await mark(ctx.runId, "step1"); + return 1; + }); + const b = await ctx.step.run("step2", async () => { + await mark(ctx.runId, "step2"); + return 2; + }); + const c = await ctx.step.run("step3", async () => { + await mark(ctx.runId, "step3", "start"); + // 崩溃测试:第一个 worker 卡在这里被 kill -9 + if (process.env.HANG_STEP3 === "1") await sleep(10 * 60_000); + await mark(ctx.runId, "step3", "end"); + return 3; + }); + return { label: ctx.input.label, sum: a + b + c }; + }, + }); + + const cancellable = defineTask({ + name: "cancellable", + version: 1, + inputSchema: z.object({}), + handler: async (ctx) => { + await ctx.step.run("first", () => mark(ctx.runId, "first").then(() => 1)); + await ctx.step.run("long", async () => { + await mark(ctx.runId, "long", "start"); + for (let i = 0; i < 300; i++) { + if (ctx.abortSignal.aborted) { + await mark(ctx.runId, "long", "aborted"); + throw ctx.abortSignal.reason; + } + await sleep(100); + } + return 1; + }); + await ctx.step.run("after", () => mark(ctx.runId, "after").then(() => 1)); + return "done"; + }, + }); + + const waitFacts = defineTask({ + name: "waitFacts", + version: 1, + inputSchema: z.object({}), + handler: async (ctx) => { + await ctx.step.run("before", async () => { + await mark(ctx.runId, "before"); + await sleep(1500); + return 1; + }); + const a = await ctx.step.waitForFact("a", { timeout: "60s" }); + await ctx.step.run("got-a", () => mark(ctx.runId, "got-a").then(() => 1)); + const b = await ctx.step.waitForFact("b", { timeout: "60s" }); + return { a, b }; + }, + }); + + const echo = defineTask({ + name: "echo", + version: 1, + inputSchema: z.object({ n: z.number() }), + handler: async (ctx) => ctx.step.run("echo", async () => ({ n: ctx.input.n, trace: ctx.traceId })), + }); + + return new TaskRegistry().register(threeSteps, cancellable, waitFacts, echo); +} diff --git a/packages/tasks/test/fixtures/worker.ts b/packages/tasks/test/fixtures/worker.ts new file mode 100644 index 00000000..1006c109 --- /dev/null +++ b/packages/tasks/test/fixtures/worker.ts @@ -0,0 +1,13 @@ +// 崩溃恢复测试用的 worker 子进程:SCHEMA / OW_SCHEMA 指定测试 schema,HANG_STEP3=1 让第 3 步卡住 +import { connectDb, setupEngine } from "../helpers"; +import { createTestRegistry } from "./tasks"; + +const db = connectDb(process.env.SCHEMA!); +const engine = await setupEngine(db, { ow: process.env.OW_SCHEMA! }, createTestRegistry(db)); +await engine.startWorker({ versions: ["threeSteps@1"], concurrency: 2 }); +console.log(`worker ready pid=${process.pid}`); +process.on("SIGTERM", async () => { + await engine.stop(); + await db.destroy(); + process.exit(0); +}); diff --git a/packages/tasks/test/helpers.ts b/packages/tasks/test/helpers.ts new file mode 100644 index 00000000..3b4fb86d --- /dev/null +++ b/packages/tasks/test/helpers.ts @@ -0,0 +1,102 @@ +import { Kysely, PostgresDialect, sql } from "kysely"; +import pg from "pg"; +import { migrateTasksTables, OpenWorkflowEngine, silentSink, type TaskRegistry } from "../src"; + +export const PG_URL = + process.env.TASKS_TEST_DATABASE_URL ?? process.env.DATABASE_URL ?? "postgres://umt:umt@localhost:55433/umt"; + +/** 探测开发库;不可达时集成测试跳过(CI 机器应提供 55433) */ +export async function pgAvailable(): Promise { + const c = new pg.Client({ connectionString: PG_URL, connectionTimeoutMillis: 2000 }); + try { + await c.connect(); + await c.end(); + return true; + } catch { + console.warn(`[tasks] PostgreSQL ${PG_URL} 不可达,跳过集成测试`); + return false; + } +} + +// biome-ignore lint/suspicious/noExplicitAny: 测试直接用原生 SQL +export type TestDb = Kysely; + +/** 每个测试文件用独立 schema:业务表 ,引擎表 _ow */ +export function testSchemas(suffix: string) { + const biz = `tasks_test_${suffix}`; + return { biz, ow: `${biz}_ow` }; +} + +export function connectDb(schema: string): TestDb { + const pool = new pg.Pool({ connectionString: PG_URL, max: 6, options: `-c search_path=${schema}` }); + return new Kysely({ dialect: new PostgresDialect({ pool }) }); +} + +export async function resetSchemas(schemas: { biz: string; ow: string }) { + const admin = connectDb("public"); + try { + await sql.raw(`DROP SCHEMA IF EXISTS ${schemas.biz} CASCADE`).execute(admin); + await sql.raw(`DROP SCHEMA IF EXISTS ${schemas.ow} CASCADE`).execute(admin); + await sql.raw(`CREATE SCHEMA ${schemas.biz}`).execute(admin); + } finally { + await admin.destroy(); + } +} + +export async function dropSchemas(schemas: { biz: string; ow: string }) { + const admin = connectDb("public"); + try { + await sql.raw(`DROP SCHEMA IF EXISTS ${schemas.biz} CASCADE`).execute(admin); + await sql.raw(`DROP SCHEMA IF EXISTS ${schemas.ow} CASCADE`).execute(admin); + } finally { + await admin.destroy(); + } +} + +/** 与 packages/db 0001_init 的 outbox_event 同构 */ +export async function createOutboxTable(db: TestDb) { + await sql`CREATE TABLE outbox_event ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + at timestamptz NOT NULL DEFAULT now(), + type text NOT NULL, + payload jsonb NOT NULL DEFAULT '{}'::jsonb, + published_at timestamptz)`.execute(db); +} + +export async function createSideEffectTable(db: TestDb) { + await sql`CREATE TABLE IF NOT EXISTS side_effect ( + id bigserial PRIMARY KEY, run_id text NOT NULL, step text NOT NULL, note text, pid int, + at timestamptz NOT NULL DEFAULT now())`.execute(db); +} + +export async function sideEffects(db: TestDb, runId: string) { + const r = await sql<{ step: string; note: string | null; pid: number }>` + SELECT step, note, pid FROM side_effect WHERE run_id = ${runId} ORDER BY id`.execute(db); + return r.rows; +} + +export async function setupEngine(db: TestDb, schemas: { ow: string }, registry: TaskRegistry) { + await migrateTasksTables(db); + return OpenWorkflowEngine.connect({ + databaseUrl: PG_URL, + db, + registry, + schema: schemas.ow, + logSink: silentSink, + cancelPollMs: 200, + }); +} + +export async function waitFor( + fn: () => Promise, + timeoutMs: number, + label = "condition", +) { + const deadline = Date.now() + timeoutMs; + for (;;) { + const v = await fn(); + if (v) return v; + if (Date.now() > deadline) throw new Error(`timeout waiting for ${label}`); + await new Promise((r) => setTimeout(r, 200)); + } +} diff --git a/packages/tasks/test/index.test.ts b/packages/tasks/test/index.test.ts deleted file mode 100644 index 97a38026..00000000 --- a/packages/tasks/test/index.test.ts +++ /dev/null @@ -1,6 +0,0 @@ -import { expect, it } from "vitest"; -import { PACKAGE_NAME } from "../src"; - -it("包入口可加载", () => { - expect(PACKAGE_NAME).toBe("@umt/tasks"); -}); diff --git a/packages/tasks/test/op-ledger.test.ts b/packages/tasks/test/op-ledger.test.ts new file mode 100644 index 00000000..411dafde --- /dev/null +++ b/packages/tasks/test/op-ledger.test.ts @@ -0,0 +1,105 @@ +import { sql } from "kysely"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { FencingError, getOperation, migrateTasksTables, withAttemptToken } from "../src"; +import { connectDb, dropSchemas, pgAvailable, resetSchemas, type TestDb, testSchemas } from "./helpers"; + +const schemas = testSchemas("ledger"); +const hasPg = await pgAvailable(); + +describe.skipIf(!hasPg)("withAttemptToken(操作账本 fencing)", () => { + let db: TestDb; + + beforeAll(async () => { + await resetSchemas(schemas); + db = connectDb(schemas.biz); + await migrateTasksTables(db); + await sql`CREATE TABLE download_task (id text PRIMARY KEY, hash text)`.execute(db); + }); + + afterAll(async () => { + await db?.destroy(); + await dropSchemas(schemas); + }); + + it("成功后记录结果;再次调用直接返回结果不重做", async () => { + let calls = 0; + const run = () => + withAttemptToken( + db, + "op-1", + async (info) => { + calls++; + return { hash: "h1", attempt: info.attempt }; + }, + { + runId: "r1", + commit: async (trx, result) => { + await sql`INSERT INTO download_task (id, hash) VALUES ('d1', ${result.hash})`.execute(trx); + }, + }, + ); + expect(await run()).toEqual({ hash: "h1", attempt: 1 }); + expect(await run()).toEqual({ hash: "h1", attempt: 1 }); + expect(calls).toBe(1); + const row = await getOperation(db, "op-1"); + expect(row).toMatchObject({ status: "succeeded", attempts: 1, runId: "r1" }); + }); + + it("租约丢失:旧 attempt 执行完后提交被拒绝,新 attempt 的结果生效", async () => { + let releaseOld!: () => void; + const oldGate = new Promise((r) => { + releaseOld = r; + }); + let oldStarted!: () => void; + const oldStartedP = new Promise((r) => { + oldStarted = r; + }); + + // 旧 worker(被 SIGSTOP 之类卡住):已开始执行,尚未提交 + const oldAttempt = withAttemptToken( + db, + "op-2", + async () => { + oldStarted(); + await oldGate; + return "old"; + }, + { + commit: async (trx) => void (await sql`INSERT INTO download_task (id, hash) VALUES ('d2', 'old')`.execute(trx)), + }, + ); + await oldStartedP; + + // 新 worker 接管同一操作:resumed=true,需要先对账外部状态 + let resumed = false; + const fresh = await withAttemptToken( + db, + "op-2", + async (info) => { + resumed = info.resumed; + return "new"; + }, + { + commit: async (trx) => void (await sql`INSERT INTO download_task (id, hash) VALUES ('d2', 'new')`.execute(trx)), + }, + ); + expect(fresh).toBe("new"); + expect(resumed).toBe(true); + + releaseOld(); + await expect(oldAttempt).rejects.toBeInstanceOf(FencingError); + const r = await sql<{ hash: string }>`SELECT hash FROM download_task WHERE id = 'd2'`.execute(db); + expect(r.rows).toEqual([{ hash: "new" }]); + expect(await getOperation(db, "op-2")).toMatchObject({ status: "succeeded", attempts: 2, result: "new" }); + }); + + it("fn 抛错:记录 last_error,不写结果,可重试", async () => { + await expect( + withAttemptToken(db, "op-3", async () => { + throw new Error("qb timeout"); + }), + ).rejects.toThrow("qb timeout"); + expect(await getOperation(db, "op-3")).toMatchObject({ status: "started", lastError: "qb timeout" }); + expect(await withAttemptToken(db, "op-3", async (i) => i.attempt)).toBe(2); + }); +}); diff --git a/packages/tasks/test/outbox.test.ts b/packages/tasks/test/outbox.test.ts new file mode 100644 index 00000000..565ea205 --- /dev/null +++ b/packages/tasks/test/outbox.test.ts @@ -0,0 +1,105 @@ +import { sql } from "kysely"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { type EnqueueOptions, OutboxRelay, type TaskEngine } from "../src"; +import { createTestRegistry } from "./fixtures/tasks"; +import { + connectDb, + createOutboxTable, + createSideEffectTable, + dropSchemas, + pgAvailable, + resetSchemas, + setupEngine, + type TestDb, + testSchemas, +} from "./helpers"; + +const schemas = testSchemas("outbox"); +const hasPg = await pgAvailable(); + +describe.skipIf(!hasPg)("OutboxRelay", () => { + let db: TestDb; + let engine: Awaited>; + + beforeAll(async () => { + await resetSchemas(schemas); + db = connectDb(schemas.biz); + await createOutboxTable(db); + await createSideEffectTable(db); + engine = await setupEngine(db, schemas, createTestRegistry(db)); + }, 30_000); + + afterAll(async () => { + await engine?.stop(); + await db?.destroy(); + await dropSchemas(schemas); + }, 30_000); + + it("两个 relay 并发处理 60 条事件:每条只 enqueue 一次,全部标记已投递", async () => { + // 业务事务写 outbox + await db.transaction().execute(async (trx) => { + for (let i = 0; i < 60; i++) { + await sql`INSERT INTO outbox_event (type, payload) VALUES ('echo.requested', ${JSON.stringify({ n: i, traceId: `tr-${i}` })}::jsonb)`.execute( + trx, + ); + } + await sql`INSERT INTO outbox_event (type, payload) VALUES ('other.type', '{}'::jsonb)`.execute(trx); + }); + + const calls: string[] = []; + // 包一层计数,确认没有重复 enqueue(引擎幂等只是兜底) + const counting: Pick = { + enqueue: async (task: string, input: unknown, opts: EnqueueOptions) => { + calls.push(opts.idempotencyKey); + await new Promise((r) => setTimeout(r, 5)); + return engine.enqueue(task, input, opts); + }, + }; + const routes = { "echo.requested": (ev: { payload: unknown }) => ({ task: "echo", input: ev.payload }) }; + const relays = [0, 1].map((i) => new OutboxRelay({ db, engine: counting, routes, batchSize: 7 + i })); + + const drain = async (r: OutboxRelay) => { + const perRelay = { published: 0 }; + for (;;) { + const res = await r.pollOnce(); + perRelay.published += res.published; + if (res.published === 0 && res.failed === 0) return perRelay.published; + } + }; + const [p0, p1] = await Promise.all(relays.map(drain)); + + expect(p0! + p1!).toBe(60); + expect(p0).toBeGreaterThan(0); + expect(p1).toBeGreaterThan(0); + expect(calls).toHaveLength(60); + expect(new Set(calls).size).toBe(60); + + const pending = await sql<{ type: string }>`SELECT type FROM outbox_event WHERE published_at IS NULL`.execute(db); + expect(pending.rows).toEqual([{ type: "other.type" }]); // 未配置路由的 type 不认领 + + const runs = await sql<{ n: string; keys: string }>` + SELECT count(*) n, count(DISTINCT idempotency_key) keys FROM tasks_run_meta WHERE task_name = 'echo'`.execute(db); + expect(runs.rows[0]).toEqual({ n: "60", keys: "60" }); + const traced = await sql<{ trace_id: string }>` + SELECT m.trace_id FROM tasks_run_meta m + JOIN outbox_event o ON m.idempotency_key = 'outbox:' || o.id::text + WHERE o.payload->>'n' = '3'`.execute(db); + expect(traced.rows).toEqual([{ trace_id: "tr-3" }]); + }, 60_000); + + it("enqueue 失败的事件保持未投递,下一轮重试", async () => { + await sql`INSERT INTO outbox_event (type, payload) VALUES ('echo.requested', '{"n":"not-a-number"}'::jsonb)`.execute( + db, + ); + const relay = new OutboxRelay({ + db, + engine, + routes: { "echo.requested": (ev) => ({ task: "echo", input: ev.payload }) }, + logSink: () => {}, + }); + expect(await relay.pollOnce()).toEqual({ published: 0, failed: 1 }); + const left = await sql<{ n: string }>` + SELECT count(*) n FROM outbox_event WHERE published_at IS NULL AND type = 'echo.requested'`.execute(db); + expect(left.rows[0]!.n).toBe("1"); + }); +}); diff --git a/packages/tasks/test/registry.test.ts b/packages/tasks/test/registry.test.ts new file mode 100644 index 00000000..81d29951 --- /dev/null +++ b/packages/tasks/test/registry.test.ts @@ -0,0 +1,49 @@ +import { describe, expect, it } from "vitest"; +import { z } from "zod"; +import { createDefaultRegistry, defineTask, redact, TASK_NAMES, TaskRegistry, UnknownTaskError } from "../src"; + +describe("TaskRegistry", () => { + it("注册任务书 §3.1-8 的九个空壳任务", () => { + const reg = createDefaultRegistry(); + expect(reg.names()).toEqual([...TASK_NAMES].sort()); + for (const n of TASK_NAMES) expect(reg.latest(n).version).toBe(1); + }); + + it("latest 取最高版本;select 按 name@version / @version 过滤", () => { + const mk = (name: string, version: number) => + defineTask({ name, version, inputSchema: z.object({}), handler: async () => null }); + const reg = new TaskRegistry().register(mk("a", 1), mk("a", 2), mk("b", 1)); + expect(reg.latest("a").version).toBe(2); + expect(reg.select(["a@1"]).map((d) => `${d.name}@${d.version}`)).toEqual(["a@1"]); + expect(reg.select(["@1"]).map((d) => `${d.name}@${d.version}`)).toEqual(["a@1", "b@1"]); + expect(reg.select(["a"])).toHaveLength(2); + expect(reg.select()).toHaveLength(3); + expect(() => reg.register(mk("a", 1))).toThrow(/already registered/); + expect(() => reg.latest("zzz")).toThrow(UnknownTaskError); + expect(() => mk("a", 0)).toThrow(/invalid task version/); + }); + + it("空壳 handler 回显输入", async () => { + const def = createDefaultRegistry().latest("submitDownload"); + const input = def.inputSchema.parse({ acquisitionPlanId: "p1" }); + const out = await def.handler({ input, logger: { info() {} } } as never); + expect(out).toEqual({ todo: true, task: "submitDownload", input: { acquisitionPlanId: "p1" } }); + expect(() => def.inputSchema.parse({})).toThrow(); + }); +}); + +describe("redact", () => { + it("敏感键与 URL 凭证参数脱敏", () => { + expect( + redact({ + url: "https://pt.example/download.php?id=1&passkey=abcdef123", + nested: { apiKey: "k", password: "p", ok: 1 }, + list: ["http://x/rss?sign=zz&x=1"], + }), + ).toEqual({ + url: "https://pt.example/download.php?id=1&passkey=***", + nested: { apiKey: "***", password: "***", ok: 1 }, + list: ["http://x/rss?sign=***&x=1"], + }); + }); +});