From 183cf653f43536af580eb63021c476fad5e7cdfb Mon Sep 17 00:00:00 2001 From: lyon Date: Sat, 20 Jun 2026 20:37:08 +0800 Subject: [PATCH] fix: resequence existing workbench trace migration --- .../migrations/0001_cloud_core_skeleton.sql | 68 +++++++++++++++---- internal/db/schema.test.ts | 10 +-- scripts/src/dev-runtime-migration.mjs | 40 +++++++++-- scripts/src/dev-runtime-migration.test.mjs | 36 +++++++++- 4 files changed, 132 insertions(+), 22 deletions(-) diff --git a/internal/db/migrations/0001_cloud_core_skeleton.sql b/internal/db/migrations/0001_cloud_core_skeleton.sql index d56be2d0..75bcf7d8 100644 --- a/internal/db/migrations/0001_cloud_core_skeleton.sql +++ b/internal/db/migrations/0001_cloud_core_skeleton.sql @@ -294,27 +294,71 @@ CREATE TABLE IF NOT EXISTS workbench_trace_events ( ); CREATE INDEX IF NOT EXISTS idx_workbench_trace_events_trace_seq ON workbench_trace_events(trace_id, source_seq, projected_seq, id); CREATE INDEX IF NOT EXISTS idx_workbench_trace_events_session_seq ON workbench_trace_events(session_id, source_seq, projected_seq, id); -WITH traces_needing_resequence AS ( + +WITH duplicate_trace_source_events AS ( + SELECT + id, + ROW_NUMBER() OVER ( + PARTITION BY trace_id, source_event_id + ORDER BY occurred_at, id + ) AS duplicate_rank + FROM workbench_trace_events + WHERE trace_id IS NOT NULL AND source_event_id IS NOT NULL +), trace_source_event_rewrites AS ( + SELECT + id, + 'workbench_trace_events:' || id AS stable_source_event_id + FROM duplicate_trace_source_events + WHERE duplicate_rank > 1 +) +UPDATE workbench_trace_events AS target +SET + source_event_id = trace_source_event_rewrites.stable_source_event_id, + updated_at = COALESCE(NULLIF(target.updated_at, ''), target.occurred_at, CURRENT_TIMESTAMP::text) +FROM trace_source_event_rewrites +WHERE target.id = trace_source_event_rewrites.id + AND target.source_event_id IS DISTINCT FROM trace_source_event_rewrites.stable_source_event_id; + +WITH trace_event_projected_seq_conflicts AS ( SELECT trace_id FROM workbench_trace_events WHERE trace_id IS NOT NULL GROUP BY trace_id HAVING COUNT(*) <> COUNT(DISTINCT projected_seq) -), ranked_trace_events AS ( +), trace_event_resequence AS ( SELECT workbench_trace_events.id, - (ROW_NUMBER() OVER (PARTITION BY workbench_trace_events.trace_id ORDER BY workbench_trace_events.occurred_at, workbench_trace_events.id))::integer AS durable_projection_seq + (ROW_NUMBER() OVER ( + PARTITION BY workbench_trace_events.trace_id + ORDER BY + CASE WHEN workbench_trace_events.projected_seq > 0 THEN 0 ELSE 1 END, + workbench_trace_events.projected_seq, + workbench_trace_events.occurred_at, + workbench_trace_events.id + ))::integer AS durable_projected_seq FROM workbench_trace_events - INNER JOIN traces_needing_resequence - ON traces_needing_resequence.trace_id = workbench_trace_events.trace_id + INNER JOIN trace_event_projected_seq_conflicts + ON trace_event_projected_seq_conflicts.trace_id = workbench_trace_events.trace_id ) -UPDATE workbench_trace_events AS existing_trace_events -SET projected_seq = ranked_trace_events.durable_projection_seq -FROM ranked_trace_events -WHERE existing_trace_events.id = ranked_trace_events.id - AND existing_trace_events.projected_seq IS DISTINCT FROM ranked_trace_events.durable_projection_seq; -CREATE UNIQUE INDEX IF NOT EXISTS idx_workbench_trace_events_trace_source_event ON workbench_trace_events(trace_id, source_event_id) WHERE source_event_id IS NOT NULL; -CREATE UNIQUE INDEX IF NOT EXISTS idx_workbench_trace_events_trace_projected_seq ON workbench_trace_events(trace_id, projected_seq); +UPDATE workbench_trace_events AS target +SET + projected_seq = trace_event_resequence.durable_projected_seq, + source_seq = CASE + WHEN target.source_seq > 0 THEN target.source_seq + ELSE trace_event_resequence.durable_projected_seq + END, + updated_at = COALESCE(NULLIF(target.updated_at, ''), target.occurred_at, CURRENT_TIMESTAMP::text) +FROM trace_event_resequence +WHERE target.id = trace_event_resequence.id + AND ( + target.projected_seq IS DISTINCT FROM trace_event_resequence.durable_projected_seq + OR target.source_seq <= 0 + ); + +DROP INDEX IF EXISTS idx_workbench_trace_events_trace_source_event; +CREATE UNIQUE INDEX idx_workbench_trace_events_trace_source_event ON workbench_trace_events(trace_id, source_event_id) WHERE source_event_id IS NOT NULL; +DROP INDEX IF EXISTS idx_workbench_trace_events_trace_projected_seq; +CREATE UNIQUE INDEX idx_workbench_trace_events_trace_projected_seq ON workbench_trace_events(trace_id, projected_seq); CREATE TABLE IF NOT EXISTS workbench_projection_checkpoints ( trace_id TEXT PRIMARY KEY, diff --git a/internal/db/schema.test.ts b/internal/db/schema.test.ts index d6d176c3..44827f0f 100644 --- a/internal/db/schema.test.ts +++ b/internal/db/schema.test.ts @@ -84,10 +84,12 @@ test("initial migration declares Workbench fact backfill sources", async () => { assert.ok(traceBackfillMatch, "missing workbench_trace_events idempotent backfill"); assert.match(traceBackfillMatch[0], /ROW_NUMBER\(\) OVER \(PARTITION BY trace_id ORDER BY occurred_at, id\)/u); assert.match(traceBackfillMatch[0], /durable_projection_seq/u); - const traceResequenceMatch = sql.match(/WITH traces_needing_resequence AS \([\s\S]*?CREATE UNIQUE INDEX IF NOT EXISTS idx_workbench_trace_events_trace_projected_seq/u); - assert.ok(traceResequenceMatch, "missing workbench_trace_events projected_seq resequence before unique index"); - assert.match(traceResequenceMatch[0], /COUNT\(\*\) <> COUNT\(DISTINCT projected_seq\)/u); - assert.match(traceResequenceMatch[0], /UPDATE workbench_trace_events AS existing_trace_events/u); + assert.match(sql, /duplicate_trace_source_events/u); + assert.match(sql, /workbench_trace_events:/u); + assert.match(sql, /trace_event_projected_seq_conflicts/u); + assert.match(sql, /COUNT\(\*\) <> COUNT\(DISTINCT projected_seq\)/u); + assert.match(sql, /DROP INDEX IF EXISTS idx_workbench_trace_events_trace_projected_seq/u); + assert.match(sql, /CREATE UNIQUE INDEX idx_workbench_trace_events_trace_projected_seq/u); }); test("protocol record guards catch schema drift before runtime writes", () => { diff --git a/scripts/src/dev-runtime-migration.mjs b/scripts/src/dev-runtime-migration.mjs index edd6532f..b3b11958 100644 --- a/scripts/src/dev-runtime-migration.mjs +++ b/scripts/src/dev-runtime-migration.mjs @@ -299,11 +299,7 @@ async function applyMigration(report, sql, env, options) { } catch (error) { await rollbackQuietly(client); const classified = classifyMigrationError(error); - addBlocker(report, "runtime_blocker", blockerScope(classified.blocker), classified.summary, { - blocker: classified.blocker, - errorCode: classified.errorCode, - secretValuesPrinted: false - }); + addBlocker(report, "runtime_blocker", blockerScope(classified.blocker), classified.summary, migrationErrorEvidence(classified)); } finally { await closeClient(); } @@ -586,8 +582,10 @@ function blockerScope(blocker) { function classifyMigrationError(error) { const errorCode = typeof error?.code === "string" ? error.code : "UNKNOWN"; + const details = migrationErrorDetails(error); if (errorCode === "HWLAB_PG_DRIVER_MISSING") { return { + ...details, blocker: RUNTIME_DURABLE_ADAPTER_DRIVER_MISSING, errorCode, summary: "Postgres driver is unavailable for DEV runtime migration apply." @@ -595,6 +593,7 @@ function classifyMigrationError(error) { } if (isMigrationSslError(error)) { return { + ...details, blocker: RUNTIME_DURABLE_ADAPTER_SSL_BLOCKED, errorCode, summary: "DEV runtime migration reached Postgres, but SSL negotiation or SSL mode compatibility blocked apply." @@ -602,6 +601,7 @@ function classifyMigrationError(error) { } if (["28P01", "28000", "42501"].includes(errorCode)) { return { + ...details, blocker: RUNTIME_DURABLE_ADAPTER_AUTH_BLOCKED, errorCode, summary: "DEV runtime migration reached Postgres but auth or authorization blocked apply." @@ -609,18 +609,48 @@ function classifyMigrationError(error) { } if (["42P01", "42703", "3F000"].includes(errorCode)) { return { + ...details, blocker: RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED, errorCode, summary: "DEV runtime migration apply was blocked by schema/search_path state." }; } return { + ...details, blocker: RUNTIME_DURABLE_ADAPTER_QUERY_BLOCKED, errorCode, summary: "DEV runtime migration apply could not complete." }; } +function migrationErrorEvidence(classified) { + return { + blocker: classified.blocker, + errorCode: classified.errorCode, + ...(classified.errorConstraint ? { errorConstraint: classified.errorConstraint } : {}), + ...(classified.errorTable ? { errorTable: classified.errorTable } : {}), + ...(classified.errorMessage ? { errorMessage: classified.errorMessage } : {}), + ...(classified.errorDetail ? { errorDetail: classified.errorDetail } : {}), + secretValuesPrinted: false + }; +} + +function migrationErrorDetails(error) { + return { + errorMessage: safeDiagnosticText(error?.message, 500), + errorDetail: safeDiagnosticText(error?.detail, 500), + errorConstraint: safeDiagnosticText(error?.constraint, 160), + errorTable: safeDiagnosticText(error?.table, 160) + }; +} + +function safeDiagnosticText(value, maxLength) { + if (typeof value !== "string") return null; + const redacted = redactFailureText(value).trim(); + if (!redacted) return null; + return redacted.length > maxLength ? `${redacted.slice(0, maxLength)}...` : redacted; +} + function isMigrationSslError(error) { const errorCode = typeof error?.code === "string" ? error.code : ""; if (errorCode.startsWith("ERR_SSL") || errorCode.startsWith("ERR_TLS")) { diff --git a/scripts/src/dev-runtime-migration.test.mjs b/scripts/src/dev-runtime-migration.test.mjs index 73db813f..fc81c8af 100644 --- a/scripts/src/dev-runtime-migration.test.mjs +++ b/scripts/src/dev-runtime-migration.test.mjs @@ -314,6 +314,37 @@ test("apply mode records ledger and verifies durable runtime readiness", async ( assert.equal(JSON.stringify(report).includes("redacted@example.invalid"), false); }); +test("apply mode reports sanitized migration diagnostics", async () => { + const migrationApplyError = new Error("duplicate key blocked postgresql://hwlab:secret-password@db.example.test:5432/hwlab"); + migrationApplyError.code = "23505"; + migrationApplyError.constraint = "idx_workbench_trace_events_trace_source_event"; + migrationApplyError.table = "workbench_trace_events"; + migrationApplyError.detail = "Key (trace_id, source_event_id)=(trc_1, password=secret) already exists."; + const queryClient = createFakeQueryClient({ migrationReady: false, migrationApplyError }); + + const report = await buildDevRuntimeMigrationReport( + parseArgs(["--apply", "--confirm-dev", "--confirmed-non-production"]), + { + env: { + HWLAB_CLOUD_DB_URL: "postgresql://hwlab:redacted@example.invalid:5432/hwlab", + HWLAB_CLOUD_DB_SSL_MODE: "disable" + }, + queryClient, + now: () => "2026-05-22T00:00:00.000Z" + } + ); + + assert.equal(report.conclusion.status, "blocked"); + assert.equal(report.actions.migrationApplied, false); + assert.equal(report.blockers[0].evidence.errorCode, "23505"); + assert.equal(report.blockers[0].evidence.errorConstraint, "idx_workbench_trace_events_trace_source_event"); + assert.equal(report.blockers[0].evidence.errorTable, "workbench_trace_events"); + assert.match(report.blockers[0].evidence.errorMessage, /\[redacted-postgres-url\]/u); + assert.match(report.blockers[0].evidence.errorDetail, /password=\[redacted\]/u); + assert.equal(JSON.stringify(report.blockers[0].evidence).includes("secret-password"), false); + assert.ok(queryClient.calls.some((call) => call.sql === "ROLLBACK")); +}); + test("live dry-run closes pg pool after readiness verification", async () => { const backingClient = createFakeQueryClient({ migrationReady: true }); const pools = []; @@ -388,7 +419,7 @@ test("apply mode uses a dedicated pg client transaction", async () => { assert.ok(pools[1].calls.some((call) => call.type === "end")); }); -function createFakeQueryClient({ migrationReady, countErrorCode = null }) { +function createFakeQueryClient({ migrationReady, countErrorCode = null, migrationApplyError = null }) { const state = { migrationReady, calls: [] @@ -402,6 +433,9 @@ function createFakeQueryClient({ migrationReady, countErrorCode = null }) { if (sql === "BEGIN" || sql === "COMMIT" || sql === "ROLLBACK") { return { rows: [] }; } + if (migrationApplyError && sql.includes("CREATE TABLE IF NOT EXISTS projects")) { + throw migrationApplyError; + } if (sql.includes("INSERT INTO hwlab_schema_migrations")) { state.migrationReady = true; return { rows: [] };