mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: resequence existing workbench trace migration

Browse Source
This commit is contained in:
lyon
2026-06-20 20:37:08 +08:00
parent 971922be47
commit 183cf653f4
4 changed files with 132 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/db/migrations/0001_cloud_core_skeleton.sql
+56 -12
View File
@@ -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,
internal/db/schema.test.ts
+6 -4
View File
@@ -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", () => {
scripts/src/dev-runtime-migration.mjs
+35 -5
View File
@@ -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")) {
scripts/src/dev-runtime-migration.test.mjs
+35 -1
View File
@@ -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: [] };