feat(migration): add verified import and rollback workflow
This commit is contained in:
chick
@@ -1,8 +1,36 @@
|
||||
import Database from 'better-sqlite3';
|
||||
import { chmod, copyFile, lstat, mkdir, open, rename, rm, stat } from 'node:fs/promises';
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { createHash, randomUUID } from 'node:crypto';
|
||||
import type { FileHandle } from 'node:fs/promises';
|
||||
import {
|
||||
chmod,
|
||||
copyFile,
|
||||
lstat,
|
||||
mkdir,
|
||||
open,
|
||||
readFile,
|
||||
rename,
|
||||
rm,
|
||||
stat,
|
||||
writeFile,
|
||||
} from 'node:fs/promises';
|
||||
import { dirname, resolve } from 'node:path';
|
||||
import { beginRestore, getManagedDatabaseIdentity, type SqliteDatabase } from './database.js';
|
||||
import { MIGRATIONS, migrateDatabase } from './migrations.js';
|
||||
|
||||
export interface DatabaseArtifact {
|
||||
readonly path: string;
|
||||
readonly sha256: string;
|
||||
}
|
||||
|
||||
export interface RollbackDrillRecord {
|
||||
readonly kind: 'rollback-drill';
|
||||
readonly snapshotSha256: string;
|
||||
readonly integrity: 'ok';
|
||||
readonly schemaDigest: string;
|
||||
readonly migrationsDigest: string;
|
||||
readonly rowCounts: Readonly<Record<string, number>>;
|
||||
readonly reconciliationDigest: string;
|
||||
}
|
||||
|
||||
async function assertRegularFile(path: string, label: string): Promise<void> {
|
||||
const details = await lstat(path);
|
||||
@@ -34,14 +62,102 @@ async function assertValidDatabase(path: string): Promise<void> {
|
||||
try {
|
||||
database = new Database(path, { fileMustExist: true, readonly: true });
|
||||
const integrity = database.pragma('integrity_check', { simple: true });
|
||||
if (integrity !== 'ok') throw new Error(`Backup integrity check failed: ${String(integrity)}`);
|
||||
if (integrity !== 'ok')
|
||||
throw new Error(`Database integrity check failed: ${String(integrity)}`);
|
||||
assertExpectedSchema(database);
|
||||
const migrations = database
|
||||
.prepare('SELECT id,name,checksum FROM schema_migrations ORDER BY id')
|
||||
.all() as { id: number; name: string; checksum: string }[];
|
||||
if (migrations.length !== MIGRATIONS.length)
|
||||
throw new Error('Application migration set incomplete');
|
||||
for (const [index, migration] of MIGRATIONS.entries()) {
|
||||
const actual = migrations[index];
|
||||
const checksum = createHash('sha256')
|
||||
.update(JSON.stringify(migration.statements))
|
||||
.digest('hex');
|
||||
if (
|
||||
!actual ||
|
||||
actual.id !== migration.id ||
|
||||
actual.name !== migration.name ||
|
||||
actual.checksum !== checksum
|
||||
)
|
||||
throw new Error(`Application migration ${migration.id} mismatch`);
|
||||
}
|
||||
const violations = database.pragma('foreign_key_check') as unknown[];
|
||||
if (violations.length) throw new Error('Application database has foreign key violations');
|
||||
} catch (error) {
|
||||
throw new Error('Backup is not a valid SQLite database', { cause: error });
|
||||
throw new Error(
|
||||
`Artifact is not a valid application SQLite database: ${(error as Error).message}`,
|
||||
{
|
||||
cause: error,
|
||||
},
|
||||
);
|
||||
} finally {
|
||||
database?.close();
|
||||
}
|
||||
}
|
||||
|
||||
interface SchemaObject {
|
||||
readonly type: string;
|
||||
readonly name: string;
|
||||
readonly tableName: string;
|
||||
readonly sql: string | null;
|
||||
}
|
||||
|
||||
function readSchema(database: Database.Database): SchemaObject[] {
|
||||
return database
|
||||
.prepare(
|
||||
`SELECT type,name,tbl_name AS tableName,sql
|
||||
FROM sqlite_master
|
||||
WHERE name NOT LIKE 'sqlite_%'
|
||||
ORDER BY type,name`,
|
||||
)
|
||||
.all()
|
||||
.map((object) => {
|
||||
const value = object as SchemaObject;
|
||||
return {
|
||||
...value,
|
||||
sql: value.sql?.trim().replaceAll(/\s+/g, ' ') ?? null,
|
||||
};
|
||||
}) as SchemaObject[];
|
||||
}
|
||||
|
||||
let expectedSchema: readonly SchemaObject[] | undefined;
|
||||
|
||||
function getExpectedSchema(): readonly SchemaObject[] {
|
||||
if (expectedSchema) return expectedSchema;
|
||||
const reference = new Database(':memory:');
|
||||
try {
|
||||
migrateDatabase(reference);
|
||||
expectedSchema = Object.freeze(readSchema(reference).map((object) => Object.freeze(object)));
|
||||
return expectedSchema;
|
||||
} finally {
|
||||
reference.close();
|
||||
}
|
||||
}
|
||||
|
||||
function assertExpectedSchema(database: Database.Database): void {
|
||||
const actual = readSchema(database);
|
||||
const expected = getExpectedSchema();
|
||||
if (JSON.stringify(actual) !== JSON.stringify(expected))
|
||||
throw new Error('Application schema object set mismatch');
|
||||
}
|
||||
|
||||
async function digestFile(path: string): Promise<string> {
|
||||
return createHash('sha256')
|
||||
.update(await readFile(path))
|
||||
.digest('hex');
|
||||
}
|
||||
|
||||
async function digestFileHandle(handle: FileHandle): Promise<string> {
|
||||
const details = await handle.stat();
|
||||
const bytes = Buffer.alloc(details.size);
|
||||
const { bytesRead } = await handle.read(bytes, 0, bytes.length, 0);
|
||||
if (bytesRead !== bytes.length)
|
||||
throw new Error('Staged restore candidate could not be read fully');
|
||||
return createHash('sha256').update(bytes).digest('hex');
|
||||
}
|
||||
|
||||
async function syncFile(path: string): Promise<void> {
|
||||
const handle = await open(path, 'r');
|
||||
try {
|
||||
@@ -60,21 +176,24 @@ async function syncDirectory(path: string): Promise<void> {
|
||||
}
|
||||
}
|
||||
|
||||
export async function backupDatabase(database: SqliteDatabase, backupPath: string): Promise<void> {
|
||||
export async function backupDatabase(
|
||||
database: SqliteDatabase,
|
||||
backupPath: string,
|
||||
): Promise<DatabaseArtifact> {
|
||||
const source = getManagedDatabaseIdentity(database);
|
||||
const directory = dirname(backupPath);
|
||||
await mkdir(directory, { recursive: true });
|
||||
try {
|
||||
const destination = await stat(backupPath);
|
||||
if (`${destination.dev}:${destination.ino}` === source.deviceInode) {
|
||||
if (`${destination.dev}:${destination.ino}` === source.deviceInode)
|
||||
throw new Error('Backup destination aliases the live database identity');
|
||||
}
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error;
|
||||
}
|
||||
const temporaryPath = `${backupPath}.${randomUUID()}.tmp`;
|
||||
try {
|
||||
await reservePrivateFile(temporaryPath);
|
||||
// SQLite's online backup API includes committed WAL pages in one consistent snapshot.
|
||||
await database.backup(temporaryPath);
|
||||
await chmod(temporaryPath, 0o600);
|
||||
await assertValidDatabase(temporaryPath);
|
||||
@@ -82,35 +201,240 @@ export async function backupDatabase(database: SqliteDatabase, backupPath: strin
|
||||
await rename(temporaryPath, backupPath);
|
||||
await chmod(backupPath, 0o600);
|
||||
await syncDirectory(directory);
|
||||
return Object.freeze({ path: backupPath, sha256: await digestFile(backupPath) });
|
||||
} finally {
|
||||
await rm(temporaryPath, { force: true });
|
||||
}
|
||||
}
|
||||
|
||||
export async function restoreDatabase(databasePath: string, backupPath: string): Promise<void> {
|
||||
const endRestore = beginRestore(databasePath);
|
||||
const directory = dirname(databasePath);
|
||||
const temporaryPath = `${databasePath}.${randomUUID()}.restore`;
|
||||
export async function prepareRestoreCandidate(
|
||||
backupPath: string,
|
||||
candidatePath: string,
|
||||
): Promise<DatabaseArtifact> {
|
||||
await assertDistinctFiles(candidatePath, backupPath);
|
||||
await assertValidDatabase(backupPath);
|
||||
await mkdir(dirname(candidatePath), { recursive: true });
|
||||
const temporaryPath = `${candidatePath}.${randomUUID()}.tmp`;
|
||||
try {
|
||||
await assertDistinctFiles(databasePath, backupPath);
|
||||
await assertValidDatabase(backupPath);
|
||||
await mkdir(directory, { recursive: true });
|
||||
await reservePrivateFile(temporaryPath);
|
||||
await copyFile(backupPath, temporaryPath);
|
||||
await chmod(temporaryPath, 0o600);
|
||||
await assertValidDatabase(temporaryPath);
|
||||
await syncFile(temporaryPath);
|
||||
|
||||
// Atomic replacement comes first: any failure before this point leaves live and sidecars intact.
|
||||
await rename(temporaryPath, databasePath);
|
||||
await chmod(databasePath, 0o600);
|
||||
await syncDirectory(directory);
|
||||
// No managed connection can exist under the exclusive gate; stale sidecars are now safe to remove.
|
||||
await rm(`${databasePath}-wal`, { force: true });
|
||||
await rm(`${databasePath}-shm`, { force: true });
|
||||
await syncDirectory(directory);
|
||||
await rename(temporaryPath, candidatePath);
|
||||
await syncDirectory(dirname(candidatePath));
|
||||
return Object.freeze({ path: candidatePath, sha256: await digestFile(candidatePath) });
|
||||
} finally {
|
||||
await rm(temporaryPath, { force: true });
|
||||
}
|
||||
}
|
||||
|
||||
export async function activateRestoreCandidate(
|
||||
databasePath: string,
|
||||
candidate: DatabaseArtifact,
|
||||
confirmedDigest: string,
|
||||
hooks: { readonly beforeActivation?: (stagedPath: string) => Promise<void> } = {},
|
||||
): Promise<void> {
|
||||
if (confirmedDigest !== candidate.sha256) throw new Error('Restore candidate digest mismatch');
|
||||
const endRestore = beginRestore(databasePath);
|
||||
const directory = dirname(databasePath);
|
||||
const stagedPath = `${databasePath}.${randomUUID()}.restore`;
|
||||
try {
|
||||
await assertDistinctFiles(databasePath, candidate.path);
|
||||
await mkdir(directory, { recursive: true });
|
||||
await reservePrivateFile(stagedPath);
|
||||
const before = await stat(candidate.path);
|
||||
await copyFile(candidate.path, stagedPath);
|
||||
const after = await stat(candidate.path);
|
||||
if (
|
||||
before.dev !== after.dev ||
|
||||
before.ino !== after.ino ||
|
||||
before.size !== after.size ||
|
||||
before.mtimeMs !== after.mtimeMs
|
||||
)
|
||||
throw new Error('Restore candidate changed while being staged');
|
||||
// Deny path-based writers after population, then hold the exact inode open through
|
||||
// validation and the final path identity check.
|
||||
await chmod(stagedPath, 0o400);
|
||||
const stagedHandle = await open(stagedPath, 'r');
|
||||
try {
|
||||
const stagedIdentity = await stagedHandle.stat();
|
||||
const stagedDigest = await digestFileHandle(stagedHandle);
|
||||
if (stagedDigest !== candidate.sha256) throw new Error('Restore candidate digest mismatch');
|
||||
await assertValidDatabase(stagedPath);
|
||||
if ((await digestFileHandle(stagedHandle)) !== stagedDigest)
|
||||
throw new Error('Staged restore candidate changed during validation');
|
||||
await stagedHandle.sync();
|
||||
await activateStagedDatabase(
|
||||
databasePath,
|
||||
stagedPath,
|
||||
stagedHandle,
|
||||
stagedIdentity,
|
||||
confirmedDigest,
|
||||
hooks.beforeActivation,
|
||||
);
|
||||
} finally {
|
||||
await stagedHandle.close();
|
||||
}
|
||||
} finally {
|
||||
await rm(stagedPath, { force: true });
|
||||
endRestore();
|
||||
}
|
||||
}
|
||||
|
||||
/** Internal compatibility helper. Public callers must use prepare + digest-confirmed activation. */
|
||||
export async function restoreDatabase(databasePath: string, backupPath: string): Promise<void> {
|
||||
await assertDistinctFiles(databasePath, backupPath);
|
||||
const candidatePath = `${databasePath}.${randomUUID()}.candidate`;
|
||||
try {
|
||||
const candidate = await prepareRestoreCandidate(backupPath, candidatePath);
|
||||
await activateRestoreCandidate(databasePath, candidate, candidate.sha256);
|
||||
} finally {
|
||||
await rm(candidatePath, { force: true });
|
||||
}
|
||||
}
|
||||
|
||||
interface RestoreMarker {
|
||||
readonly version: 1;
|
||||
readonly quarantinedSidecars: readonly string[];
|
||||
}
|
||||
|
||||
async function readPreviousRestoreMarker(markerPath: string): Promise<RestoreMarker | undefined> {
|
||||
try {
|
||||
await assertRegularFile(markerPath, 'Restore marker');
|
||||
const value = JSON.parse(await readFile(markerPath, 'utf8')) as Partial<RestoreMarker>;
|
||||
if (value.version !== 1 || !Array.isArray(value.quarantinedSidecars))
|
||||
throw new Error('Restore marker is invalid; manual recovery is required');
|
||||
return value as RestoreMarker;
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code === 'ENOENT') return undefined;
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async function writeRestoreMarker(markerPath: string, marker: RestoreMarker): Promise<void> {
|
||||
const temporary = `${markerPath}.${randomUUID()}.tmp`;
|
||||
try {
|
||||
await writeFile(temporary, JSON.stringify(marker), { flag: 'wx', mode: 0o600 });
|
||||
await syncFile(temporary);
|
||||
await rename(temporary, markerPath);
|
||||
await syncDirectory(dirname(markerPath));
|
||||
} finally {
|
||||
await rm(temporary, { force: true });
|
||||
}
|
||||
}
|
||||
|
||||
async function activateStagedDatabase(
|
||||
databasePath: string,
|
||||
stagedPath: string,
|
||||
stagedHandle: FileHandle,
|
||||
stagedIdentity: Awaited<ReturnType<FileHandle['stat']>>,
|
||||
confirmedDigest: string,
|
||||
beforeActivation?: (stagedPath: string) => Promise<void>,
|
||||
): Promise<void> {
|
||||
const directory = dirname(databasePath);
|
||||
const markerPath = `${databasePath}.restore-in-progress`;
|
||||
const previous = await readPreviousRestoreMarker(markerPath);
|
||||
const transactionId = randomUUID();
|
||||
const sidecars = [`${databasePath}-wal`, `${databasePath}-shm`];
|
||||
const quarantined = sidecars.map((path) => `${path}.${transactionId}.quarantine`);
|
||||
const allQuarantined = [...(previous?.quarantinedSidecars ?? []), ...quarantined];
|
||||
await writeRestoreMarker(markerPath, { version: 1, quarantinedSidecars: allQuarantined });
|
||||
try {
|
||||
// Make old WAL state unreachable before replacing the main file. The durable marker
|
||||
// forces managed opens to fail closed at every crash boundary.
|
||||
for (const [index, sidecar] of sidecars.entries()) {
|
||||
try {
|
||||
await rename(sidecar, quarantined[index]!);
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error;
|
||||
}
|
||||
}
|
||||
await syncDirectory(directory);
|
||||
await beforeActivation?.(stagedPath);
|
||||
if ((await digestFileHandle(stagedHandle)) !== confirmedDigest)
|
||||
throw new Error('Staged restore candidate digest changed before activation');
|
||||
const [held, path] = await Promise.all([stagedHandle.stat(), lstat(stagedPath)]);
|
||||
if (
|
||||
!path.isFile() ||
|
||||
path.isSymbolicLink() ||
|
||||
(path.mode & 0o222) !== 0 ||
|
||||
held.dev !== stagedIdentity.dev ||
|
||||
held.ino !== stagedIdentity.ino ||
|
||||
held.size !== stagedIdentity.size ||
|
||||
held.mtimeMs !== stagedIdentity.mtimeMs ||
|
||||
path.dev !== stagedIdentity.dev ||
|
||||
path.ino !== stagedIdentity.ino ||
|
||||
path.size !== stagedIdentity.size ||
|
||||
path.mtimeMs !== stagedIdentity.mtimeMs
|
||||
)
|
||||
throw new Error('Staged restore candidate identity or bytes changed before activation');
|
||||
await rename(stagedPath, databasePath);
|
||||
await chmod(databasePath, 0o600);
|
||||
await syncDirectory(directory);
|
||||
for (const path of allQuarantined) await rm(path, { force: true });
|
||||
await rm(markerPath);
|
||||
await syncDirectory(directory);
|
||||
} catch (error) {
|
||||
throw new Error(
|
||||
`Restore activation did not complete; ${markerPath} was retained to force recovery: ${(error as Error).message}`,
|
||||
{ cause: error },
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
export async function runRollbackDrill(
|
||||
backupPath: string,
|
||||
candidatePath: string,
|
||||
): Promise<RollbackDrillRecord> {
|
||||
const candidate = await prepareRestoreCandidate(backupPath, candidatePath);
|
||||
let database: Database.Database | undefined;
|
||||
try {
|
||||
database = new Database(candidate.path, { readonly: true, fileMustExist: true });
|
||||
const integrity = database.pragma('integrity_check', { simple: true });
|
||||
if (integrity !== 'ok') throw new Error('Rollback drill integrity check failed');
|
||||
const schema = database
|
||||
.prepare(
|
||||
"SELECT type,name,tbl_name,sql FROM sqlite_master WHERE name NOT LIKE 'sqlite_%' ORDER BY type,name",
|
||||
)
|
||||
.all();
|
||||
const tables = database
|
||||
.prepare(
|
||||
"SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name",
|
||||
)
|
||||
.all() as { name: string }[];
|
||||
const rowCounts: Record<string, number> = {};
|
||||
for (const { name } of tables) {
|
||||
const quoted = `"${name.replaceAll('"', '""')}"`;
|
||||
rowCounts[name] = (
|
||||
database.prepare(`SELECT COUNT(*) AS count FROM ${quoted}`).get() as { count: number }
|
||||
).count;
|
||||
}
|
||||
const schemaDigest = createHash('sha256').update(JSON.stringify(schema)).digest('hex');
|
||||
const migrations = database
|
||||
.prepare('SELECT id,name,checksum FROM schema_migrations ORDER BY id')
|
||||
.all();
|
||||
const migrationsDigest = createHash('sha256').update(JSON.stringify(migrations)).digest('hex');
|
||||
const reconciliationDigest = createHash('sha256')
|
||||
.update(
|
||||
JSON.stringify({
|
||||
snapshotSha256: candidate.sha256,
|
||||
schemaDigest,
|
||||
migrationsDigest,
|
||||
rowCounts,
|
||||
}),
|
||||
)
|
||||
.digest('hex');
|
||||
return Object.freeze({
|
||||
kind: 'rollback-drill',
|
||||
snapshotSha256: candidate.sha256,
|
||||
integrity: 'ok',
|
||||
schemaDigest,
|
||||
migrationsDigest,
|
||||
rowCounts: Object.freeze(rowCounts),
|
||||
reconciliationDigest,
|
||||
});
|
||||
} finally {
|
||||
database?.close();
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user