From 90350d63e7cfa0c9f7f09699c206e50c3090b313 Mon Sep 17 00:00:00 2001 From: root Date: Tue, 21 Jul 2026 12:07:13 +0200 Subject: [PATCH] fix(tasktree): serialize schema migrations --- internal/tasktree/dispatcher.ts | 9 +- internal/tasktree/store.ts | 126 +++++++++++---- package.json | 1 + scripts/tasktree-schema-concurrency-smoke.mjs | 151 ++++++++++++++++++ 4 files changed, 255 insertions(+), 32 deletions(-) create mode 100644 scripts/tasktree-schema-concurrency-smoke.mjs diff --git a/internal/tasktree/dispatcher.ts b/internal/tasktree/dispatcher.ts index 579a49fe..9da5220f 100644 --- a/internal/tasktree/dispatcher.ts +++ b/internal/tasktree/dispatcher.ts @@ -111,7 +111,14 @@ export function createTaskTreeDispatcher(options: TaskTreeDispatcherOptions) { else throw codedError("unsupported_operation", `unsupported operation: ${String(operation)}`); return { ok: true, operation, data }; } catch (error: any) { - return { ok: false, operation: command.operation, error: { code: error?.code ?? "tasktree_error", message: error?.message ?? String(error) } }; + const migrationDetail = error?.tasktreeMigrationId + ? { migrationId: String(error.tasktreeMigrationId), sqlState: error?.code ? String(error.code) : undefined } + : undefined; + return { + ok: false, + operation: command.operation, + error: { code: error?.code ?? "tasktree_error", message: error?.message ?? String(error), ...(migrationDetail ? { detail: migrationDetail } : {}) }, + }; } }; } diff --git a/internal/tasktree/store.ts b/internal/tasktree/store.ts index b77770dc..88ae8380 100644 --- a/internal/tasktree/store.ts +++ b/internal/tasktree/store.ts @@ -11,58 +11,100 @@ import type { MdtodoImportPlan } from "./mdtodo-import.ts"; const { Pool } = pg; -const schema = [ - `CREATE TABLE IF NOT EXISTS tasktree_schema_migrations (migration_id TEXT PRIMARY KEY, applied_at TIMESTAMPTZ NOT NULL DEFAULT now())`, - `CREATE TABLE IF NOT EXISTS tasktree_groups ( - id TEXT PRIMARY KEY, name TEXT NOT NULL, description TEXT NOT NULL DEFAULT '', - created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now())`, - `CREATE TABLE IF NOT EXISTS tasktree_tasks ( - id TEXT PRIMARY KEY, group_id TEXT NOT NULL REFERENCES tasktree_groups(id) ON DELETE CASCADE, - parent_id TEXT REFERENCES tasktree_tasks(id) ON DELETE CASCADE, kind TEXT NOT NULL CHECK (kind IN ('task','subtask','subsubtask')), - title TEXT NOT NULL, description TEXT NOT NULL DEFAULT '', status TEXT NOT NULL DEFAULT 'pending' - CHECK (status IN ('pending','in_progress','completed','blocked')), - start_at TIMESTAMPTZ, due_at TIMESTAMPTZ, sort_order INTEGER NOT NULL DEFAULT 0, - created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), - CHECK (due_at IS NULL OR start_at IS NULL OR due_at >= start_at))`, - `CREATE TABLE IF NOT EXISTS tasktree_milestones ( - id TEXT PRIMARY KEY, group_id TEXT NOT NULL REFERENCES tasktree_groups(id) ON DELETE CASCADE, - task_id TEXT REFERENCES tasktree_tasks(id) ON DELETE CASCADE, title TEXT NOT NULL, - occurs_at TIMESTAMPTZ NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT now())`, - `CREATE TABLE IF NOT EXISTS tasktree_execution_reports ( - id TEXT PRIMARY KEY, task_id TEXT NOT NULL REFERENCES tasktree_tasks(id) ON DELETE CASCADE, - title TEXT NOT NULL, body TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'succeeded', - created_at TIMESTAMPTZ NOT NULL DEFAULT now())`, - `CREATE INDEX IF NOT EXISTS idx_tasktree_tasks_group_order ON tasktree_tasks(group_id, sort_order, created_at)`, - `CREATE INDEX IF NOT EXISTS idx_tasktree_tasks_parent ON tasktree_tasks(parent_id)`, - `CREATE INDEX IF NOT EXISTS idx_tasktree_reports_task ON tasktree_execution_reports(task_id, created_at DESC)` - ,`CREATE TABLE IF NOT EXISTS tasktree_backup_state ( +type SchemaMigration = { id: string; statements: string[] }; + +const migrationLedgerStatement = `CREATE TABLE IF NOT EXISTS tasktree_schema_migrations ( + migration_id TEXT PRIMARY KEY, applied_at TIMESTAMPTZ NOT NULL DEFAULT now())`; +const schemaMigrations: SchemaMigration[] = [ + { + id: "tasktree-20260716-v1", + statements: [ + `CREATE TABLE IF NOT EXISTS tasktree_groups ( + id TEXT PRIMARY KEY, name TEXT NOT NULL, description TEXT NOT NULL DEFAULT '', + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now())`, + `CREATE TABLE IF NOT EXISTS tasktree_tasks ( + id TEXT PRIMARY KEY, group_id TEXT NOT NULL REFERENCES tasktree_groups(id) ON DELETE CASCADE, + parent_id TEXT REFERENCES tasktree_tasks(id) ON DELETE CASCADE, kind TEXT NOT NULL CHECK (kind IN ('task','subtask')), + title TEXT NOT NULL, description TEXT NOT NULL DEFAULT '', status TEXT NOT NULL DEFAULT 'pending' + CHECK (status IN ('pending','in_progress','completed','blocked')), + start_at TIMESTAMPTZ, due_at TIMESTAMPTZ, sort_order INTEGER NOT NULL DEFAULT 0, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CHECK (due_at IS NULL OR start_at IS NULL OR due_at >= start_at))`, + `CREATE TABLE IF NOT EXISTS tasktree_milestones ( + id TEXT PRIMARY KEY, group_id TEXT NOT NULL REFERENCES tasktree_groups(id) ON DELETE CASCADE, + task_id TEXT REFERENCES tasktree_tasks(id) ON DELETE CASCADE, title TEXT NOT NULL, + occurs_at TIMESTAMPTZ NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT now())`, + `CREATE TABLE IF NOT EXISTS tasktree_execution_reports ( + id TEXT PRIMARY KEY, task_id TEXT NOT NULL REFERENCES tasktree_tasks(id) ON DELETE CASCADE, + title TEXT NOT NULL, body TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'succeeded', + created_at TIMESTAMPTZ NOT NULL DEFAULT now())`, + `CREATE INDEX IF NOT EXISTS idx_tasktree_tasks_group_order ON tasktree_tasks(group_id, sort_order, created_at)`, + `CREATE INDEX IF NOT EXISTS idx_tasktree_tasks_parent ON tasktree_tasks(parent_id)`, + `CREATE INDEX IF NOT EXISTS idx_tasktree_reports_task ON tasktree_execution_reports(task_id, created_at DESC)`, + ], + }, + { id: "tasktree-20260717-v2-subsubtask", statements: [ + `ALTER TABLE tasktree_tasks DROP CONSTRAINT IF EXISTS tasktree_tasks_kind_check`, + `ALTER TABLE tasktree_tasks ADD CONSTRAINT tasktree_tasks_kind_check CHECK (kind IN ('task','subtask','subsubtask'))`, + ] }, + { id: "tasktree-20260718-v3-github-backup", statements: [ + `CREATE TABLE IF NOT EXISTS tasktree_backup_state ( target TEXT PRIMARY KEY, pending BOOLEAN NOT NULL DEFAULT false, requested_at TIMESTAMPTZ, request_reason TEXT NOT NULL DEFAULT '', last_attempt_at TIMESTAMPTZ, last_success_at TIMESTAMPTZ, content_sha256 TEXT NOT NULL DEFAULT '', git_commit TEXT NOT NULL DEFAULT '', status TEXT NOT NULL DEFAULT 'never' CHECK (status IN ('never','pending','succeeded','failed')), error_code TEXT NOT NULL DEFAULT '', error_message TEXT NOT NULL DEFAULT '', - updated_at TIMESTAMPTZ NOT NULL DEFAULT now())` - ,`ALTER TABLE tasktree_tasks DROP CONSTRAINT IF EXISTS tasktree_tasks_kind_check` - ,`ALTER TABLE tasktree_tasks ADD CONSTRAINT tasktree_tasks_kind_check CHECK (kind IN ('task','subtask','subsubtask'))` + updated_at TIMESTAMPTZ NOT NULL DEFAULT now())`, + ] }, ]; +const schemaMigrationIds = schemaMigrations.map((migration) => migration.id); +const schemaAdvisoryLockKeys = [0x48574c42, 0x54545245] as const; export class TaskTreeStore { private readonly pool: pg.Pool; private ready = false; + private schemaPromise: Promise | null = null; constructor(databaseUrl: string) { if (!databaseUrl) throw new Error("TASKTREE_DATABASE_URL or DATABASE_URL is required"); - this.pool = new Pool({ connectionString: databaseUrl, max: 4, connectionTimeoutMillis: 5000, ssl: { rejectUnauthorized: false } }); + this.pool = new Pool({ connectionString: databaseUrl, max: 4, connectionTimeoutMillis: 5000, ssl: taskTreePostgresSsl(databaseUrl) }); } async ensureSchema() { if (this.ready) return; + this.schemaPromise ??= this.migrateSchema(); + try { + await this.schemaPromise; + } catch (error) { + this.schemaPromise = null; + throw error; + } + } + + private async migrateSchema() { + if (await schemaIsCurrent(this.pool)) { + this.ready = true; + return; + } const client = await this.pool.connect(); try { await client.query("BEGIN"); - for (const statement of schema) await client.query(statement); - await client.query("INSERT INTO tasktree_schema_migrations (migration_id) VALUES ($1),($2),($3) ON CONFLICT DO NOTHING", ["tasktree-20260716-v1", "tasktree-20260717-v2-subsubtask", "tasktree-20260718-v3-github-backup"]); + await client.query("SELECT pg_advisory_xact_lock($1, $2)", [...schemaAdvisoryLockKeys]); + const ledger = await client.query("SELECT to_regclass('tasktree_schema_migrations') AS relation"); + if (ledger.rows[0]?.relation == null) await client.query(migrationLedgerStatement); + const appliedResult = await client.query("SELECT migration_id FROM tasktree_schema_migrations"); + const applied = new Set(appliedResult.rows.map((row) => String(row.migration_id))); + for (const migration of schemaMigrations) { + if (applied.has(migration.id)) continue; + try { + for (const statement of migration.statements) await client.query(statement); + await client.query("INSERT INTO tasktree_schema_migrations (migration_id) VALUES ($1)", [migration.id]); + } catch (error) { + if (error && typeof error === "object") Object.assign(error, { tasktreeMigrationId: migration.id }); + throw error; + } + } await client.query("COMMIT"); this.ready = true; } catch (error) { @@ -581,6 +623,21 @@ export function projectTimelineTasks(tasks: TaskItem[], reports: ExecutionReport }); } +async function schemaIsCurrent(pool: pg.Pool): Promise { + const relation = await pool.query("SELECT to_regclass('tasktree_schema_migrations') AS relation"); + if (relation.rows[0]?.relation == null) return false; + try { + const applied = await pool.query( + "SELECT COUNT(*)::int AS count FROM tasktree_schema_migrations WHERE migration_id = ANY($1::text[])", + [schemaMigrationIds], + ); + return Number(applied.rows[0]?.count ?? 0) === schemaMigrationIds.length; + } catch (error: any) { + if (error?.code === "42P01") return false; + throw error; + } +} + function groupRow(row: any): TaskGroup { return { id: row.id, name: row.name, description: row.description, createdAt: iso(row.created_at), updatedAt: iso(row.updated_at) }; } function overviewRow(row: any): TaskGroupOverview { return { group: groupRow(row), taskCount: Number(row.task_count), subtaskCount: Number(row.subtask_count), subsubtaskCount: Number(row.subsubtask_count), reportCount: Number(row.report_count), startAt: nullableIso(row.start_at), dueAt: nullableIso(row.due_at) }; } function taskRow(row: any): TaskItem { return { id: row.id, groupId: row.group_id, parentId: row.parent_id, kind: row.kind, title: row.title, description: row.description, status: row.status, startAt: nullableIso(row.start_at), dueAt: nullableIso(row.due_at), sortOrder: row.sort_order, createdAt: iso(row.created_at), updatedAt: iso(row.updated_at) }; } @@ -605,6 +662,13 @@ function backupStateRow(row: any): TaskTreeBackupState { function iso(value: unknown): string { return new Date(value as any).toISOString(); } function nullableIso(value: unknown): string | null { return value == null ? null : iso(value); } function domainError(code: string, message: string) { return Object.assign(new Error(message), { code }); } +function taskTreePostgresSsl(databaseUrl: string) { + try { + return new URL(databaseUrl).searchParams.get("sslmode") === "disable" ? false : { rejectUnauthorized: false }; + } catch { + return { rejectUnauthorized: false }; + } +} function ensureTimeRange(startAt: string | null, dueAt: string | null) { if (startAt !== null && dueAt !== null && new Date(startAt).valueOf() > new Date(dueAt).valueOf()) { throw domainError("invalid_time_range", "startAt must be before or equal to dueAt"); diff --git a/package.json b/package.json index f1696a95..00f261b8 100644 --- a/package.json +++ b/package.json @@ -18,6 +18,7 @@ "harnessrl:native:l1-smoke": "node scripts/harnessrl-native-l1-smoke.mjs", "tasktree:backup:l1-smoke": "bun scripts/tasktree-backup-native-l1-smoke.mjs", "tasktree:backup:restore:l1-smoke": "bun scripts/tasktree-backup-restore-l1-smoke.mjs", + "tasktree:schema:concurrency-smoke": "bun scripts/tasktree-schema-concurrency-smoke.mjs", "workbench:api:dev": "bun --watch cmd/hwlab-workbench-api/main.ts", "workbench:worker:dev": "bun --watch cmd/hwlab-workbench-worker/main.ts", "workbench:web:dev": "cd web/hwlab-cloud-web && WORKBENCH_VITE_USE_POLLING=1 bun run dev", diff --git a/scripts/tasktree-schema-concurrency-smoke.mjs b/scripts/tasktree-schema-concurrency-smoke.mjs new file mode 100644 index 00000000..aaab4d34 --- /dev/null +++ b/scripts/tasktree-schema-concurrency-smoke.mjs @@ -0,0 +1,151 @@ +import assert from "node:assert/strict"; +import { spawn } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { fileURLToPath } from "node:url"; + +import pg from "pg"; + +const { Pool } = pg; +const clients = positiveInteger(process.env.TASKTREE_MIGRATION_SMOKE_CLIENTS, 8); +const rounds = positiveInteger(process.env.TASKTREE_MIGRATION_SMOKE_ROUNDS, 3); +const adminUrl = process.env.TASKTREE_MIGRATION_SMOKE_DATABASE_URL + || "postgresql:///postgres?host=/var/run/postgresql&user=root"; +const databaseName = `tasktree_smoke_${randomUUID().replaceAll("-", "")}`; +const cliPath = fileURLToPath(new URL("../tools/hwlab-cli/bin/hwlab-cli.ts", import.meta.url)); +const scopedUrl = new URL(adminUrl); +scopedUrl.pathname = `/${databaseName}`; +scopedUrl.searchParams.set("sslmode", "disable"); +scopedUrl.searchParams.set("options", "-clock_timeout=1500ms"); +const databaseUrl = scopedUrl.toString(); +const admin = new Pool({ connectionString: adminUrl, max: 2 }); +const scoped = new Pool({ connectionString: databaseUrl, max: 4 }); + +try { + await admin.query(`CREATE DATABASE ${databaseName}`); + + const fresh = await Promise.all(Array.from({ length: clients }, () => runCli(["group", "list"]))); + fresh.forEach((result) => assertEnvelope(result, "group.list")); + await assertMigrations(); + + const createdGroup = await runCli(["group", "create", "--name", `Concurrency smoke ${Date.now()}`]); + assertEnvelope(createdGroup, "group.create"); + const groupId = createdGroup.data.id; + const createdTask = await runCli(["task", "create", "--group", groupId, "--title", "Read envelope seed"]); + assertEnvelope(createdTask, "task.create"); + const taskId = createdTask.data.id; + const report = await runCli(["report", "write", "--task", taskId, "--title", "Smoke", "--body", "passed"]); + assertEnvelope(report, "report.write"); + assert.equal(report.data.mutation, true); + + await scoped.query( + "DELETE FROM tasktree_schema_migrations WHERE migration_id = ANY($1::text[])", + [["tasktree-20260717-v2-subsubtask", "tasktree-20260718-v3-github-backup"]], + ); + await scoped.query("ALTER TABLE tasktree_tasks DROP CONSTRAINT tasktree_tasks_kind_check"); + await scoped.query( + "INSERT INTO tasktree_tasks (id,group_id,kind,title) VALUES ($1,$2,$3,$4)", + ["tt_invalid_migration_smoke", groupId, "invalid", "Migration rollback sentinel"], + ); + const failedMigration = await runCli(["group", "list"], true); + assert.equal(failedMigration.ok, false); + assert.equal(failedMigration.error.code, "23514"); + assert.deepEqual(failedMigration.error.detail, { + migrationId: "tasktree-20260717-v2-subsubtask", + sqlState: "23514", + }); + const afterFailure = await scoped.query("SELECT migration_id FROM tasktree_schema_migrations ORDER BY migration_id"); + assert.deepEqual(afterFailure.rows.map((row) => row.migration_id), ["tasktree-20260716-v1"]); + + await scoped.query("DELETE FROM tasktree_tasks WHERE id=$1", ["tt_invalid_migration_smoke"]); + assertEnvelope(await runCli(["group", "list"]), "group.list"); + await assertMigrations(); + + const blocker = await scoped.connect(); + try { + await blocker.query("BEGIN"); + await blocker.query("SELECT id FROM tasktree_tasks LIMIT 1"); + const commands = [ + ["group", "list"], + ["task", "list", "--group", groupId], + ["report", "list", "--task", taskId], + ]; + for (let round = 0; round < rounds; round += 1) { + const results = await Promise.all(Array.from({ length: clients }, (_, index) => runCli(commands[index % commands.length]))); + results.forEach((result, index) => assertEnvelope(result, ["group.list", "task.list", "report.list"][index % commands.length])); + } + await blocker.query("COMMIT"); + } catch (error) { + await blocker.query("ROLLBACK").catch(() => {}); + throw error; + } finally { + blocker.release(); + } + + console.log(JSON.stringify({ + ok: true, + operation: "tasktree.schema-concurrency-smoke", + data: { + clients, + rounds, + freshCliEnvelopes: fresh.length, + existingCliEnvelopes: clients * rounds, + migrationRollbackVerified: true, + existingReadsWithTableLockVerified: true, + migrationIds: await migrationIds(), + }, + }, null, 2)); +} finally { + await scoped.end().catch(() => {}); + await admin.query(`DROP DATABASE IF EXISTS ${databaseName} WITH (FORCE)`).catch(() => {}); + await admin.end().catch(() => {}); +} + +async function runCli(args, allowFailure = false) { + const child = spawn(process.execPath, [cliPath, "tasktree", ...args], { + cwd: fileURLToPath(new URL("..", import.meta.url)), + env: { ...process.env, TASKTREE_DATABASE_URL: databaseUrl, DATABASE_URL: "" }, + stdio: ["ignore", "pipe", "pipe"], + }); + let stdout = ""; + let stderr = ""; + child.stdout.setEncoding("utf8").on("data", (chunk) => { stdout += chunk; }); + child.stderr.setEncoding("utf8").on("data", (chunk) => { stderr += chunk; }); + const exitCode = await new Promise((resolve, reject) => { + child.once("error", reject); + child.once("close", resolve); + }); + assert.equal(stderr.trim(), "", `CLI stderr must stay empty: ${stderr}`); + assert.ok(stdout.trim(), "CLI must return one JSON envelope on stdout"); + const result = JSON.parse(stdout); + if (!allowFailure) assert.equal(exitCode, 0, JSON.stringify(result)); + return result; +} + +function assertEnvelope(result, operation) { + assert.equal(result.ok, true, JSON.stringify(result)); + assert.equal(result.operation, operation); + assert.ok(Object.hasOwn(result, "data"), "successful CLI envelope must contain data"); +} + +async function assertMigrations() { + assert.deepEqual(await migrationIds(), [ + "tasktree-20260716-v1", + "tasktree-20260717-v2-subsubtask", + "tasktree-20260718-v3-github-backup", + ]); + const constraint = await scoped.query( + "SELECT COUNT(*)::int AS count FROM pg_constraint WHERE conrelid='tasktree_tasks'::regclass AND conname='tasktree_tasks_kind_check'", + ); + assert.equal(constraint.rows[0].count, 1); +} + +async function migrationIds() { + const result = await scoped.query("SELECT migration_id FROM tasktree_schema_migrations ORDER BY migration_id"); + return result.rows.map((row) => row.migration_id); +} + +function positiveInteger(value, fallback) { + const parsed = Number(value ?? fallback); + assert.ok(Number.isInteger(parsed) && parsed > 0, "smoke concurrency parameters must be positive integers"); + return parsed; +}