# Files changed · PR \#23159 · pingdotgg/t3code [View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/23159/diffs) Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code) Visibility: public [fix\(server\): migrate-dev-db refuses unimported legacy threads](https://git.cafe/pingdotgg/t3code/pull/23159.md) Version: 11 Head commit: 5de88300e7d8271031784359714ad6c23afdc2dd Base commit: bb7783e295506ad423f527ed28a019836e331e8c Comparison base: bb7783e295506ad423f527ed28a019836e331e8c Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing. ## apps/server/scripts/migrate-dev-db.test.ts Status: modified; +124; −4 ```diff @@ -9,9 +9,9 @@ import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; import { runMigrateDevDb } from "./migrate-dev-db.ts"; -const withDatabase = ( +const withDatabase = ( databasePath: string, - effect: Effect.Effect, + effect: Effect.Effect, ) => effect.pipe(Effect.provide(NodeSqliteClient.layer({ filename: databasePath }))); const createFixtureSource = Effect.fn("createMigrateDevDbFixtureSource")(function* ( @@ -69,9 +69,14 @@ (run_id, thread_id, ordinal, provider, status, requested_at, payload_json) VALUES (${`run-${threadId}`}, ${threadId}, 1, 'codex', ${runStatus}, '2026-08-01', '{}')`; yield* sql`INSERT INTO orchestration_events - (event_id, aggregate_kind, stream_id, stream_version, event_type, occurred_at, actor_kind, payload_json, metadata_json) - VALUES (${`event-${threadId}`}, 'thread', ${threadId}, 0, 'thread.created', '2026-08-01', 'user', '{}', '{}')`; + (event_id, aggregate_kind, stream_id, stream_version, event_type, occurred_at, actor_kind, payload_json, metadata_json, application_event_version) + VALUES (${`event-${threadId}`}, 'thread', ${threadId}, 0, 'thread.created', '2026-08-01', 'user', '{}', '{}', 2)`; } + // A V1 thread the server already imported must not trip the legacy guard. + yield* sql`INSERT INTO projection_threads + (thread_id, project_id, title, model_selection_json, runtime_mode, created_at, updated_at) + VALUES ('stopped-thread', 'project-kept', 'stopped-thread', + '{"instanceId":"codex","model":"gpt-5.4"}', 'full-access', '2026-08-01', '2026-08-01')`; yield* sql`INSERT INTO orchestration_v2_projection_provider_sessions (provider_session_id, thread_id, provider, status, updated_at, payload_json) VALUES ('session-shared', 'running-thread', 'codex', 'stopped', '2026-08-01', '{}')`; @@ -228,6 +233,121 @@ }), ); + // The server imports V1 threads at boot, not in a migration, so a source can + // reach the V2 schema with its legacy threads still unimported. + it.effect.each([ + { name: "a pre-V2 source", migrateToLatest: false }, + { name: "a V2 source whose legacy threads were never imported", migrateToLatest: true }, + ])("refuses $name", ({ migrateToLatest }) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const sourceDir = yield* fs.makeTempDirectoryScoped({ prefix: "migrate-dev-db-v1-data-" }); + const destDir = yield* fs.makeTempDirectoryScoped({ + prefix: "migrate-dev-db-v1-data-dest-", + }); + const stateDir = path.join(sourceDir, "userdata"); + const source = path.join(stateDir, "statev2.sqlite"); + yield* fs.makeDirectory(stateDir, { recursive: true }); + yield* withDatabase( + source, + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* runMigrations({ toMigrationInclusive: 54 }); + yield* sql`INSERT INTO projection_projects + (project_id, title, workspace_root, scripts_json, created_at, updated_at) + VALUES ('project-legacy', 'Legacy', '/tmp/legacy', '[]', '2026-08-01', '2026-08-01')`; + yield* sql`INSERT INTO projection_threads + (thread_id, project_id, title, model_selection_json, runtime_mode, created_at, updated_at) + VALUES ('thread-legacy', 'project-legacy', 'Legacy thread', + '{"instanceId":"codex","model":"gpt-5.4"}', 'full-access', '2026-08-01', '2026-08-01')`; + if (migrateToLatest) yield* runMigrations(); + }), + ); + + // A live destination connection keeps uncheckpointed rows in the WAL; + // the rejected run must leave those bytes alone. + const destination = yield* createFixtureSource(destDir); + yield* withDatabase( + destination, + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql.unsafe("PRAGMA journal_mode = WAL").unprepared; + yield* sql`UPDATE projection_projects SET title = 'Edited' WHERE project_id = 'project-kept'`; + const walBefore = yield* fs.readFile(`${destination}-wal`); + assert.isAbove(walBefore.length, 0); + + const error = yield* runMigrateDevDb( + { baseDir: destDir, source, projects: 5, threadsPerProject: 10 }, + { sharedHome: sourceDir }, + ).pipe(Effect.flip); + assert.equal(error._tag, "MigrateDevDbLegacySourceError"); + assert.deepStrictEqual(yield* fs.readFile(`${destination}-wal`), walBefore); + }), + ); + }), + ); + + // An old V1 server on an empty pre-V2 source can create its first thread + // after the guard passes; the unfiltered copy would take it and prune drop it. + it.effect("refuses a legacy thread committed after the guard", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const sourceDir = yield* fs.makeTempDirectoryScoped({ prefix: "migrate-dev-db-race-" }); + const destDir = yield* fs.makeTempDirectoryScoped({ prefix: "migrate-dev-db-race-dest-" }); + const source = path.join(sourceDir, "userdata", "statev2.sqlite"); + yield* fs.makeDirectory(path.dirname(source), { recursive: true }); + yield* withDatabase(source, runMigrations({ toMigrationInclusive: 54 })); + const destination = yield* createFixtureSource(destDir); + + // The first filesystem call after the guard is creating the destination + // state dir; the V1 server commits its thread there. + const commitLegacyThread = withDatabase( + source, + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql`INSERT INTO projection_projects + (project_id, title, workspace_root, scripts_json, created_at, updated_at) + VALUES ('project-legacy', 'Legacy', '/tmp/legacy', '[]', '2026-08-01', '2026-08-01')`; + yield* sql`INSERT INTO projection_threads + (thread_id, project_id, title, model_selection_json, runtime_mode, created_at, updated_at) + VALUES ('thread-legacy', 'project-legacy', 'Legacy thread', + '{"instanceId":"codex","model":"gpt-5.4"}', 'full-access', '2026-08-01', '2026-08-01')`; + }), + ); + const racingFs: FileSystem.FileSystem = { + ...fs, + makeDirectory: (directory, options) => + directory === path.dirname(destination) + ? commitLegacyThread.pipe( + Effect.orDie, + Effect.andThen(fs.makeDirectory(directory, options)), + ) + : fs.makeDirectory(directory, options), + }; + + yield* withDatabase( + destination, + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql.unsafe("PRAGMA journal_mode = WAL").unprepared; + yield* sql`UPDATE projection_projects SET title = 'Edited' WHERE project_id = 'project-kept'`; + const databaseBefore = yield* fs.readFile(destination); + const walBefore = yield* fs.readFile(`${destination}-wal`); + + const error = yield* runMigrateDevDb( + { baseDir: destDir, source, projects: 5, threadsPerProject: 10 }, + { sharedHome: sourceDir }, + ).pipe(Effect.provideService(FileSystem.FileSystem, racingFs), Effect.flip); + assert.equal(error._tag, "MigrateDevDbLegacySourceError"); + assert.deepStrictEqual(yield* fs.readFile(destination), databaseBefore); + assert.deepStrictEqual(yield* fs.readFile(`${destination}-wal`), walBefore); + }), + ); + }), + ); + it.effect("refuses while a dev server holds the destination", () => Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; ``` ## apps/server/scripts/migrate-dev-db.ts Status: modified; +70; −21 ```diff @@ -60,6 +60,18 @@ } } +export class MigrateDevDbLegacySourceError extends Schema.TaggedError()( + "MigrateDevDbLegacySourceError", + { + sourcePath: Schema.String, + legacyThreads: Schema.Number, + }, +) { + override get message(): string { + return `Source database '${this.sourcePath}' holds ${this.legacyThreads} legacy threads the server has not imported into V2 yet, which this command cannot carry over. Start a server once on a T3 home holding a copy of it so it imports them, then pass the userdata/statev2.sqlite that server leaves behind as --source.`; + } +} + export class MigrateDevDbServerRunningError extends Schema.TaggedError()( "MigrateDevDbServerRunningError", { @@ -227,32 +239,62 @@ /** * One transaction reads the source, so the copy is consistent while the real server writes. + * + * The server imports V1 threads into the V2 tables at runtime, not in a + * migration, and the copy keeps only threads that already have V2 rows, so a + * source with unimported threads would migrate cleanly and silently lose them. + * The check reads the same snapshot as the copy: a V1 server can commit its + * first thread at any moment before it. Mirrors the first half of + * LegacyV1ThreadImporter's pendingThreadCount; the other half (lazy transcript + * imports) survives the copy. */ const copySourceSlice = Effect.fn("copyDevDbSourceSlice")(function* (sourcePath: string) { const sql = yield* SqlClient.SqlClient; yield* sql`ATTACH DATABASE ${`${NodeURL.pathToFileURL(sourcePath).href}?mode=ro`} AS src`; - const schema = yield* sql<{ type: string; name: string; sql: string }>` - SELECT type, name, sql FROM src.sqlite_master - WHERE sql IS NOT NULL AND name NOT LIKE 'sqlite_%'`; - const threadColumns = yield* sql<{ name: string; notnull: number }>` - SELECT m.name, c."notnull" FROM src.sqlite_master m, pragma_table_info(m.name, 'src') c - WHERE m.type = 'table' AND c.name = 'thread_id'`; - const threadIdNullable = new Map(threadColumns.map((row) => [row.name, row.notnull === 0])); - // FTS5 makes its own shadow tables and contentless rows cannot be copied: - // create the virtual table, skip its shadow tables and rows, and rebuild the - // index once the messages and triggers are in. - const virtualTables = schema - .filter((entry) => entry.type === "table" && /^CREATE VIRTUAL TABLE/i.test(entry.sql)) - .map((entry) => entry.name); - const isShadowTable = (name: string) => - virtualTables.some((virtual) => - FTS5_SHADOW_SUFFIXES.some((suffix) => name === `${virtual}${suffix}`), - ); - const tables = schema.filter((entry) => entry.type === "table" && !isShadowTable(entry.name)); - const filtered = tables.some((table) => table.name === "orchestration_v2_projection_threads"); yield* sql.withTransaction( Effect.gen(function* () { + const schema = yield* sql<{ type: string; name: string; sql: string }>` + SELECT type, name, sql FROM src.sqlite_master + WHERE sql IS NOT NULL AND name NOT LIKE 'sqlite_%'`; + const threadColumns = yield* sql<{ name: string; notnull: number }>` + SELECT m.name, c."notnull" FROM src.sqlite_master m, pragma_table_info(m.name, 'src') c + WHERE m.type = 'table' AND c.name = 'thread_id'`; + const threadIdNullable = new Map(threadColumns.map((row) => [row.name, row.notnull === 0])); + // FTS5 makes its own shadow tables and contentless rows cannot be copied: + // create the virtual table, skip its shadow tables and rows, and rebuild the + // index once the messages and triggers are in. + const virtualTables = schema + .filter((entry) => entry.type === "table" && /^CREATE VIRTUAL TABLE/i.test(entry.sql)) + .map((entry) => entry.name); + const isShadowTable = (name: string) => + virtualTables.some((virtual) => + FTS5_SHADOW_SUFFIXES.some((suffix) => name === `${virtual}${suffix}`), + ); + const tables = schema.filter((entry) => entry.type === "table" && !isShadowTable(entry.name)); + const filtered = tables.some((table) => table.name === "orchestration_v2_projection_threads"); + + if (tables.some((table) => table.name === "projection_threads")) { + // The V2 tables, application_event_version, and the partial index arrive in + // the same migration. Without the hint SQLite scans each thread's stream. + const [pending] = filtered + ? yield* sql<{ count: number }>` + SELECT COUNT(*) AS count FROM src.projection_threads AS thread + WHERE NOT EXISTS ( + SELECT 1 + FROM src.orchestration_events AS event INDEXED BY orchestration_events_v2_created_threads_idx + WHERE event.application_event_version = 2 + AND event.aggregate_kind = 'thread' + AND event.stream_id = thread.thread_id + AND event.event_type = 'thread.created' + )` + : yield* sql<{ count: number }>`SELECT COUNT(*) AS count FROM src.projection_threads`; + const legacyThreads = Number(pending?.count ?? 0); + if (legacyThreads > 0) { + return yield* new MigrateDevDbLegacySourceError({ sourcePath, legacyThreads }); + } + } + for (const table of tables) yield* sql.unsafe(table.sql).unprepared; if (filtered) { yield* sql`CREATE TEMP TABLE copied_threads (thread_id TEXT PRIMARY KEY)`; @@ -477,7 +519,6 @@ } yield* fs.makeDirectory(stateDir, { recursive: true }); - yield* ensureNotInUse(databasePath); const wrapPhase = (phase: MigrateDevDbPhaseError["phase"], phaseDatabasePath: string) => @@ -493,8 +534,16 @@ yield* Console.log(`Copying visible threads from ${sourcePath} (read-only)...`); yield* copySourceSlice(sourcePath).pipe( Effect.provide(NodeSqliteClient.layer({ filename: snapshotPath })), + Effect.catchTags({ + SqlError: (cause) => + Effect.fail( + new MigrateDevDbPhaseError({ phase: "snapshot", databasePath: sourcePath, cause }), + ), - wrapPhase("snapshot", sourcePath), + }), ); + // After the copy, whose legacy check must refuse before this checkpoint + // rewrites the destination. + yield* ensureNotInUse(databasePath); // Migrate before pruning: a source older than this checkout would // otherwise crash the prune queries on columns that don't exist yet. ```