pjandro/OmniRoute
1
0
Fork 0
mirror of https://github.com/diegosouzapw/OmniRoute.git synced 2026-09-14 02:42:24 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(db): harden migration recovery snapshots (#12435)

Browse Source 
Validado em lote numa worktree combinada com os 14 PRs desta campanha de error-boundary sobre o tip de `release/v3.8.51`: `typecheck:core` limpo e **120/120** nos 23 arquivos de teste que os PRs trazem.

Um ponto que só apareceu no tree combinado: **#12465 e #12466 criam o mesmo arquivo novo** `open-sse/utils/streamReadiness.ts` (que não existe no tip) com desenhos divergentes de cancelamento — `cancelled` + `releaseLock` imediato num, `readInFlight`/`cancelRequested` com `cancelReader` fire-and-forget no outro. Adotei a versão do #12466, que difere e defere o release do lock para quando a leitura em voo termina, e validei a escolha rodando as suítes dos **dois** PRs contra ela: 21/21 no readiness compartilhado e 22/22 incluindo o boundary do Perplexity.
This commit is contained in:
Diego Rodrigues de Sa e Souza
2026-09-03 21:01:09 -03:00
committed by GitHub
parent 627fcba605
commit 7ae8bf4e05
19 changed files with 2303 additions and 625 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
src/lib/dataPaths.ts
View File

@@ -101,30 +101,70 @@ export function isTestContext(): boolean {
);
}
/**
* `node --eval` / `node -e` (and their print variants) are common shapes used by
* one-off import probes.
* Such a process has no application entry point from which to establish storage intent,
* so defaulting it to the operator's durable database is unsafe. A deliberate production
* inspection can still opt in with an explicit DATA_DIR (preferred) or
* OMNIROUTE_ALLOW_DEFAULT_DATA_DIR=1.
*/
function isEvalProbeContext(): boolean {
return process.execArgv.some(
(arg) =>
arg === "--eval" ||
arg === "-e" ||
arg === "-pe" ||
arg === "-ep" ||
arg.startsWith("--eval=") ||
arg === "--print" ||
arg === "-p" ||
arg.startsWith("--print=")
);
}
/** Process-wide redirect target, so repeated calls share one DB instead of one per call. */
let testContextDataDir: string | null = null;
let testContextCleanupRegistered = false;
export function resolveWritableDataDir({ isCloud = false }: { isCloud?: boolean } = {}): string {
const resolved = resolveDataDir({ isCloud });
const configured = normalizeConfiguredPath(process.env.DATA_DIR);
// Cloud/serverless never owns a writable home dir; leave its sentinel alone.
if (isCloud) return resolved;
// #10428: a test/ad-hoc run that never chose a DATA_DIR would otherwise open the
// #10428: a test/eval-probe run that never chose a DATA_DIR would otherwise open the
// OPERATOR'S REAL database (~/.omniroute/storage.sqlite — live provider credentials).
// Redirect to a throwaway dir instead of throwing: the documented single-file command
// (`node --import tsx/esm --test tests/unit/x.test.ts`) does not load the isolation
// setup, and a hard failure there would only teach people to disable the guard.
// `OMNIROUTE_ALLOW_DEFAULT_DATA_DIR=1` opts back in, so the intent is recorded.
if (
!process.env.DATA_DIR &&
isTestContext() &&
!configured &&
(isTestContext() || isEvalProbeContext()) &&
process.env.OMNIROUTE_ALLOW_DEFAULT_DATA_DIR !== "1"
) {
if (!testContextDataDir) {
testContextDataDir = fs.mkdtempSync(path.join(os.tmpdir(), `${APP_NAME}-testctx-`));
if (!testContextCleanupRegistered) {
testContextCleanupRegistered = true;
process.once("exit", () => {
if (!testContextDataDir) return;
try {
fs.rmSync(testContextDataDir, {
recursive: true,
force: true,
maxRetries: 5,
retryDelay: 25,
});
} catch {
// An unclean exit is left to the operating system's temp-directory policy.
}
});
}
console.warn(
`[DATA_DIR] test context without DATA_DIR → using '${testContextDataDir}' instead of ` +
`[DATA_DIR] test/eval context without DATA_DIR → using '${testContextDataDir}' instead of ` +
`'${resolved}'. Set DATA_DIR explicitly (or load tests/_setup/isolateDataDir.ts) to silence this.`
);
}
@@ -132,7 +172,6 @@ export function resolveWritableDataDir({ isCloud = false }: { isCloud?: boolean
}
// No explicit override → already the default user dir; nothing to fall back to.
const configured = normalizeConfiguredPath(process.env.DATA_DIR);
if (!configured) return resolved;
try {

49
src/lib/db/backup.ts
View File

@@ -101,6 +101,24 @@ function getBackupDir() {
return DB_BACKUPS_DIR || path.join(DATA_DIR, "db_backups");
}
function listBackupFilesNewestFirst(backupDir: string) {
return fs
.readdirSync(backupDir)
.filter((filename) => filename.startsWith("db_") && filename.endsWith(".sqlite"))
.flatMap((filename) => {
try {
return [{ filename, stat: fs.statSync(path.join(backupDir, filename)) }];
} catch {
// A concurrent retention pass may remove an entry after readdir.
return [];
}
})
.sort(
(left, right) =>
right.stat.mtimeMs - left.stat.mtimeMs || right.filename.localeCompare(left.filename)
);
}
export function cleanupDbBackups(options?: {
maxFiles?: number;
retentionDays?: number;
@@ -272,16 +290,26 @@ export function backupDbFile(reason = "auto") {
if (reason !== "manual" && reason !== "pre-restore") {
// Shrink detection is useful for automatic safety backups, but it should
// never block an explicit operator action like manual backup or pre-restore.
// Only timestamp-named automatic/manual backups are shrink baselines. The
// content-addressed migration snapshots are restore points, not periodic size
// samples; excluding them also keeps this lookup to names only with a single stat
// even in legacy directories containing tens of thousands of timestamp backups.
const existingBackups = fs
.readdirSync(backupDir)
.filter((f) => f.startsWith("db_") && f.endsWith(".sqlite"))
.filter((filename) => /^db_\d{4}-.*\.sqlite$/.test(filename))
.sort();
if (existingBackups.length > 0) {
const latestBackup = existingBackups[existingBackups.length - 1];
const latestStat = fs.statSync(path.join(backupDir, latestBackup));
if (latestStat.size > 4096 && stat.size < latestStat.size * 0.5) {
console.warn(`[DB] Backup SKIPPED — DB shrank from ${latestStat.size}B to ${stat.size}B`);
return null;
const latestBackup = existingBackups.at(-1)!;
try {
const latestStat = fs.statSync(path.join(backupDir, latestBackup));
if (latestStat.size > 4096 && stat.size < latestStat.size * 0.5) {
console.warn(
`[DB] Backup SKIPPED — DB shrank from ${latestStat.size}B to ${stat.size}B`
);
return null;
}
} catch (error: unknown) {
if ((error as NodeJS.ErrnoException | null)?.code !== "ENOENT") throw error;
}
}
}
@@ -316,16 +344,11 @@ export async function listDbBackups() {
try {
if (!fs.existsSync(backupDir)) return [];
const entries = fs
.readdirSync(backupDir)
.filter((f) => f.startsWith("db_") && f.endsWith(".sqlite"))
.sort()
.reverse();
const entries = listBackupFilesNewestFirst(backupDir);
const { tryOpenSync } = await import("@/lib/db/adapters/driverFactory");
return entries.map((filename) => {
return entries.map(({ filename, stat }) => {
const filePath = path.join(backupDir, filename);
const stat = fs.statSync(filePath);
const match = filename.match(/^db_(.+?)_([^.]+)\.sqlite$/);
const reason = match ? match[2] : "unknown";

17
src/lib/db/backupRetention.ts
View File

@@ -1,17 +1,12 @@
/**
* Backup retention primitives — pure filesystem work, no `core.ts` dependency.
*
* This module exists so BOTH backup call sites can share one retention policy:
*
* - `backup.ts` (manual/API/auto backups) — resolves the operator's settings from the
* database and delegates here.
* - `migrationRunner.ts` (pre-migration snapshots) — cannot import `backup.ts`, because
* `core.ts` already imports `migrationRunner.ts` and `backup.ts` imports `core.ts`;
* that edge would close a cycle. Keeping the policy here, free of `core`, lets the
* migration path prune without one.
*
* Before #10421 the migration path had no retention at all and `db_backups/` grew
* without bound (observed: 48.999 files / 204 GB against a 5,3 MB live database).
* `backup.ts` (manual/API/auto backups) resolves the operator's settings from the
* database and delegates pure family pruning here. The migration runner deliberately
* does not prune during its concurrent safety window: its snapshots are content-addressed
* and reused for an identical DB state, while manual/scheduled cleanup remains the single
* retention boundary. Before #10421, repeated failed starts created distinct timestamped
* snapshots and `db_backups/` grew without bound (observed: 48,999 files / 204 GB).
*/
import fs from "fs";

10
src/lib/db/core.ts
View File

@@ -1118,10 +1118,10 @@ export function getDbInstance(): SqliteDatabase {
// This is needed so the migration runner skips the mass-migration safety abort
// that would otherwise trigger because heuristic seeding marks some migrations
// as applied, making the fresh DB look like a wiped existing DB (#1328).
// #9934: also classify as fresh a file that `omniroute setup` created with
// only the clipped skeleton schema (see the probe below) — even though the
// file exists, it has never had migrations run.
let isNewDb = !fs.existsSync(sqliteFile);
// #9934: also classify a setup-created skeleton as logically fresh for the mass guard,
// while tracking its pre-existing file independently for mandatory snapshot safety.
const databaseExistedBeforeInitialization = fs.existsSync(sqliteFile);
let isNewDb = !databaseExistedBeforeInitialization;
// Detect and handle old schema format — preserve data when possible (#146)
// Uses a single probe connection that becomes the real connection when possible.
@@ -1310,7 +1310,7 @@ export function getDbInstance(): SqliteDatabase {
VALUES ('001', 'initial_schema');
`);
runMigrations(db, { isNewDb });
runMigrations(db, { isNewDb, databaseExistedBeforeInitialization });
// Fresh installs need the same post-migration index guarantee as upgraded
// databases, including recovery from an interrupted migration 127 attempt.
ensureUsageHistoryAccountIndex(db);

814
src/lib/db/migrationRunner.ts
View File

@@ -21,37 +21,29 @@ import type { SqliteAdapter } from "./adapters/types";
import { DEFAULT_DATABASE_SETTINGS } from "@/types/databaseSettings";
import { isAutomatedTestProcess } from "@/shared/utils/testProcess";
import {
RENAMED_MIGRATION_COMPATIBILITY,
LEGACY_VERSION_SLOT_MIGRATIONS,
SUPERSEDED_DUPLICATE_MIGRATIONS,
PHYSICAL_SCHEMA_SENTINELS,
INITIAL_SCHEMA_SENTINELS,
OPTIONAL_FTS5_MIGRATION_VERSIONS,
RENAMED_MIGRATION_COMPATIBILITY,
SUPERSEDED_DUPLICATE_MIGRATIONS,
} from "./migrationRunner/constants";
import { getExtraMigrationFiles } from "./migrationRunner/extraDirs";
// Retention primitives live in their own `core`-free module: `core.ts` imports this file,
// so importing `backup.ts` (which imports `core.ts`) here would close a dependency cycle.
import { migrationConsole as console } from "./migrationRunner/logger";
import {
MAX_DB_BACKUPS,
DEFAULT_DB_BACKUP_RETENTION_DAYS,
parsePositiveInt,
parseNonNegativeInt,
pruneBackupDirectory,
} from "./backupRetention";
const isNodeTestRunnerChild = typeof process.env.NODE_TEST_CONTEXT === "string";
const console = {
log: (...args: unknown[]) => {
if (!isNodeTestRunnerChild) globalThis.console.log(...args);
},
warn: (...args: unknown[]) => {
if (!isNodeTestRunnerChild) globalThis.console.warn(...args);
},
error: (...args: unknown[]) => {
globalThis.console.error(...args);
},
};
createPreMigrationBackup,
hashFileSync,
type PreMigrationBackupReceipt,
} from "./migrationRunner/preMigrationBackup";
import {
detectNameMismatches,
getPlausiblePendingCount,
hasColumn,
hasLedgerRepairCandidates,
hasPhysicalTable,
hasTable,
inferPhysicalSchemaBaseline,
reconcileRenumberedMigrations,
rehomeLegacyVersionSlotMigrations,
} from "./migrationRunner/schemaState";
/**
* Resolve the migrations directory path safely across platforms.
@@ -336,16 +328,96 @@ function getAppliedRecords(db: SqliteAdapter): Array<{ version: string; name: st
}>;
}
function hasTable(db: SqliteAdapter, tableName: string): boolean {
const row = db
.prepare("SELECT name FROM sqlite_master WHERE type IN ('table', 'view') AND name = ?")
.get(tableName) as { name?: string } | undefined;
return Boolean(row?.name);
/**
* Reopen a narrowly selected migration when the table it creates is physically absent.
*
* Historical databases can carry `074_discovery_results` or the rehomed
* `081_inspector_custom_hosts` in the ledger without the table itself (for example after a
* version-slot collision or an incomplete manual recovery). Treating either marker as
* authoritative leaves an incomplete schema. A same-named view does not count as the table;
* replaying the owning migration fails closed instead of silently advancing.
*
* This intentionally detects table absence only. It is not a general schema-healing layer:
* column/rebuild migrations continue to use targeted idempotency checks elsewhere.
*/
const REQUIRED_PHYSICAL_MIGRATIONS = [
{ version: "074", name: "discovery_results", tableName: "discovery_results" },
{ version: "081", name: "inspector_custom_hosts", tableName: "inspector_custom_hosts" },
] as const;
function validateRequiredPhysicalMigrationProvenance(
db: SqliteAdapter,
files: Array<{ version: string; name: string; path: string }>
): void {
for (const required of REQUIRED_PHYSICAL_MIGRATIONS) {
if (hasPhysicalTable(db, required.tableName)) continue;
const migrationExists = files.some(
(file) => file.version === required.version && file.name === required.name
);
if (!migrationExists) continue;
const occupied = db
.prepare("SELECT version, name FROM _omniroute_migrations WHERE version = ?")
.get(required.version) as { version: string; name: string } | undefined;
if (!occupied || occupied.name === required.name) continue;
const knownRenumberedCollision = RENAMED_MIGRATION_COMPATIBILITY.some(
(compatibility) =>
compatibility.fromVersion === occupied.version &&
compatibility.fromName === occupied.name &&
files.some(
(file) => file.version === compatibility.toVersion && file.name === compatibility.toName
) &&
files.some(
(file) =>
file.version === compatibility.fromVersion && file.name !== compatibility.fromName
)
);
const knownLegacySlotCollision = LEGACY_VERSION_SLOT_MIGRATIONS.some(
(legacy) =>
legacy.version === occupied.version &&
legacy.name === occupied.name &&
files.some((file) => file.version === legacy.version && file.name !== legacy.name)
);
const knownRepairableCollision = knownRenumberedCollision || knownLegacySlotCollision;
if (knownRepairableCollision) continue;
throw new Error(
`[Migration] Required table "${required.tableName}" is missing, but version ` +
`${required.version} is recorded as unknown migration "${occupied.name}" instead of ` +
`"${required.name}". Refusing to treat this database as current.`
);
}
}
function hasColumn(db: SqliteAdapter, tableName: string, columnName: string): boolean {
const columns = db.prepare(`PRAGMA table_info(${tableName})`).all() as Array<{ name?: string }>;
return columns.some((column) => column.name === columnName);
function findAtomicPhysicalReplays(
db: SqliteAdapter,
files: Array<{ version: string; name: string; path: string }>
): Set<string> {
const replayVersions = new Set<string>();
for (const required of REQUIRED_PHYSICAL_MIGRATIONS) {
if (hasPhysicalTable(db, required.tableName)) continue;
const migrationExists = files.some(
(file) => file.version === required.version && file.name === required.name
);
if (!migrationExists) continue;
const applied = db
.prepare("SELECT version, name FROM _omniroute_migrations WHERE version = ? AND name = ?")
.get(required.version, required.name) as { version: string; name: string } | undefined;
if (!applied) continue;
replayVersions.add(required.version);
console.warn(
`[Migration] Will atomically replay ${required.version}_${required.name}: ledger recorded ` +
`"${applied.name}" but required table "${required.tableName}" is missing.`
);
}
return replayVersions;
}
function ensureColumn(db: SqliteAdapter, tableName: string, columnName: string, ddl: string): void {
@@ -651,276 +723,31 @@ function applyCompressionCombosMigration(db: SqliteAdapter, migrationPath: strin
`);
}
function inferPhysicalSchemaBaseline(db: SqliteAdapter): {
version: string;
description: string;
} | null {
for (const sentinel of PHYSICAL_SCHEMA_SENTINELS) {
if (hasTable(db, sentinel.tableName)) {
return {
version: sentinel.version,
description: sentinel.description,
};
}
}
const hasInitialSchema = INITIAL_SCHEMA_SENTINELS.every((tableName) => hasTable(db, tableName));
if (hasInitialSchema) {
return {
version: "001",
description: "initial schema tables",
};
}
return null;
}
function getPlausiblePendingCount(
files: Array<{ version: string; name: string; path: string }>,
baselineVersion: string
): number {
const baseline = Number.parseInt(baselineVersion, 10);
return files.filter((file) => Number.parseInt(file.version, 10) > baseline).length;
}
/**
* Detect migration name mismatches — when a migration version number
* has been reused/renumbered with a different name. This is a strong signal
* that the migration tracking is corrupted or migrations were renumbered.
*/
function detectNameMismatches(
appliedRecords: Array<{ version: string; name: string }>,
files: Array<{ version: string; name: string; path: string }>
): Array<{ version: string; appliedName: string; diskName: string }> {
const appliedByName = new Map(appliedRecords.map((r) => [r.version, r.name]));
const mismatches: Array<{ version: string; appliedName: string; diskName: string }> = [];
for (const file of files) {
const appliedName = appliedByName.get(file.version);
if (appliedName && appliedName !== file.name) {
mismatches.push({
version: file.version,
appliedName,
diskName: file.name,
});
}
}
return mismatches;
}
function reconcileRenumberedMigrations(
db: SqliteAdapter,
files: Array<{ version: string; name: string; path: string }>
): boolean {
let repaired = false;
for (const compatibility of RENAMED_MIGRATION_COMPATIBILITY) {
const hasTargetFile = files.some(
(file) => file.version === compatibility.toVersion && file.name === compatibility.toName
);
const hasSourceFile = files.some(
(file) => file.version === compatibility.fromVersion && file.name !== compatibility.fromName
);
if (!hasTargetFile || !hasSourceFile) {
continue;
}
const legacyRow = db
.prepare("SELECT version, name FROM _omniroute_migrations WHERE version = ? AND name = ?")
.get(compatibility.fromVersion, compatibility.fromName) as
{ version: string; name: string } | undefined;
if (!legacyRow) {
continue;
}
const targetRow = db
.prepare("SELECT version FROM _omniroute_migrations WHERE version = ?")
.get(compatibility.toVersion) as { version: string } | undefined;
const applyRepair = db.transaction(() => {
if (targetRow) {
db.prepare("DELETE FROM _omniroute_migrations WHERE version = ? AND name = ?").run(
compatibility.fromVersion,
compatibility.fromName
);
} else {
db.prepare(
"UPDATE _omniroute_migrations SET version = ?, name = ? WHERE version = ? AND name = ?"
).run(
compatibility.toVersion,
compatibility.toName,
compatibility.fromVersion,
compatibility.fromName
);
}
});
applyRepair();
repaired = true;
console.warn(
`[Migration] Reconciled renamed migration ${compatibility.fromVersion}_${compatibility.fromName} ` +
`to ${compatibility.toVersion}_${compatibility.toName} to preserve pending migrations.`
);
// After the compat rewrite, verify the old version slot is now free.
// A residual row (from a failed prior run, manual intervention, or edge-case
// UPDATE conflict) at the old version would shadow a NEW migration file
// placed at that version number — e.g. 028_create_files_and_batches.sql
// would be skipped because getAppliedVersions() still sees version "028".
const residualRow = db
.prepare("SELECT version, name FROM _omniroute_migrations WHERE version = ?")
.get(compatibility.fromVersion) as { version: string; name: string } | undefined;
if (residualRow) {
console.warn(
`[Migration] ⚠️ Residual row at version ${compatibility.fromVersion} ` +
`(name: "${residualRow.name}") still present after compat rewrite — ` +
`removing to unblock new migration at this version slot.`
);
db.prepare("DELETE FROM _omniroute_migrations WHERE version = ?").run(
compatibility.fromVersion
);
}
}
return repaired;
}
function rehomeLegacyVersionSlotMigrations(
db: SqliteAdapter,
files: Array<{ version: string; name: string; path: string }>
): boolean {
let repaired = false;
const diskNamesByVersion = new Map(files.map((file) => [file.version, file.name]));
for (const legacy of LEGACY_VERSION_SLOT_MIGRATIONS) {
const diskName = diskNamesByVersion.get(legacy.version);
if (!diskName || diskName === legacy.name) {
continue;
}
const legacyRow = db
.prepare("SELECT version, name FROM _omniroute_migrations WHERE version = ? AND name = ?")
.get(legacy.version, legacy.name) as { version: string; name: string } | undefined;
if (!legacyRow) {
continue;
}
const legacyVersion = `legacy-${legacy.version}-${legacy.name}`;
const applyRepair = db.transaction(() => {
const existingLegacyRow = db
.prepare("SELECT version FROM _omniroute_migrations WHERE version = ?")
.get(legacyVersion) as { version: string } | undefined;
if (existingLegacyRow) {
db.prepare("DELETE FROM _omniroute_migrations WHERE version = ? AND name = ?").run(
legacy.version,
legacy.name
);
return;
}
db.prepare("UPDATE _omniroute_migrations SET version = ? WHERE version = ? AND name = ?").run(
legacyVersion,
legacy.version,
legacy.name
);
});
applyRepair();
repaired = true;
console.warn(
`[Migration] Rehomed legacy migration ${legacy.version}_${legacy.name} ` +
`to ${legacyVersion} so current ${legacy.version}_${diskName} can apply.`
);
}
return repaired;
}
/**
* Read a persisted `dbBackup` retention setting through the adapter that is ALREADY open
* for this migration run.
* Run a callback while holding SQLite's IMMEDIATE writer transaction.
*
* `backup.ts`'s equivalent goes through `getDbInstance()`, which is unsafe here: this
* code runs from inside database initialization, so asking for the singleton would
* re-enter it. Reading off `db` keeps the same stored values without that risk. A DB too
* old to have `key_value` yet simply falls back to the default.
* Production adapters expose `immediate()` directly. A small number of long-standing
* migration tests and external callers still pass a raw better-sqlite3 Database, whose
* transaction wrapper exposes `.immediate()` instead. Supporting both shapes here keeps
* the safety transaction real: this must never degrade to a plain callback invocation.
*/
function readStoredBackupSetting(db: SqliteAdapter, key: string, min: number): number | undefined {
try {
const row = db
.prepare("SELECT value FROM key_value WHERE namespace = ? AND key = ?")
.get("dbBackup", key) as { value?: string } | undefined;
if (!row?.value) return undefined;
const parsed = JSON.parse(row.value);
return Number.isInteger(parsed) && parsed >= min ? parsed : undefined;
} catch {
return undefined;
function runImmediateTransaction<T>(db: SqliteAdapter, fn: () => T): T {
const adapterImmediate = (db as Partial<SqliteAdapter>).immediate;
if (typeof adapterImmediate === "function") {
let result!: T;
adapterImmediate.call(db, () => {
result = fn();
});
return result;
}
}
/**
* Enforce the backup retention budget after a pre-migration snapshot (#10421).
*
* Precedence matches `backup.ts`: env override → persisted operator setting → default.
* Never throws: a migration must not fail because housekeeping did.
*/
function pruneMigrationBackups(db: SqliteAdapter, backupDir: string): void {
try {
const maxFiles = process.env.DB_BACKUP_MAX_FILES
? parsePositiveInt(process.env.DB_BACKUP_MAX_FILES, MAX_DB_BACKUPS)
: (readStoredBackupSetting(db, "maxFiles", 1) ?? MAX_DB_BACKUPS);
const retentionDays = process.env.DB_BACKUP_RETENTION_DAYS
? parseNonNegativeInt(process.env.DB_BACKUP_RETENTION_DAYS, DEFAULT_DB_BACKUP_RETENTION_DAYS)
: (readStoredBackupSetting(db, "retentionDays", 0) ?? DEFAULT_DB_BACKUP_RETENTION_DAYS);
const result = pruneBackupDirectory({ backupDir, maxFiles, retentionDays });
if (result.deletedFiles > 0) {
console.log(
`[Migration] Pruned ${result.deletedFiles} old backup file(s) ` +
`(${result.keptBackupFamilies} kept, maxFiles=${maxFiles}, retentionDays=${retentionDays}).`
);
}
} catch (err: unknown) {
const message = err instanceof Error ? err.message : String(err);
console.warn(`[Migration] Failed to prune old backups: ${message}`);
}
}
/**
* Create a pre-migration backup of the SQLite database using VACUUM INTO.
* Returns the backup path on success, null on failure.
*/
function createPreMigrationBackup(db: SqliteAdapter): string | null {
try {
const sqliteFile = db.name;
if (!sqliteFile || sqliteFile === ":memory:") return null;
const backupDir = path.join(path.dirname(sqliteFile), "db_backups");
if (!fs.existsSync(backupDir)) {
fs.mkdirSync(backupDir, { recursive: true });
}
const timestamp = new Date().toISOString().replace(/[:.]/g, "-");
const backupPath = path.join(backupDir, `db_${timestamp}_pre-migration.sqlite`);
const escapedBackupPath = backupPath.replace(/'/g, "''");
db.exec(`VACUUM INTO '${escapedBackupPath}'`);
console.log(`[Migration] Pre-migration backup created: ${backupPath}`);
// #10421: apply the operator's retention budget right here. Without this the
// migration path was the one backup producer that never pruned, so every process
// start with a pending migration added ~5 MB forever (observed: 49k files / 204 GB).
pruneMigrationBackups(db, backupDir);
return backupPath;
} catch (err: unknown) {
const message = err instanceof Error ? err.message : String(err);
console.warn(`[Migration] Failed to create pre-migration backup: ${message}`);
return null;
const rawTransaction = db.transaction(fn) as ReturnType<SqliteAdapter["transaction"]> & {
immediate?: () => T;
};
if (typeof rawTransaction.immediate !== "function") {
throw new Error("[Migration] Database adapter does not support IMMEDIATE transactions.");
}
return rawTransaction.immediate();
}
/**
@@ -932,15 +759,243 @@ function createPreMigrationBackup(db: SqliteAdapter): string | null {
* 2. Aborts if too many pending migrations on an existing DB (likely wipe)
* 3. Creates automatic backup before running any migrations
*/
export function runMigrations(db: SqliteAdapter, options?: { isNewDb?: boolean }): number {
export function runMigrations(
db: SqliteAdapter,
options?: { isNewDb?: boolean; databaseExistedBeforeInitialization?: boolean }
): number {
const isNewDb = options?.isNewDb === true;
// `isNewDb` also covers a setup-created skeleton so it can bypass the mass-migration
// false positive. Snapshot eligibility must use the independent physical-file fact:
// that skeleton can already contain provider credentials and other operator state.
const databaseExistedBeforeInitialization =
options?.databaseExistedBeforeInitialization ?? !isNewDb;
ensureMigrationsTable(db);
const files = filterSupersededDuplicateMigrations(getMigrationFiles());
rehomeLegacyVersionSlotMigrations(db, files);
reconcileRenumberedMigrations(db, files);
const applied = getAppliedVersions(db);
const appliedRecords = getAppliedRecords(db);
validateRequiredPhysicalMigrationProvenance(db, files);
let preMigrationBackup: PreMigrationBackupReceipt | null = null;
let plan!: {
atomicPhysicalReplays: Set<string>;
appliedRecords: Array<{ version: string; name: string }>;
pending: typeof files;
deferredUnsupported: typeof files;
highestAppliedBeforeMigrations: number;
};
let count = 0;
const preliminaryApplied = getAppliedVersions(db);
const preliminaryAtomicReplays = findAtomicPhysicalReplays(db, files);
const preliminaryPending = files.filter(
(file) => !preliminaryApplied.has(file.version) || preliminaryAtomicReplays.has(file.version)
);
const preliminaryDeferred = preliminaryPending.filter((migration) =>
isDeferredUnsupportedMigration(db, migration)
);
const preliminaryActionable = preliminaryPending.filter(
(migration) => !preliminaryDeferred.some((deferred) => deferred.version === migration.version)
);
const preliminaryHasRepairCandidates = hasLedgerRepairCandidates(db, files);
// Preserve the historical read-only/no-op path. Merely checking an already-current
// database must not acquire a writer lock (or fail SQLITE_BUSY because another supported
// host currently owns one). Safety state is recomputed under IMMEDIATE whenever work exists.
if (preliminaryActionable.length === 0 && !preliminaryHasRepairCandidates) {
const numericApplied = Array.from(preliminaryApplied)
.map((version) => Number.parseInt(version, 10))
.filter((version) => !Number.isNaN(version));
plan = {
atomicPhysicalReplays: preliminaryAtomicReplays,
appliedRecords: getAppliedRecords(db),
pending: preliminaryPending,
deferredUnsupported: preliminaryDeferred,
highestAppliedBeforeMigrations: numericApplied.length > 0 ? Math.max(...numericApplied) : 0,
};
}
// sql.js export() finalizes its active SAVEPOINT, so exporting from inside
// `db.immediate()` would make a later safety throw unable to roll repairs back.
// Its adapter is synchronous and in-memory, so no JavaScript writer can interleave
// between this preflight/export and the immediately following savepoint.
if (
!plan &&
db.driver === "sql.js" &&
(preliminaryActionable.length > 0 || preliminaryHasRepairCandidates)
) {
const needsSnapshot =
(preliminaryActionable.length > 0 || preliminaryHasRepairCandidates) &&
db.name !== ":memory:" &&
databaseExistedBeforeInitialization;
if (needsSnapshot) {
preMigrationBackup = createPreMigrationBackup(db);
if (!preMigrationBackup) {
throw new Error(
"[Migration] Refusing to migrate an existing database without a durable snapshot. " +
"The DATA_DIR filesystem must support atomic hard-link publication."
);
}
}
}
// Hold SQLite's native writer lock through snapshot selection, compatibility repairs,
// and the mass-safety decision. Native adapters open a separate read-only connection
// for VACUUM INTO while competing writers remain blocked. The outer transaction then
// commits before migrations so the repository's one-transaction-per-file contract stays
// intact: an earlier successful migration remains committed if a later file fails.
if (!plan)
runImmediateTransaction(db, () => {
const appliedBeforeRepair = getAppliedVersions(db);
const hadAppliedBeforeRepair = appliedBeforeRepair.size > 0;
const preliminaryAtomicReplays = findAtomicPhysicalReplays(db, files);
const preliminaryPending = files.filter(
(file) =>
!appliedBeforeRepair.has(file.version) || preliminaryAtomicReplays.has(file.version)
);
const preliminaryActionable = preliminaryPending.filter(
(migration) => !isDeferredUnsupportedMigration(db, migration)
);
const mayWriteExistingDatabase =
preliminaryActionable.length > 0 || hasLedgerRepairCandidates(db, files);
const needsSnapshot =
mayWriteExistingDatabase && db.name !== ":memory:" && databaseExistedBeforeInitialization;
if (needsSnapshot && !preMigrationBackup) {
if (db.driver === "sql.js") {
throw new Error(
"[Migration] sql.js safety state changed after its pre-transaction snapshot preflight; " +
"refusing to export from inside the rollback savepoint."
);
}
preMigrationBackup = createPreMigrationBackup(db);
if (!preMigrationBackup) {
throw new Error(
"[Migration] Refusing to migrate an existing database without a durable snapshot. " +
"The DATA_DIR filesystem must support atomic hard-link publication."
);
}
}
rehomeLegacyVersionSlotMigrations(db, files);
reconcileRenumberedMigrations(db, files);
const atomicPhysicalReplays = findAtomicPhysicalReplays(db, files);
const applied = getAppliedVersions(db);
const appliedRecords = getAppliedRecords(db);
const pending = files.filter(
(file) => !applied.has(file.version) || atomicPhysicalReplays.has(file.version)
);
const deferredUnsupported = pending.filter((migration) =>
isDeferredUnsupportedMigration(db, migration)
);
const actionablePending = pending.filter(
(migration) =>
!deferredUnsupported.some((deferred) => deferred.version === migration.version)
);
const isFreshSeedOnly =
applied.size === 1 &&
applied.has("001") &&
inferPhysicalSchemaBaseline(db) === null &&
hasTable(db, "provider_connections");
const requiresDurableBackup =
actionablePending.length > 0 &&
db.name !== ":memory:" &&
databaseExistedBeforeInitialization;
// Recompute under the same writer transaction as repairs and fail before any
// ledger mutation can commit if the durable-snapshot requirement is not met.
if (requiresDurableBackup && !preMigrationBackup) {
throw new Error(
"[Migration] Refusing to migrate an existing database without a durable snapshot. " +
"The DATA_DIR filesystem must support atomic hard-link publication."
);
}
const isTestEnvironment = isAutomatedTestProcess();
const maxPendingMigrations = resolveMaxPendingMigrations();
if (
actionablePending.length > 0 &&
!isTestEnvironment &&
!isNewDb &&
!isFreshSeedOnly &&
maxPendingMigrations > 0 &&
(applied.size > 0 || hadAppliedBeforeRepair) &&
actionablePending.length > maxPendingMigrations
) {
const physicalBaseline = inferPhysicalSchemaBaseline(db);
const plausiblePendingCount = physicalBaseline
? getPlausiblePendingCount(files, physicalBaseline.version)
: null;
if (plausiblePendingCount !== null && actionablePending.length <= plausiblePendingCount) {
console.warn(
`[Migration] Allowing ${actionablePending.length} pending migrations on an existing database ` +
`because the physical schema only proves ${physicalBaseline?.version} ` +
`(${physicalBaseline?.description}).`
);
} else {
const schemaHint =
physicalBaseline && plausiblePendingCount !== null
? ` Physical schema already shows ${physicalBaseline.version} ` +
`(${physicalBaseline.description}), so at most ${plausiblePendingCount} pending ` +
`migration(s) are expected from a legitimate upgrade.`
: "";
const bypassHint =
` To bypass this check (e.g. after restoring a backup where the migration ` +
`tracking table was wiped), set OMNIROUTE_MAX_PENDING_MIGRATIONS=0 in your ` +
`server.env or DATA_DIR/.env and restart.`;
const msg =
`[Migration] 🛑 ABORT: Detected ${actionablePending.length} pending migrations on an existing database ` +
`(threshold is ${maxPendingMigrations}). ` +
`This usually means the migration tracking table was accidentally wiped. ` +
`Running all migrations from scratch will cause data loss or schema errors.` +
schemaHint +
bypassHint;
if (memoizedSafetyAbort && memoizedSafetyAbort.message === msg) {
console.error(
`[Migration] 🛑 ABORT (repeat — see earlier detail): ` +
`${actionablePending.length} pending > threshold ${maxPendingMigrations}. ` +
`Set OMNIROUTE_MAX_PENDING_MIGRATIONS=0 to bypass.`
);
throw memoizedSafetyAbort;
}
console.error(msg);
memoizedSafetyAbort = new MigrationSafetyAbortError(msg);
throw memoizedSafetyAbort;
}
}
if (
preMigrationBackup &&
hashFileSync(preMigrationBackup.path) !== preMigrationBackup.sha256
) {
throw new Error(
"[Migration] Refusing to migrate because the pre-migration snapshot changed before use."
);
}
const numericApplied = Array.from(applied)
.map((version) => Number.parseInt(version, 10))
.filter((version) => !Number.isNaN(version));
const highestAppliedBeforeMigrations =
numericApplied.length > 0 ? Math.max(...numericApplied) : 0;
plan = {
atomicPhysicalReplays,
appliedRecords,
pending,
deferredUnsupported,
highestAppliedBeforeMigrations,
};
});
const {
atomicPhysicalReplays,
appliedRecords,
pending,
deferredUnsupported,
highestAppliedBeforeMigrations,
} = plan;
// ── Safety Check 1: Detect migration name mismatches (renumbering) ──
const mismatches = detectNameMismatches(appliedRecords, files);
@@ -963,34 +1018,15 @@ export function runMigrations(db: SqliteAdapter, options?: { isNewDb?: boolean }
);
}
// ── Gap Reconciliation: Identify non-contiguous missing migrations ──
// Do not rely on any highest-version-applied heuristic. We must explicitly
// iterate through all missing files on disk and apply them if they are missing
// from the _omniroute_migrations table.
const numericApplied = Array.from(applied)
.map((v) => Number.parseInt(v, 10))
.filter((n) => !Number.isNaN(n));
const highestApplied = numericApplied.length > 0 ? Math.max(...numericApplied) : 0;
const pending = files.filter((f) => {
const isMissing = !applied.has(f.version);
if (isMissing && Number(f.version) < highestApplied) {
for (const migration of pending) {
if (Number(migration.version) < highestAppliedBeforeMigrations) {
console.warn(
`[Migration] 🔄 RECONCILIATION: Found missing intermediate migration ` +
`${f.version}_${f.name} (highest applied is ${highestApplied}). ` +
`${migration.version}_${migration.name} ` +
`(highest applied is ${highestAppliedBeforeMigrations}). ` +
`This gap will be back-filled to ensure schema integrity.`
);
}
return isMissing;
});
const deferredUnsupported = pending.filter((migration) =>
isDeferredUnsupportedMigration(db, migration)
);
const actionablePending = pending.filter(
(migration) => !deferredUnsupported.some((deferred) => deferred.version === migration.version)
);
if (pending.length === 0) {
return 0; // Nothing to do
}
if (deferredUnsupported.length > 0) {
@@ -1003,101 +1039,28 @@ export function runMigrations(db: SqliteAdapter, options?: { isNewDb?: boolean }
);
}
// ── Safety Check 2: Mass-migration detection (abort if existing DB + many migrations) ──
// Skip in test environments where fresh DBs legitimately have many pending migrations.
const isTestEnvironment = isAutomatedTestProcess();
// #3416: resolve the threshold at call time so OMNIROUTE_MAX_PENDING_MIGRATIONS
// can override the default (0 disables the check). The abort message below
// interpolates this resolved value, so it auto-reflects any override.
const maxPendingMigrations = resolveMaxPendingMigrations();
// #9934: `omniroute setup`'s openOmniRouteDb writes a partial skeleton file
// (provider_connections + key_value) that has never had migrations run. When
// the first `serve` opens it and auto-seeds only the 001 marker, the applied
// set is exactly {001} — which would otherwise look like a wiped existing DB
// and trip this abort on a brand-new install. This is distinct from a real
// wiped/backup-restored database: that case has a non-trivial physical schema
// (baseline inference is non-null) and full data tables, so it still aborts.
// The 001-marker-only state on a provider_connections skeleton is the fresh
// auto-seed — let it through. A genuinely empty table is already exempt via
// `applied.size > 0`, and an upgraded DB has a non-trivial applied set.
const isFreshSeedOnly =
applied.size === 1 &&
applied.has("001") &&
inferPhysicalSchemaBaseline(db) === null &&
hasTable(db, "provider_connections");
if (
!isTestEnvironment &&
!isNewDb &&
!isFreshSeedOnly &&
process.env.DISABLE_SQLITE_AUTO_BACKUP !== "true" &&
maxPendingMigrations > 0 &&
applied.size > 0 &&
actionablePending.length > maxPendingMigrations
) {
const physicalBaseline = inferPhysicalSchemaBaseline(db);
const plausiblePendingCount = physicalBaseline
? getPlausiblePendingCount(files, physicalBaseline.version)
: null;
if (plausiblePendingCount !== null && actionablePending.length <= plausiblePendingCount) {
console.warn(
`[Migration] Allowing ${actionablePending.length} pending migrations on an existing database ` +
`because the physical schema only proves ${physicalBaseline?.version} ` +
`(${physicalBaseline?.description}).`
);
} else {
const schemaHint =
physicalBaseline && plausiblePendingCount !== null
? ` Physical schema already shows ${physicalBaseline.version} ` +
`(${physicalBaseline.description}), so at most ${plausiblePendingCount} pending ` +
`migration(s) are expected from a legitimate upgrade.`
: "";
const bypassHint =
` To bypass this check (e.g. after restoring a backup where the migration ` +
`tracking table was wiped), set OMNIROUTE_MAX_PENDING_MIGRATIONS=0 in your ` +
`server.env or DATA_DIR/.env and restart.`;
const msg =
`[Migration] 🛑 ABORT: Detected ${actionablePending.length} pending migrations on an existing database ` +
`(threshold is ${maxPendingMigrations}). ` +
`This usually means the migration tracking table was accidentally wiped. ` +
`Running all migrations from scratch will cause data loss or schema errors.` +
schemaHint +
bypassHint;
// #6260: memoize so the cascade of downstream ensureDbInitialized() calls
// that re-open the DB throw the SAME instance and only log once.
if (memoizedSafetyAbort && memoizedSafetyAbort.message === msg) {
console.error(
`[Migration] 🛑 ABORT (repeat — see earlier detail): ` +
`${actionablePending.length} pending > threshold ${maxPendingMigrations}. ` +
`Set OMNIROUTE_MAX_PENDING_MIGRATIONS=0 to bypass.`
);
throw memoizedSafetyAbort;
}
console.error(msg);
memoizedSafetyAbort = new MigrationSafetyAbortError(msg);
throw memoizedSafetyAbort;
}
if (preMigrationBackup && hashFileSync(preMigrationBackup.path) !== preMigrationBackup.sha256) {
throw new Error(
"[Migration] Refusing to migrate because the pre-migration snapshot changed before use."
);
}
// ── Safety Check 3: Pre-migration backup ──
// Skip backup if it's a completely fresh database (0 applied and all pending)
// or if running in tests (where AUTO_BACKUP might be disabled)
if (applied.size > 0 && process.env.DISABLE_SQLITE_AUTO_BACKUP !== "true") {
createPreMigrationBackup(db);
}
let count = 0;
for (const migration of pending) {
if (isDeferredUnsupportedMigration(db, migration)) {
continue;
}
if (isDeferredUnsupportedMigration(db, migration)) continue;
const applyMigration = db.transaction(() => {
if (atomicPhysicalReplays.has(migration.version)) {
const removed = db
.prepare("DELETE FROM _omniroute_migrations WHERE version = ? AND name = ?")
.run(migration.version, migration.name);
if (removed.changes !== 1) {
throw new Error(
`[Migration] Atomic replay lost its expected ledger marker for ` +
`${migration.version}_${migration.name}.`
);
}
}
if (isSchemaAlreadyApplied(db, migration)) {
console.warn(
`[Migration] Skipped executing ${migration.version}_${migration.name} as schema changes are already present (Idempotency check).`
@@ -1120,29 +1083,36 @@ export function runMigrations(db: SqliteAdapter, options?: { isNewDb?: boolean }
try {
applyMigration();
count++;
count += 1;
console.log(`[Migration] Applied: ${migration.version}_${migration.name}`);
} catch (err: unknown) {
const message = err instanceof Error ? err.message : String(err);
// "duplicate column name" means the column already exists — end state achieved, mark applied.
if (message.includes("duplicate column name")) {
if (
message.includes("duplicate column name") &&
!atomicPhysicalReplays.has(migration.version)
) {
const applyMarkerOnly = db.transaction(() => {
db.prepare(
"INSERT OR IGNORE INTO _omniroute_migrations (version, name) VALUES (?, ?)"
).run(migration.version, migration.name);
});
applyMarkerOnly();
count++;
count += 1;
console.log(
`[Migration] Applied (column pre-exists): ${migration.version}_${migration.name}`
);
} else {
console.error(`[Migration] FAILED: ${migration.version}_${migration.name} — ${message}`);
throw err; // Re-throw to prevent DB from starting in inconsistent state
throw err;
}
}
}
// Retention intentionally does not run inside the migration window. Another process
// may still be using a different snapshot as its in-flight restore point. Manual and
// scheduled backup paths continue to enforce the operator's retention policy; retries
// here are bounded by the deterministic content address instead of destructive pruning.
if (count > 0) {
console.log(`[Migration] ${count} migration(s) applied successfully.`);
}
@@ -1175,7 +1145,7 @@ function insertDefaultDatabaseSettings(db: SqliteAdapter) {
// Run in an immediate transaction to avoid nested transactions
try {
db.immediate(() => {
runImmediateTransaction(db, () => {
tx();
});
} catch (error) {

8
src/lib/db/migrationRunner/constants.ts
View File

@@ -158,6 +158,14 @@ export const RENAMED_MIGRATION_COMPATIBILITY = [
toVersion: "151",
toName: "windsurf_to_devin_desktop",
},
{
// inspector_custom_hosts was once published in slot 074, now occupied by
// discovery_results. Its canonical idempotent migration lives at 081.
fromVersion: "074",
fromName: "inspector_custom_hosts",
toVersion: "081",
toName: "inspector_custom_hosts",
},
{
fromVersion: "134",
fromName: "ccr_blocks",

13
src/lib/db/migrationRunner/logger.ts Normal file
View File

@@ -0,0 +1,13 @@
const isNodeTestRunnerChild = typeof process.env.NODE_TEST_CONTEXT === "string";
export const migrationConsole = {
log: (...args: unknown[]) => {
if (!isNodeTestRunnerChild) globalThis.console.log(...args);
},
warn: (...args: unknown[]) => {
if (!isNodeTestRunnerChild) globalThis.console.warn(...args);
},
error: (...args: unknown[]) => {
globalThis.console.error(...args);
},
};

293
src/lib/db/migrationRunner/preMigrationBackup.ts Normal file
View File

@@ -0,0 +1,293 @@
import { createHash } from "crypto";
import fs from "fs";
import path from "path";
import type { SqliteAdapter } from "../adapters/types";
import { tryOpenSync } from "../adapters/driverFactory";
import { migrationConsole as console } from "./logger";
export type PreMigrationBackupReceipt = {
path: string;
sha256: string;
};
function fsyncDirectoryEntry(directory: string): void {
let fd: number | null = null;
try {
fd = fs.openSync(directory, "r");
fs.fsyncSync(fd);
} catch (error: unknown) {
const code = (error as NodeJS.ErrnoException | null)?.code;
const windowsDirectoryHandleUnsupported =
process.platform === "win32" &&
(code === "EACCES" || code === "EPERM" || code === "EISDIR" || code === "EINVAL");
if (!windowsDirectoryHandleUnsupported) throw error;
} finally {
if (fd !== null) fs.closeSync(fd);
}
}
export function hashFileSync(filePath: string): string {
const hash = createHash("sha256");
const fd = fs.openSync(filePath, "r");
const buffer = Buffer.allocUnsafe(1024 * 1024);
let position = 0;
try {
while (true) {
const bytesRead = fs.readSync(fd, buffer, 0, buffer.length, position);
if (bytesRead === 0) break;
hash.update(buffer.subarray(0, bytesRead));
position += bytesRead;
}
} finally {
fs.closeSync(fd);
}
return hash.digest("hex");
}
function getReusablePreMigrationBackup(
candidatePath: string,
expectedSha256: string
): PreMigrationBackupReceipt | null {
if (!fs.existsSync(candidatePath)) return null;
const before = fs.lstatSync(candidatePath);
if (!before.isFile() || hashFileSync(candidatePath) !== expectedSha256) {
throw new Error(
`[Migration] Content-addressed snapshot path exists with unexpected content: ${candidatePath}`
);
}
const after = fs.lstatSync(candidatePath);
if (
before.dev !== after.dev ||
before.ino !== after.ino ||
before.size !== after.size ||
before.mtimeMs !== after.mtimeMs
) {
throw new Error(
`[Migration] Content-addressed snapshot changed while it was being validated: ${candidatePath}`
);
}
return { path: candidatePath, sha256: expectedSha256 };
}
function publishSnapshotWithoutOverwrite(tempPath: string, destination: string): void {
// link() publishes a complete same-filesystem image atomically and, unlike rename(),
// fails with EEXIST instead of overwriting a path created by another process. There is
// deliberately no copy/rename fallback: filesystems without this primitive fail closed
// instead of exposing a partial canonical `.sqlite` file after a crash.
fs.linkSync(tempPath, destination);
const publishedFd = fs.openSync(destination, "r+");
try {
// Flush through the published name as well as the already-fsynced temp handle.
// On Windows this maps to FlushFileBuffers and is the strongest file-level
// durability proof available when directory handles are unsupported by Node.
fs.fsyncSync(publishedFd);
} finally {
fs.closeSync(publishedFd);
}
fsyncDirectoryEntry(path.dirname(destination));
}
function fsyncReusableSnapshot(snapshotPath: string): void {
const fd = fs.openSync(snapshotPath, "r+");
try {
fs.fsyncSync(fd);
} finally {
fs.closeSync(fd);
}
}
type SqlJsSnapshotClone = {
run(sql: string): void;
export(): Uint8Array;
close(): void;
};
const SQLITE_HEADER_MIN_BYTES = 100;
const SQLITE_HEADER_MAGIC = "SQLite format 3\0";
const SQLITE_CHANGE_COUNTER_OFFSET = 24;
const SQLITE_VERSION_VALID_FOR_OFFSET = 92;
const SQLITE_STANDALONE_CHANGE_COUNTER = 1;
function exportCanonicalSqlJsSnapshot(raw: { export: () => Uint8Array }): Buffer {
const RawDatabase = (
raw as unknown as { constructor: new (data: Uint8Array) => SqlJsSnapshotClone }
).constructor;
let clone: SqlJsSnapshotClone | null = null;
try {
// A rolled-back sql.js SAVEPOINT can leave SQLite's physical change counter advanced
// even though every logical row/schema change was undone. Canonicalize only a detached
// clone: VACUUM removes rollback-only page artifacts without touching the live database.
clone = new RawDatabase(raw.export());
clone.run("VACUUM");
const canonical = Buffer.from(clone.export());
if (
canonical.length < SQLITE_HEADER_MIN_BYTES ||
canonical.subarray(0, SQLITE_HEADER_MAGIC.length).toString("binary") !== SQLITE_HEADER_MAGIC
) {
throw new Error("sql.js export did not produce a valid SQLite file header");
}
// SQLite file-header offsets 24 and 92 are the change counter and
// version-valid-for number. VACUUM keeps the two equal, but seeds them from the
// source image, so an otherwise identical rolled-back retry still gets a different
// byte hash. A standalone snapshot has no open readers to invalidate; assigning the
// same stable value to both fields preserves a valid/restorable header while making
// the complete canonical image deterministic.
canonical.writeUInt32BE(SQLITE_STANDALONE_CHANGE_COUNTER, SQLITE_CHANGE_COUNTER_OFFSET);
canonical.writeUInt32BE(SQLITE_STANDALONE_CHANGE_COUNTER, SQLITE_VERSION_VALID_FOR_OFFSET);
return canonical;
} finally {
clone?.close();
}
}
function writeSqlJsSnapshot(raw: { export: () => Uint8Array }, tempPath: string): void {
let fd: number | null = null;
try {
fd = fs.openSync(tempPath, "wx");
fs.writeFileSync(fd, exportCanonicalSqlJsSnapshot(raw));
fs.fsyncSync(fd);
fs.closeSync(fd);
fd = null;
} catch (error: unknown) {
if (fd !== null) {
try {
fs.closeSync(fd);
} catch {
// The original snapshot error remains authoritative.
}
}
throw error;
}
}
function cleanupOwnedSnapshotTemp(tempDir: string | null, tempPath: string | null): void {
if (!tempDir || !fs.existsSync(tempDir)) return;
try {
// `tempDir` comes only from mkdtempSync below. Removing that exact owned directory
// lets Node retry Windows/AV EBUSY and EPERM failures without touching canonical backups.
fs.rmSync(tempDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 25 });
} catch (error: unknown) {
const message = error instanceof Error ? error.message : String(error);
console.warn(
`[Migration] Failed to remove owned snapshot temp directory` +
`${tempPath ? ` (${tempPath})` : ""}: ${message}`
);
}
}
/**
* Create a synchronous pre-migration snapshot.
*
* Native SQLite drivers use VACUUM INTO. sql.js has an in-memory VFS, so a host
* path passed to VACUUM INTO is not writable; export its current database image
* directly instead. The SHA-256 content address lives in the first portion of the
* canonical `db_<snapshot-id>_<reason>.sqlite` shape, preserving reason parsing while
* making unchanged retries an O(1) lookup even with tens of thousands of old backups.
* Work happens inside an exclusively-created
* temp directory, so failure cleanup has exact ownership. Publication uses an atomic,
* no-overwrite hard link. If the filesystem cannot provide that primitive, the caller
* fails closed instead of exposing a partial canonical `.sqlite` file. A content hash
* reuses an identical prior snapshot, so repeated zero-progress startups retain one
* restore point for that database state without ever deleting a published backup.
*/
export function createPreMigrationBackup(db: SqliteAdapter): PreMigrationBackupReceipt | null {
let backupPath: string | null = null;
let tempPath: string | null = null;
let tempDir: string | null = null;
try {
const sqliteFile = db.name;
if (!sqliteFile || sqliteFile === ":memory:") return null;
const backupDir = path.join(path.dirname(sqliteFile), "db_backups");
if (!fs.existsSync(backupDir)) {
fs.mkdirSync(backupDir, { recursive: true });
fsyncDirectoryEntry(path.dirname(backupDir));
}
tempDir = fs.mkdtempSync(path.join(backupDir, ".migration-snapshot-"));
tempPath = path.join(tempDir, "snapshot.sqlite");
if (db.driver === "sql.js") {
const raw = db.raw as { export?: () => Uint8Array } | null;
if (!raw || typeof raw.export !== "function") {
throw new Error("sql.js adapter does not expose database export()");
}
writeSqlJsSnapshot(raw as { export: () => Uint8Array }, tempPath);
} else {
const escapedTempPath = tempPath.replace(/'/g, "''");
const snapshotDb = tryOpenSync(sqliteFile, { readonly: true, fileMustExist: true });
if (!snapshotDb) {
throw new Error("no synchronous read-only SQLite driver is available for snapshotting");
}
try {
snapshotDb.exec(`VACUUM INTO '${escapedTempPath}'`);
} finally {
snapshotDb.close();
}
const fd = fs.openSync(tempPath, "r+");
try {
fs.fsyncSync(fd);
} finally {
fs.closeSync(fd);
}
}
const sha256 = hashFileSync(tempPath);
backupPath = path.join(backupDir, `db_state-${sha256}_pre-migration.sqlite`);
const reusable = getReusablePreMigrationBackup(backupPath, sha256);
if (reusable) {
fsyncReusableSnapshot(reusable.path);
fsyncDirectoryEntry(backupDir);
cleanupOwnedSnapshotTemp(tempDir, tempPath);
tempDir = null;
tempPath = null;
console.log(`[Migration] Reusing identical pre-migration backup: ${reusable.path}`);
return reusable;
}
try {
publishSnapshotWithoutOverwrite(tempPath, backupPath);
} catch (error: unknown) {
if ((error as NodeJS.ErrnoException | null)?.code !== "EEXIST") throw error;
const racedReusable = getReusablePreMigrationBackup(backupPath, sha256);
if (!racedReusable) throw error;
fsyncReusableSnapshot(racedReusable.path);
fsyncDirectoryEntry(backupDir);
cleanupOwnedSnapshotTemp(tempDir, tempPath);
tempDir = null;
tempPath = null;
console.log(`[Migration] Reusing concurrently published backup: ${racedReusable.path}`);
return racedReusable;
}
cleanupOwnedSnapshotTemp(tempDir, tempPath);
tempDir = null;
tempPath = null;
console.log(`[Migration] Pre-migration backup created: ${backupPath}`);
return { path: backupPath, sha256 };
} catch (error: unknown) {
// Never unlink a canonical backup here: publication may have failed because another
// actor created it first. The exclusive temp directory is the only cleanup authority.
cleanupOwnedSnapshotTemp(tempDir, tempPath);
const message = error instanceof Error ? error.message : String(error);
console.warn(`[Migration] Failed to create pre-migration backup: ${message}`);
throw new Error(
`[Migration] Refusing to migrate an existing database without a durable snapshot. ` +
`Snapshot creation failed: ${message}. The DATA_DIR filesystem must support atomic ` +
`no-overwrite hard links, durable file synchronization, and directory synchronization ` +
`where the platform exposes it.`,
{ cause: error instanceof Error ? error : undefined }
);
}
}

248
src/lib/db/migrationRunner/schemaState.ts Normal file
View File

@@ -0,0 +1,248 @@
import type { SqliteAdapter } from "../adapters/types";
import {
INITIAL_SCHEMA_SENTINELS,
LEGACY_VERSION_SLOT_MIGRATIONS,
PHYSICAL_SCHEMA_SENTINELS,
RENAMED_MIGRATION_COMPATIBILITY,
} from "./constants";
import { migrationConsole as console } from "./logger";
type MigrationFile = { version: string; name: string; path: string };
export function hasTable(db: SqliteAdapter, tableName: string): boolean {
const row = db
.prepare("SELECT name FROM sqlite_master WHERE type IN ('table', 'view') AND name = ?")
.get(tableName) as { name?: string } | undefined;
return Boolean(row?.name);
}
export function hasPhysicalTable(db: SqliteAdapter, tableName: string): boolean {
const row = db
.prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?")
.get(tableName) as { name?: string } | undefined;
return Boolean(row?.name);
}
export function hasColumn(db: SqliteAdapter, tableName: string, columnName: string): boolean {
const columns = db.prepare(`PRAGMA table_info(${tableName})`).all() as Array<{ name?: string }>;
return columns.some((column) => column.name === columnName);
}
export function inferPhysicalSchemaBaseline(db: SqliteAdapter): {
version: string;
description: string;
} | null {
for (const sentinel of PHYSICAL_SCHEMA_SENTINELS) {
if (hasTable(db, sentinel.tableName)) {
return {
version: sentinel.version,
description: sentinel.description,
};
}
}
const hasInitialSchema = INITIAL_SCHEMA_SENTINELS.every((tableName) => hasTable(db, tableName));
if (hasInitialSchema) {
return {
version: "001",
description: "initial schema tables",
};
}
return null;
}
export function getPlausiblePendingCount(files: MigrationFile[], baselineVersion: string): number {
const baseline = Number.parseInt(baselineVersion, 10);
return files.filter((file) => Number.parseInt(file.version, 10) > baseline).length;
}
/**
* Detect migration name mismatches — when a migration version number
* has been reused/renumbered with a different name. This is a strong signal
* that the migration tracking is corrupted or migrations were renumbered.
*/
export function detectNameMismatches(
appliedRecords: Array<{ version: string; name: string }>,
files: MigrationFile[]
): Array<{ version: string; appliedName: string; diskName: string }> {
const appliedByName = new Map(appliedRecords.map((record) => [record.version, record.name]));
const mismatches: Array<{ version: string; appliedName: string; diskName: string }> = [];
for (const file of files) {
const appliedName = appliedByName.get(file.version);
if (appliedName && appliedName !== file.name) {
mismatches.push({
version: file.version,
appliedName,
diskName: file.name,
});
}
}
return mismatches;
}
export function reconcileRenumberedMigrations(db: SqliteAdapter, files: MigrationFile[]): boolean {
let repaired = false;
for (const compatibility of RENAMED_MIGRATION_COMPATIBILITY) {
const hasTargetFile = files.some(
(file) => file.version === compatibility.toVersion && file.name === compatibility.toName
);
const hasSourceFile = files.some(
(file) => file.version === compatibility.fromVersion && file.name !== compatibility.fromName
);
if (!hasTargetFile || !hasSourceFile) {
continue;
}
const legacyRow = db
.prepare("SELECT version, name FROM _omniroute_migrations WHERE version = ? AND name = ?")
.get(compatibility.fromVersion, compatibility.fromName) as
{ version: string; name: string } | undefined;
if (!legacyRow) {
continue;
}
const targetRow = db
.prepare("SELECT version, name FROM _omniroute_migrations WHERE version = ?")
.get(compatibility.toVersion) as { version: string; name: string } | undefined;
const isSameSlotReplacement = compatibility.fromVersion === compatibility.toVersion;
if (targetRow && !isSameSlotReplacement && targetRow.name !== compatibility.toName) {
throw new Error(
`[Migration] Cannot reconcile ${compatibility.fromVersion}_${compatibility.fromName}: ` +
`target version ${compatibility.toVersion} is occupied by unknown migration ` +
`"${targetRow.name}" (expected "${compatibility.toName}").`
);
}
const applyRepair = db.transaction(() => {
if (targetRow) {
db.prepare("DELETE FROM _omniroute_migrations WHERE version = ? AND name = ?").run(
compatibility.fromVersion,
compatibility.fromName
);
} else {
db.prepare(
"UPDATE _omniroute_migrations SET version = ?, name = ? WHERE version = ? AND name = ?"
).run(
compatibility.toVersion,
compatibility.toName,
compatibility.fromVersion,
compatibility.fromName
);
}
});
applyRepair();
repaired = true;
console.warn(
`[Migration] Reconciled renamed migration ${compatibility.fromVersion}_${compatibility.fromName} ` +
`to ${compatibility.toVersion}_${compatibility.toName} to preserve pending migrations.`
);
// After the compat rewrite, verify the old version slot is now free.
// A residual row (from a failed prior run, manual intervention, or edge-case
// UPDATE conflict) at the old version would shadow a NEW migration file
// placed at that version number — e.g. 028_create_files_and_batches.sql
// would be skipped because getAppliedVersions() still sees version "028".
const residualRow = db
.prepare("SELECT version, name FROM _omniroute_migrations WHERE version = ?")
.get(compatibility.fromVersion) as { version: string; name: string } | undefined;
if (residualRow) {
console.warn(
`[Migration] ⚠️ Residual row at version ${compatibility.fromVersion} ` +
`(name: "${residualRow.name}") still present after compat rewrite — ` +
`removing to unblock new migration at this version slot.`
);
db.prepare("DELETE FROM _omniroute_migrations WHERE version = ?").run(
compatibility.fromVersion
);
}
}
return repaired;
}
export function rehomeLegacyVersionSlotMigrations(
db: SqliteAdapter,
files: MigrationFile[]
): boolean {
let repaired = false;
const diskNamesByVersion = new Map(files.map((file) => [file.version, file.name]));
for (const legacy of LEGACY_VERSION_SLOT_MIGRATIONS) {
const diskName = diskNamesByVersion.get(legacy.version);
if (!diskName || diskName === legacy.name) {
continue;
}
const legacyRow = db
.prepare("SELECT version, name FROM _omniroute_migrations WHERE version = ? AND name = ?")
.get(legacy.version, legacy.name) as { version: string; name: string } | undefined;
if (!legacyRow) {
continue;
}
const legacyVersion = `legacy-${legacy.version}-${legacy.name}`;
const applyRepair = db.transaction(() => {
const existingLegacyRow = db
.prepare("SELECT version FROM _omniroute_migrations WHERE version = ?")
.get(legacyVersion) as { version: string } | undefined;
if (existingLegacyRow) {
db.prepare("DELETE FROM _omniroute_migrations WHERE version = ? AND name = ?").run(
legacy.version,
legacy.name
);
return;
}
db.prepare("UPDATE _omniroute_migrations SET version = ? WHERE version = ? AND name = ?").run(
legacyVersion,
legacy.version,
legacy.name
);
});
applyRepair();
repaired = true;
console.warn(
`[Migration] Rehomed legacy migration ${legacy.version}_${legacy.name} ` +
`to ${legacyVersion} so current ${legacy.version}_${diskName} can apply.`
);
}
return repaired;
}
export function hasLedgerRepairCandidates(db: SqliteAdapter, files: MigrationFile[]): boolean {
const diskNamesByVersion = new Map(files.map((file) => [file.version, file.name]));
for (const legacy of LEGACY_VERSION_SLOT_MIGRATIONS) {
const diskName = diskNamesByVersion.get(legacy.version);
if (!diskName || diskName === legacy.name) continue;
const row = db
.prepare("SELECT 1 FROM _omniroute_migrations WHERE version = ? AND name = ?")
.get(legacy.version, legacy.name);
if (row) return true;
}
for (const compatibility of RENAMED_MIGRATION_COMPATIBILITY) {
const hasTargetFile = files.some(
(file) => file.version === compatibility.toVersion && file.name === compatibility.toName
);
const hasSourceFile = files.some(
(file) => file.version === compatibility.fromVersion && file.name !== compatibility.fromName
);
if (!hasTargetFile || !hasSourceFile) continue;
const row = db
.prepare("SELECT 1 FROM _omniroute_migrations WHERE version = ? AND name = ?")
.get(compatibility.fromVersion, compatibility.fromName);
if (row) return true;
}
return false;
}