diff --git a/deno.json b/deno.json index 25c6425d..e1bf1669 100644 --- a/deno.json +++ b/deno.json @@ -65,7 +65,7 @@ "types": "mkdir -p types && deno types > types/deno.d.ts", "typecheck": "deno task types && deno task typecheck:tsc && deno task typecheck:deno", "typecheck:tsc": "deno task --members -j 1 typecheck", - "typecheck:deno": "deno check apps/server/src/main.ts packages/contracts/src/index.ts packages/db/src/index.ts packages/db/scripts/migrate.ts packages/db/scripts/gen-types.ts packages/db/migrations/*.ts packages/domain/src/index.ts packages/tasks/src/index.ts packages/tasks/migrations/*.ts packages/agent/src/index.ts packages/adapters/src/index.ts packages/testkit/src/index.ts", + "typecheck:deno": "deno check apps/server/src/main.ts packages/contracts/src/index.ts packages/db/src/index.ts packages/db/scripts/migrate.ts packages/db/scripts/gen-types.ts packages/db/migrations/*.ts packages/domain/src/index.ts packages/tasks/src/index.ts packages/agent/src/index.ts packages/adapters/src/index.ts packages/testkit/src/index.ts", "lint": "deno run -A npm:@biomejs/biome check .", "fmt": "deno run -A npm:@biomejs/biome check --write .", "test": "deno task --members -j 1 test", diff --git a/packages/db/README.md b/packages/db/README.md index 2be47f62..64866ac0 100644 --- a/packages/db/README.md +++ b/packages/db/README.md @@ -27,8 +27,10 @@ PostgreSQL 访问层:Kysely 客户端与连接池、表类型(`src/types.ts`)、 | `0010_agent` | agent_run、artifact、decision、approval、automation_grant | | `0011_ops` | reconcile_run、server_mapping | | `0012_organization` | organization_record(change_seq 序列 + 触发器)、organization_file_result;回填 asset_operation.organization_record_id | +| `0013_tasks_tables` | 并入 @umt/tasks 自有表:tasks_run_meta、tasks_fact、tasks_op_ledger(if not exists,可接管 W1 旧库) | +| `0014_agent_run_event` | agent_run_event:Agent 运行事件,主键 (run_id, seq),删 run 级联 | -## 表清单(41 张) +## 表清单(45 张) ### 底座 / 运维 | 表 | 用途与关键不变量 | @@ -100,6 +102,14 @@ PostgreSQL 访问层:Kysely 客户端与连接池、表类型(`src/types.ts`)、 | `decision` | 待处理中心条目;同一 `dedupe_key` 同时只有一个未关闭条目 | | `approval` | 单次审批,绑定 plan + payload_hash + expected_revision | | `automation_grant` | 持续授权(动作 + 范围 + 预算 + 策略版本),可撤销 | +| `agent_run_event` | Agent 运行事件(9 种,见 contracts `AgentRunEventType`);**(run_id, seq) 主键**,先落库再推 SSE;payload = 事件主体去掉 type | + +### 任务引擎(@umt/tasks 内部使用,不经 API 暴露) +| 表 | 用途与关键不变量 | +|---|---| +| `tasks_run_meta` | 引擎 run 的业务侧元数据(traceId/优先级/取消标志) | +| `tasks_fact` | signal 前先落库的事实,waitForFact 以它为准 | +| `tasks_op_ledger` | 操作账本 fencing:attempt_token,旧 attempt 提交被拒 | ### 整理 | 表 | 用途与关键不变量 | diff --git a/packages/db/migrations/0013_tasks_tables.ts b/packages/db/migrations/0013_tasks_tables.ts new file mode 100644 index 00000000..c3be3be7 --- /dev/null +++ b/packages/db/migrations/0013_tasks_tables.ts @@ -0,0 +1,60 @@ +import type { Kysely } from "kysely"; +import { dropTables, execDdl, oneOf } from "../src/migration-helpers.ts"; + +// @umt/tasks 的三张自有表并入统一迁移(W1 审查 S2;原 packages/tasks/migrations/0001_tasks_tables)。 +// 表结构与原迁移一致,tasks 包通过注入的 Kysely 访问: +// - tasks_run_meta:引擎 run 的业务侧元数据(traceId/优先级/取消标志),取消标志是 abortSignal 的来源 +// - tasks_fact:signal 之前先落库的事实;waitForFact 以它为准,信号只用来加速唤醒 +// - tasks_op_ledger:操作账本(fencing),记录 attempt_token,旧 attempt 的提交会被拒绝 +// 用 if not exists:W1 期间用 migrateTasksTables 建过表的库可以直接接管;顺手删掉旧的独立迁移记录表。 + +// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型 +export async function up(db: Kysely): Promise { + await execDdl( + db, + ` +create table if not exists tasks_run_meta ( + run_id text primary key, + task_name text not null, + task_version integer not null, + idempotency_key text not null, + trace_id text, + priority integer not null default 0, + cancel_requested_at timestamptz, + cancel_reason text, + created_at timestamptz not null default now() +); +create index if not exists tasks_run_meta_task_idx on tasks_run_meta (task_name, created_at); + +create table if not exists tasks_fact ( + id bigserial primary key, + run_id text not null, + name text not null, + payload jsonb, + created_at timestamptz not null default now() +); +create index if not exists tasks_fact_run_name_idx on tasks_fact (run_id, name, id); + +create table if not exists tasks_op_ledger ( + op_key text primary key, + run_id text, + attempt_token uuid not null, + attempts integer not null default 1, + status text not null ${oneOf("status", ["started", "succeeded"])}, + result jsonb, + last_error text, + started_at timestamptz not null default now(), + completed_at timestamptz +); + +do $$ begin + execute format('drop table if exists %I.tasks_migration, %I.tasks_migration_lock', current_schema(), current_schema()); +end $$; +`, + ); +} + +// biome-ignore lint/suspicious/noExplicitAny: 同上 +export async function down(db: Kysely): Promise { + await execDdl(db, dropTables(["tasks_op_ledger", "tasks_fact", "tasks_run_meta"])); +} diff --git a/packages/db/migrations/0014_agent_run_event.ts b/packages/db/migrations/0014_agent_run_event.ts new file mode 100644 index 00000000..b442c600 --- /dev/null +++ b/packages/db/migrations/0014_agent_run_event.ts @@ -0,0 +1,40 @@ +import type { Kysely } from "kysely"; +import { dropTables, execDdl, oneOf } from "../src/migration-helpers.ts"; + +// Agent 运行事件持久化(W1 审查 B9;REQ §10.6):先落库再推 SSE,断线/跨进程按 (run_id, seq) 回放。 +// type 取值与 contracts AgentRunEventType 一致;payload = 事件主体去掉 type(contracts agentRunEventToRow)。 + +const TYPES = [ + "run_started", + "text_delta", + "thinking_delta", + "tool_call_start", + "tool_call_end", + "tool_call_rejected", + "compaction", + "run_completed", + "run_failed", +]; + +// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型 +export async function up(db: Kysely): Promise { + await execDdl( + db, + ` +create table agent_run_event ( + run_id uuid not null references agent_run (id) on delete cascade, + seq bigint not null check (seq >= 0), + type text not null ${oneOf("type", TYPES)}, + at timestamptz not null default now(), + payload jsonb not null default '{}'::jsonb, + constraint agent_run_event_pkey primary key (run_id, seq) +); +create index agent_run_event_at_idx on agent_run_event (at); +`, + ); +} + +// biome-ignore lint/suspicious/noExplicitAny: 同上 +export async function down(db: Kysely): Promise { + await execDdl(db, dropTables(["agent_run_event"])); +} diff --git a/packages/db/src/types.ts b/packages/db/src/types.ts index 15514526..0856d6a5 100644 --- a/packages/db/src/types.ts +++ b/packages/db/src/types.ts @@ -37,6 +37,16 @@ export type AgentRunStatus = | "failed" | "cancelled" | "budget_exceeded"; +export type AgentRunEventType = + | "run_started" + | "text_delta" + | "thinking_delta" + | "tool_call_start" + | "tool_call_end" + | "tool_call_rejected" + | "compaction" + | "run_completed" + | "run_failed"; export type ApprovalStatus = "pending" | "approved" | "rejected" | "revoked" | "expired" | "consumed"; export type ArtifactKind = "evidence" | "plan" | "proposal" | "report" | "snapshot" | "log"; export type AssetLocationLocationKind = "library" | "download_cache" | "hardlink" | "external"; @@ -175,6 +185,7 @@ export type SubscriptionOrigin = "user" | "agent" | "policy" | "import"; export type SubscriptionPriority = "immediate" | "tracking" | "new_work" | "upgrade"; export type SubscriptionStatus = "active" | "paused" | "completed" | "cancelled"; export type TaskStatus = "pending" | "running" | "waiting" | "succeeded" | "failed" | "cancelled" | "dead"; +export type TasksOpLedgerStatus = "started" | "succeeded"; export type TitleAliasKind = "original" | "translated" | "short" | "site_title" | "romanized" | "other"; export type TitleAliasSource = "tmdb" | "user" | "site" | "agent" | "nfo" | "import"; @@ -250,6 +261,14 @@ export interface AgentRunTable { updated_at: GeneratedTimestamptz; } +export interface AgentRunEventTable { + run_id: string; + seq: number; + type: AgentRunEventType; + at: GeneratedTimestamptz; + payload: Generated; +} + export interface ApiKeyTable { id: Generated; name: string; @@ -951,6 +970,38 @@ export interface TaskTable { updated_at: GeneratedTimestamptz; } +export interface TasksFactTable { + id: Generated; + run_id: string; + name: string; + payload: Json | null; + created_at: GeneratedTimestamptz; +} + +export interface TasksOpLedgerTable { + op_key: string; + run_id: string | null; + attempt_token: string; + attempts: Generated; + status: TasksOpLedgerStatus; + result: Json | null; + last_error: string | null; + started_at: GeneratedTimestamptz; + completed_at: Timestamptz | null; +} + +export interface TasksRunMetaTable { + run_id: string; + task_name: string; + task_version: number; + idempotency_key: string; + trace_id: string | null; + priority: Generated; + cancel_requested_at: Timestamptz | null; + cancel_reason: string | null; + created_at: GeneratedTimestamptz; +} + export interface TitleAliasTable { id: Generated; media_entity_id: string; @@ -969,6 +1020,7 @@ export interface DB { acquisition_plan: AcquisitionPlanTable; acquisition_slot: AcquisitionSlotTable; agent_run: AgentRunTable; + agent_run_event: AgentRunEventTable; api_key: ApiKeyTable; approval: ApprovalTable; artifact: ArtifactTable; @@ -1006,6 +1058,9 @@ export interface DB { source_release: SourceReleaseTable; subscription: SubscriptionTable; task: TaskTable; + tasks_fact: TasksFactTable; + tasks_op_ledger: TasksOpLedgerTable; + tasks_run_meta: TasksRunMetaTable; title_alias: TitleAliasTable; } @@ -1018,6 +1073,9 @@ export type AcquisitionSlotUpdate = Updateable; export type AgentRunRow = Selectable; export type NewAgentRun = Insertable; export type AgentRunUpdate = Updateable; +export type AgentRunEventRow = Selectable; +export type NewAgentRunEvent = Insertable; +export type AgentRunEventUpdate = Updateable; export type ApiKeyRow = Selectable; export type NewApiKey = Insertable; export type ApiKeyUpdate = Updateable; @@ -1129,6 +1187,15 @@ export type SubscriptionUpdate = Updateable; export type TaskRow = Selectable; export type NewTask = Insertable; export type TaskUpdate = Updateable; +export type TasksFactRow = Selectable; +export type NewTasksFact = Insertable; +export type TasksFactUpdate = Updateable; +export type TasksOpLedgerRow = Selectable; +export type NewTasksOpLedger = Insertable; +export type TasksOpLedgerUpdate = Updateable; +export type TasksRunMetaRow = Selectable; +export type NewTasksRunMeta = Insertable; +export type TasksRunMetaUpdate = Updateable; export type TitleAliasRow = Selectable; export type NewTitleAlias = Insertable; export type TitleAliasUpdate = Updateable; diff --git a/packages/db/test/schema.test.ts b/packages/db/test/schema.test.ts index 7ae206f4..c3c06f92 100644 --- a/packages/db/test/schema.test.ts +++ b/packages/db/test/schema.test.ts @@ -63,8 +63,24 @@ const EXPECTED_TABLES = [ "server_mapping", "organization_record", "organization_file_result", + // tasks 包自有表(0013)与 agent 事件(0014) + "tasks_run_meta", + "tasks_fact", + "tasks_op_ledger", + "agent_run_event", ].sort(); +/** 不走 entityTable() 的表:追加型/引擎自有表,没有 uuid id + created_at/updated_at 三件套 */ +const NON_ENTITY_TABLES = [ + "settings", + "audit_event", + "outbox_event", + "tasks_run_meta", + "tasks_fact", + "tasks_op_ledger", + "agent_run_event", +]; + describe.skipIf(skip)("schema migrations (real PG)", () => { let admin: Kysely; let db: Kysely; @@ -108,7 +124,7 @@ describe.skipIf(skip)("schema migrations (real PG)", () => { select table_name, column_name, column_default from information_schema.columns where table_schema = current_schema() and column_name in ('id', 'created_at', 'updated_at')`.execute(db) ).rows; - for (const t of EXPECTED_TABLES.filter((x) => !["settings", "audit_event", "outbox_event"].includes(x))) { + for (const t of EXPECTED_TABLES.filter((x) => !NON_ENTITY_TABLES.includes(x))) { const cols = rows.filter((r) => r.table_name === t); expect(cols.map((c) => c.column_name).sort(), t).toEqual(["created_at", "id", "updated_at"]); expect(cols.find((c) => c.column_name === "id")?.column_default, t).toBe("gen_random_uuid()"); @@ -291,6 +307,23 @@ describe.skipIf(skip)("schema migrations (real PG)", () => { expect(hit.rows.map((r) => r.id)).toContain(ent.id); }); + it("agent_run_event:(run_id, seq) 唯一,删 run 级联删事件", async () => { + const run = await db + .insertInto("agent_run") + .values({ trigger: "user", goal: "g" }) + .returning("id") + .executeTakeFirstOrThrow(); + const ev = { run_id: run.id, seq: 0, type: "run_started" as const, payload: JSON.stringify({ goal: "g" }) }; + await db.insertInto("agent_run_event").values(ev).execute(); + await expect(db.insertInto("agent_run_event").values(ev).execute()).rejects.toThrow(/agent_run_event_pkey/); + await expect( + sql`insert into agent_run_event (run_id, seq, type) values (${run.id}, 1, 'bogus')`.execute(db), + ).rejects.toThrow(/check constraint/); + await db.deleteFrom("agent_run").where("id", "=", run.id).execute(); + const left = await db.selectFrom("agent_run_event").select("seq").where("run_id", "=", run.id).execute(); + expect(left).toEqual([]); + }); + it("迁移可完整 down 再 up(往返)", async () => { const migrator = createMigrator(db, { migrationTableSchema: schema }); const down = await migrator.migrateTo(NO_MIGRATIONS); diff --git a/packages/tasks/README.md b/packages/tasks/README.md index 8aa5b80a..a9f35537 100644 --- a/packages/tasks/README.md +++ b/packages/tasks/README.md @@ -16,7 +16,7 @@ | `defineTask` / `TaskRegistry` / `createDefaultRegistry()` | 任务定义与注册表;默认注册表里有 §3.1-8 的 9 个任务(目前是空壳) | | `OutboxRelay` | 用 `FOR UPDATE SKIP LOCKED` 认领 `outbox_event` 中 `status=pending` 且 `next_attempt_at` 已到的行,按 type 路由后 enqueue(幂等键取行的 `idempotency_key`,缺省 `outbox:`),成功标记 `dispatched` + `published_at`;失败 attempts+1 指数退避(上限 5 分钟),达 `maxAttempts` 置 `dead`;多进程并发时不会重复 | | `withAttemptToken(db, opKey, fn, { commit })` | 操作账本 fencing,表是 `tasks_op_ledger` | -| `migrateTasksTables(db)` | 本包迁移(`migrations/`):`tasks_run_meta`、`tasks_fact`、`tasks_op_ledger`。W2 并入 `packages/db` | +| 表 | `tasks_run_meta`、`tasks_fact`、`tasks_op_ledger` 由 `packages/db` 迁移 `0013_tasks_tables` 建立(类型见 `@umt/db` 的 `DB`);本包不再自带迁移,只通过注入的 Kysely 访问 | `TaskContext` 包含 `input`(已经过 zod 校验)、`runId`、`traceId`、`priority`、`attemptToken`(每次执行 handler 都会重新生成)、`abortSignal`、`logger`(自动脱敏)、`step.run / step.sleep / step.waitForFact`、`throwIfCanceled()`。 @@ -54,4 +54,4 @@ ## 测试 -`deno task test`(或本包 `deno task test`)会连接开发库(`DATABASE_URL`,默认 `postgres://umt:umt@localhost:55433/umt`)。每个测试文件使用独立的 `tasks_test_<名>_` schema(并发跑多份也不冲突),测试前后都会清理;库不可达时跳过集成测试。覆盖范围:kill -9 后续跑(子进程 worker)、取消、幂等 enqueue、两个 relay 并发、事实先到达、fencing。 +`deno task test`(或本包 `deno task test`)会连接开发库,每个测试 schema 先用 `@umt/db` 的 `migrateToLatest` 建全部业务表(`DATABASE_URL`,默认 `postgres://umt:umt@localhost:55433/umt`)。每个测试文件使用独立的 `tasks_test_<名>_` schema(并发跑多份也不冲突),测试前后都会清理;库不可达时跳过集成测试。覆盖范围:kill -9 后续跑(子进程 worker)、取消、幂等 enqueue、两个 relay 并发、事实先到达、fencing。 diff --git a/packages/tasks/src/index.ts b/packages/tasks/src/index.ts index 57ad8961..ada51add 100644 --- a/packages/tasks/src/index.ts +++ b/packages/tasks/src/index.ts @@ -26,7 +26,6 @@ export { silentSink, type TaskLogger, } from "./logger.ts"; -export { migrateTasksTables, TASKS_MIGRATIONS_DIR } from "./migrate.ts"; export { type AttemptInfo, assertAttemptToken, diff --git a/packages/tasks/test/crash-recovery.test.ts b/packages/tasks/test/crash-recovery.test.ts index 0ca4d3f6..c5743ef0 100644 --- a/packages/tasks/test/crash-recovery.test.ts +++ b/packages/tasks/test/crash-recovery.test.ts @@ -6,6 +6,7 @@ import { connectDb, createSideEffectTable, dropSchemas, + migrateBizSchema, pgAvailable, resetSchemas, setupEngine, @@ -46,6 +47,7 @@ describe.skipIf(!hasPg)("崩溃恢复(worker 子进程 kill -9)", () => { beforeAll(async () => { await resetSchemas(schemas); db = connectDb(schemas.biz); + await migrateBizSchema(db, schemas.biz); await createSideEffectTable(db); engine = await setupEngine(db, schemas, createTestRegistry(db)); // 只做客户端,不起 worker }, 30_000); diff --git a/packages/tasks/test/engine.test.ts b/packages/tasks/test/engine.test.ts index 7e9459eb..5674fc18 100644 --- a/packages/tasks/test/engine.test.ts +++ b/packages/tasks/test/engine.test.ts @@ -6,6 +6,7 @@ import { connectDb, createSideEffectTable, dropSchemas, + migrateBizSchema, pgAvailable, resetSchemas, setupEngine, @@ -25,6 +26,7 @@ describe.skipIf(!hasPg)("OpenWorkflowEngine(进程内 worker)", () => { beforeAll(async () => { await resetSchemas(schemas); db = connectDb(schemas.biz); + await migrateBizSchema(db, schemas.biz); await createSideEffectTable(db); engine = await setupEngine(db, schemas, createTestRegistry(db)); await engine.startWorker({ concurrency: 4 }); diff --git a/packages/tasks/test/helpers.ts b/packages/tasks/test/helpers.ts index cfc307d0..9fbfc59d 100644 --- a/packages/tasks/test/helpers.ts +++ b/packages/tasks/test/helpers.ts @@ -1,6 +1,7 @@ +import { migrateToLatest } from "@umt/db"; import { Kysely, PostgresDialect, sql } from "kysely"; import pg from "pg"; -import { migrateTasksTables, OpenWorkflowEngine, silentSink, type TaskRegistry } from "../src/index.ts"; +import { OpenWorkflowEngine, silentSink, type TaskRegistry } from "../src/index.ts"; export const PG_URL = process.env.TASKS_TEST_DATABASE_URL ?? process.env.DATABASE_URL ?? "postgres://umt:umt@localhost:55433/umt"; @@ -32,11 +33,17 @@ export function testSchemas(suffix: string) { return { biz, ow: `${biz}_ow` }; } +/** search_path 带上 public:db 迁移用到的 pg_trgm 操作符类装在 public */ export function connectDb(schema: string): TestDb { - const pool = new pg.Pool({ connectionString: PG_URL, max: 6, options: `-c search_path=${schema}` }); + const pool = new pg.Pool({ connectionString: PG_URL, max: 6, options: `-c search_path=${schema},public` }); return new Kysely({ dialect: new PostgresDialect({ pool }) }); } +/** 用 @umt/db 的统一迁移在测试 schema 里建全部业务表(含 tasks_* 与 outbox_event) */ +export async function migrateBizSchema(db: TestDb, schema: string) { + await migrateToLatest(db, { migrationTableSchema: schema }); +} + export async function resetSchemas(schemas: { biz: string; ow: string }) { const admin = connectDb("public"); try { @@ -58,21 +65,6 @@ export async function dropSchemas(schemas: { biz: string; ow: string }) { } } -/** 与 packages/db 0001_init + 0002_foundation 的 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, - idempotency_key text UNIQUE, - status text NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'dispatched', 'failed', 'dead')), - attempts integer NOT NULL DEFAULT 0, - next_attempt_at timestamptz NOT NULL DEFAULT now(), - last_error jsonb)`.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, @@ -85,8 +77,8 @@ export async function sideEffects(db: TestDb, runId: string) { return r.rows; } +/** 只连接引擎;业务表由调用方先 migrateBizSchema 建好 */ export async function setupEngine(db: TestDb, schemas: { ow: string }, registry: TaskRegistry) { - await migrateTasksTables(db); return OpenWorkflowEngine.connect({ databaseUrl: PG_URL, db, diff --git a/packages/tasks/test/op-ledger.test.ts b/packages/tasks/test/op-ledger.test.ts index 69517d53..ae250c76 100644 --- a/packages/tasks/test/op-ledger.test.ts +++ b/packages/tasks/test/op-ledger.test.ts @@ -1,7 +1,15 @@ import { sql } from "kysely"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; -import { FencingError, getOperation, migrateTasksTables, withAttemptToken } from "../src/index.ts"; -import { connectDb, dropSchemas, pgAvailable, resetSchemas, type TestDb, testSchemas } from "./helpers.ts"; +import { FencingError, getOperation, withAttemptToken } from "../src/index.ts"; +import { + connectDb, + dropSchemas, + migrateBizSchema, + pgAvailable, + resetSchemas, + type TestDb, + testSchemas, +} from "./helpers.ts"; const schemas = testSchemas("ledger"); const hasPg = await pgAvailable(); @@ -12,8 +20,8 @@ describe.skipIf(!hasPg)("withAttemptToken(操作账本 fencing)", () => { 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); + await migrateBizSchema(db, schemas.biz); + await sql`CREATE TABLE ledger_probe (id text PRIMARY KEY, hash text)`.execute(db); }); afterAll(async () => { @@ -34,7 +42,7 @@ describe.skipIf(!hasPg)("withAttemptToken(操作账本 fencing)", () => { { runId: "r1", commit: async (trx, result) => { - await sql`INSERT INTO download_task (id, hash) VALUES ('d1', ${result.hash})`.execute(trx); + await sql`INSERT INTO ledger_probe (id, hash) VALUES ('d1', ${result.hash})`.execute(trx); }, }, ); @@ -65,7 +73,7 @@ describe.skipIf(!hasPg)("withAttemptToken(操作账本 fencing)", () => { return "old"; }, { - commit: async (trx) => void (await sql`INSERT INTO download_task (id, hash) VALUES ('d2', 'old')`.execute(trx)), + commit: async (trx) => void (await sql`INSERT INTO ledger_probe (id, hash) VALUES ('d2', 'old')`.execute(trx)), }, ); await oldStartedP; @@ -80,7 +88,7 @@ describe.skipIf(!hasPg)("withAttemptToken(操作账本 fencing)", () => { return "new"; }, { - commit: async (trx) => void (await sql`INSERT INTO download_task (id, hash) VALUES ('d2', 'new')`.execute(trx)), + commit: async (trx) => void (await sql`INSERT INTO ledger_probe (id, hash) VALUES ('d2', 'new')`.execute(trx)), }, ); expect(fresh).toBe("new"); @@ -88,7 +96,7 @@ describe.skipIf(!hasPg)("withAttemptToken(操作账本 fencing)", () => { releaseOld(); await expect(oldAttempt).rejects.toBeInstanceOf(FencingError); - const r = await sql<{ hash: string }>`SELECT hash FROM download_task WHERE id = 'd2'`.execute(db); + const r = await sql<{ hash: string }>`SELECT hash FROM ledger_probe WHERE id = 'd2'`.execute(db); expect(r.rows).toEqual([{ hash: "new" }]); expect(await getOperation(db, "op-2")).toMatchObject({ status: "succeeded", attempts: 2, result: "new" }); }); diff --git a/packages/tasks/test/outbox.test.ts b/packages/tasks/test/outbox.test.ts index 4360ab11..f11712a5 100644 --- a/packages/tasks/test/outbox.test.ts +++ b/packages/tasks/test/outbox.test.ts @@ -4,9 +4,9 @@ import { type EnqueueOptions, OutboxRelay, type TaskEngine } from "../src/index. import { createTestRegistry } from "./fixtures/tasks.ts"; import { connectDb, - createOutboxTable, createSideEffectTable, dropSchemas, + migrateBizSchema, pgAvailable, resetSchemas, setupEngine, @@ -24,7 +24,7 @@ describe.skipIf(!hasPg)("OutboxRelay", () => { beforeAll(async () => { await resetSchemas(schemas); db = connectDb(schemas.biz); - await createOutboxTable(db); + await migrateBizSchema(db, schemas.biz); await createSideEffectTable(db); engine = await setupEngine(db, schemas, createTestRegistry(db)); }, 30_000); @@ -80,11 +80,12 @@ describe.skipIf(!hasPg)("OutboxRelay", () => { SELECT count(*) n FROM outbox_event WHERE status = 'dispatched' AND published_at IS NOT NULL AND attempts = 1`.execute( db, ); - expect(dispatched.rows[0]!.n).toBe("60"); + // @umt/db 把 int8 解析为 number(全局 pg 类型解析器),这里统一转成数字比较 + expect(Number(dispatched.rows[0]!.n)).toBe(60); 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" }); + expect([Number(runs.rows[0]!.n), Number(runs.rows[0]!.keys)]).toEqual([60, 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