[codex] fix sqlite migration journal recovery (#400)

* fix sqlite migration journal recovery

* address sqlite migration review follow-ups

src/server/db/migrate.test.ts

@@ -50,6 +50,8 @@ function recordAppliedMigrations(
   }
 }
 
+const STALE_JOURNAL_TIMESTAMP_DRIFT_MS = 4_196_930;
+
 describe('sqlite migrate bootstrap', () => {
   afterEach(() => {
     delete process.env.DATA_DIR;
@@ -208,6 +210,43 @@ describe('sqlite migrate bootstrap', () => {
     sqlite.close();
   });
 
+  it('updates only the latest matching migration record when reconciling stale timestamps', async () => {
+    const dataDir = mkdtempSync(join(tmpdir(), 'metapi-migrate-rowid-reconcile-'));
+    process.env.DATA_DIR = dataDir;
+    vi.resetModules();
+
+    const migrateModule = await import('./migrate.js');
+    const { __migrateTestUtils } = migrateModule;
+
+    const sqlite = new Database(':memory:');
+    sqlite.exec(`
+      CREATE TABLE "__drizzle_migrations" (
+        id SERIAL PRIMARY KEY,
+        hash text NOT NULL,
+        created_at numeric
+      );
+    `);
+    sqlite.prepare('INSERT INTO "__drizzle_migrations" ("hash", "created_at") VALUES (?, ?)').run('same-hash', 10);
+    sqlite.prepare('INSERT INTO "__drizzle_migrations" ("hash", "created_at") VALUES (?, ?)').run('same-hash', 20);
+
+    const changed = __migrateTestUtils.markMigrationRecordIfMissing(sqlite, {
+      hash: 'same-hash',
+      createdAt: 30,
+    });
+
+    const records = sqlite
+      .prepare('SELECT rowid, hash, created_at FROM "__drizzle_migrations" ORDER BY rowid ASC')
+      .all() as Array<{ rowid: number; hash: string; created_at: number }>;
+
+    expect(changed).toBe(true);
+    expect(records).toEqual([
+      { rowid: 1, hash: 'same-hash', created_at: 10 },
+      { rowid: 2, hash: 'same-hash', created_at: 30 },
+    ]);
+
+    sqlite.close();
+  });
+
   it('recovers duplicate-column errors when the failed SQL contains quoted literals', async () => {
     const dataDir = mkdtempSync(join(tmpdir(), 'metapi-migrate-recover-quoted-'));
     process.env.DATA_DIR = dataDir;
@@ -385,6 +424,87 @@ describe('sqlite migrate bootstrap', () => {
     verified.close();
   });
 
+  it('recovers sequential duplicate-column migrations when a legacy sqlite schema predates the drizzle journal', async () => {
+    const dataDir = mkdtempSync(join(tmpdir(), 'metapi-migrate-legacy-schema-'));
+    const dbPath = join(dataDir, 'hub.db');
+    const sqlite = new Database(dbPath);
+    const journalEntries = readMigrationJournalEntries();
+    const appliedEntries = journalEntries.filter((entry) => entry.tag !== '0019_proxy_logs_stream_timing');
+
+    for (const entry of appliedEntries) {
+      const sqlText = readFileSync(join(migrationsDir, `${entry.tag}.sql`), 'utf8');
+      applyMigrationSql(sqlite, sqlText);
+    }
+
+    // Simulate SQLite legacy compatibility code partially adding the latest proxy log columns
+    // before drizzle creates its own migration journal.
+    sqlite.exec('ALTER TABLE proxy_logs ADD COLUMN is_stream integer;');
+    sqlite.close();
+
+    process.env.DATA_DIR = dataDir;
+    vi.resetModules();
+
+    await expect(import('./migrate.js')).resolves.toMatchObject({
+      runSqliteMigrations: expect.any(Function),
+    });
+
+    const verified = new Database(dbPath, { readonly: true });
+    const appliedRows = verified
+      .prepare('SELECT created_at FROM __drizzle_migrations ORDER BY created_at ASC')
+      .all() as Array<{ created_at: number }>;
+    const proxyLogColumns = verified
+      .prepare('PRAGMA table_info("proxy_logs")')
+      .all() as Array<{ name: string }>;
+
+    expect(appliedRows.map((row) => Number(row.created_at))).toEqual(
+      journalEntries.map((entry) => entry.when),
+    );
+    expect(proxyLogColumns.some((column) => column.name === 'is_stream')).toBe(true);
+    expect(proxyLogColumns.some((column) => column.name === 'first_byte_latency_ms')).toBe(true);
+
+    verified.close();
+  });
+
+  it('reconciles stale migration timestamps when the latest migration hash already exists', async () => {
+    const dataDir = mkdtempSync(join(tmpdir(), 'metapi-migrate-stale-timestamp-'));
+    const dbPath = join(dataDir, 'hub.db');
+    const sqlite = new Database(dbPath);
+    const journalEntries = readMigrationJournalEntries();
+
+    for (const entry of journalEntries) {
+      const sqlText = readFileSync(join(migrationsDir, `${entry.tag}.sql`), 'utf8');
+      applyMigrationSql(sqlite, sqlText);
+    }
+    recordAppliedMigrations(sqlite, journalEntries);
+
+    const latestEntry = journalEntries.at(-1);
+    const latestSqlText = latestEntry
+      ? readFileSync(join(migrationsDir, `${latestEntry.tag}.sql`), 'utf8')
+      : '';
+    const latestHash = createHash('sha256').update(latestSqlText).digest('hex');
+    // Introduce about 70 minutes of timestamp drift so journal reconciliation has work to do.
+    sqlite
+      .prepare('UPDATE __drizzle_migrations SET created_at = ? WHERE hash = ?')
+      .run((latestEntry?.when ?? 0) - STALE_JOURNAL_TIMESTAMP_DRIFT_MS, latestHash);
+    sqlite.close();
+
+    process.env.DATA_DIR = dataDir;
+    vi.resetModules();
+
+    await expect(import('./migrate.js')).resolves.toMatchObject({
+      runSqliteMigrations: expect.any(Function),
+    });
+
+    const verified = new Database(dbPath, { readonly: true });
+    const latestApplied = verified
+      .prepare('SELECT created_at FROM __drizzle_migrations WHERE hash = ? LIMIT 1')
+      .get(latestHash) as { created_at: number } | undefined;
+
+    expect(Number(latestApplied?.created_at)).toBe(latestEntry?.when);
+
+    verified.close();
+  });
+
   it('deduplicates legacy duplicate sites before applying the oauth site unique index', async () => {
     const dataDir = mkdtempSync(join(tmpdir(), 'metapi-migrate-duplicate-sites-'));
     const dbPath = join(dataDir, 'hub.db');

src/server/db/migrate.ts

@@ -264,12 +264,19 @@ function ensureDrizzleMigrationsTable(sqlite: Database.Database): void {
 function markMigrationRecordIfMissing(sqlite: Database.Database, record: MigrationRecord): boolean {
   ensureDrizzleMigrationsTable(sqlite);
   const existing = sqlite
-    .prepare('SELECT 1 FROM "__drizzle_migrations" WHERE "hash" = ? LIMIT 1')
+    .prepare('SELECT rowid, "created_at" FROM "__drizzle_migrations" WHERE "hash" = ? ORDER BY "created_at" DESC LIMIT 1')
-    .get(record.hash);
+    .get(record.hash) as { rowid?: number; created_at?: number } | undefined;
   if (existing) {
+    if (Number(existing.created_at) === record.createdAt) {
       return false;
     }
 
+    sqlite
+      .prepare('UPDATE "__drizzle_migrations" SET "created_at" = ? WHERE rowid = ?')
+      .run(record.createdAt, existing.rowid);
+    return true;
+  }
+
   sqlite
     .prepare('INSERT INTO "__drizzle_migrations" ("hash", "created_at") VALUES (?, ?)')
     .run(record.hash, record.createdAt);
@@ -373,23 +380,29 @@ function recoverMigrationSequence(
   sqlite: Database.Database,
   migrationsFolder: string,
   failedMigrationTag: string,
-): boolean {
+): number {
   const migrations = readRecoveryMigrations(migrationsFolder);
   const failedMigrationIndex = migrations.findIndex((migration) => migration.tag === failedMigrationTag);
   if (failedMigrationIndex < 0) {
-    return false;
+    return 0;
   }
 
+  let recoveredCount = 0;
   for (const migration of migrations.slice(0, failedMigrationIndex + 1)) {
     if (hasMigrationRecord(sqlite, migration)) {
+      if (markMigrationRecordIfMissing(sqlite, migration)) {
+        recoveredCount += 1;
+      }
       continue;
     }
 
     replayMigrationStatements(sqlite, migration.statements);
-    markMigrationRecordIfMissing(sqlite, migration);
+    if (markMigrationRecordIfMissing(sqlite, migration)) {
+      recoveredCount += 1;
+    }
   }
 
-  return true;
+  return recoveredCount;
 }
 
 function backfillMissingRecordedMigrations(sqlite: Database.Database, migrationsFolder: string): number {
@@ -398,6 +411,9 @@ function backfillMissingRecordedMigrations(sqlite: Database.Database, migrations
   let recoveredCount = 0;
   for (const migration of readRecoveryMigrations(migrationsFolder)) {
     if (hasMigrationRecord(sqlite, migration)) {
+      if (markMigrationRecordIfMissing(sqlite, migration)) {
+        recoveredCount += 1;
+      }
       continue;
     }
 
@@ -414,13 +430,18 @@ function backfillMissingRecordedMigrations(sqlite: Database.Database, migrations
   return recoveredCount;
 }
 
-function tryRecoverDuplicateColumnMigrationError(
+type DuplicateColumnRecoveryResult = {
+  tag: string;
+  recoveredCount: number;
+};
+
+function recoverDuplicateColumnMigrationError(
   sqlite: Database.Database,
   migrationsFolder: string,
   error: unknown,
-): boolean {
+): DuplicateColumnRecoveryResult | null {
   if (!isDuplicateColumnError(error)) {
-    return false;
+    return null;
   }
 
   const failedSqlText = extractFailedSqlFromError(error);
@@ -429,14 +450,26 @@ function tryRecoverDuplicateColumnMigrationError(
       ?? findMatchingMigrationByErrorMessage(migrationsFolder, error)
     : findMatchingMigrationByErrorMessage(migrationsFolder, error);
   if (!matchedMigration) {
-    return false;
+    return null;
   }
 
-  const recovered = recoverMigrationSequence(sqlite, migrationsFolder, matchedMigration.tag);
+  const recoveredCount = recoverMigrationSequence(sqlite, migrationsFolder, matchedMigration.tag);
-  if (recovered) {
+  if (recoveredCount > 0) {
     console.warn(`[db] Recovered duplicate-column migration sequence through ${matchedMigration.tag}.`);
   }
-  return recovered;
+  return {
+    tag: matchedMigration.tag,
+    recoveredCount,
+  };
+}
+
+function tryRecoverDuplicateColumnMigrationError(
+  sqlite: Database.Database,
+  migrationsFolder: string,
+  error: unknown,
+): boolean {
+  const recovery = recoverDuplicateColumnMigrationError(sqlite, migrationsFolder, error);
+  return (recovery?.recoveredCount ?? 0) > 0;
 }
 
 function rewriteDownstreamSiteWeightMultipliers(
@@ -593,20 +626,31 @@ export function runSqliteMigrations(): void {
   bootstrapLegacyDrizzleMigrations(sqlite, migrationsFolder);
   backfillMissingRecordedMigrations(sqlite, migrationsFolder);
 
+  while (true) {
     try {
       migrate(drizzle(sqlite), { migrationsFolder });
+      break;
     } catch (error) {
-    const recoveredDuplicateColumns = tryRecoverDuplicateColumnMigrationError(sqlite, migrationsFolder, error);
+      const duplicateColumnRecovery = recoverDuplicateColumnMigrationError(sqlite, migrationsFolder, error);
-    const recoveredDuplicateSites = (
+      if (duplicateColumnRecovery) {
-      !recoveredDuplicateColumns
+        if (duplicateColumnRecovery.recoveredCount > 0) {
-      && isSitesPlatformUrlUniqueConflictError(error)
+          continue;
-      && deduplicateLegacySitesForUniqueIndex(sqlite)
+        }
-    );
+        sqlite.close();
-    if (!recoveredDuplicateColumns && !recoveredDuplicateSites) {
+        throw error;
+      }
+
+      const recoveredDuplicateSites = (
+        isSitesPlatformUrlUniqueConflictError(error)
+        && deduplicateLegacySitesForUniqueIndex(sqlite)
+      );
+      if (recoveredDuplicateSites) {
+        continue;
+      }
+
       sqlite.close();
       throw error;
     }
-    migrate(drizzle(sqlite), { migrationsFolder });
   }
 
   sqlite.close();