From abcd1af062f7aa0187b3bc16071770f6b86e271c Mon Sep 17 00:00:00 2001 From: zhiiw Date: Mon, 3 Aug 2026 19:32:18 +0800 Subject: [PATCH] fix(storage): retry concurrent WAL initialization --- .../sqlite-recovery-concurrency-child.ts | 28 +++++++---- .../sqlite-recovery-concurrency.test.ts | 44 +++++++++++++++-- packages/storage/src/sqlite-runtime-schema.ts | 49 +++++++++++++++++-- 3 files changed, 103 insertions(+), 18 deletions(-) diff --git a/packages/storage/src/__tests__/fixtures/sqlite-recovery-concurrency-child.ts b/packages/storage/src/__tests__/fixtures/sqlite-recovery-concurrency-child.ts index 440ac7a7c1..2f8378920e 100644 --- a/packages/storage/src/__tests__/fixtures/sqlite-recovery-concurrency-child.ts +++ b/packages/storage/src/__tests__/fixtures/sqlite-recovery-concurrency-child.ts @@ -1,4 +1,5 @@ import { existsSync, writeSync } from 'node:fs'; +import { dirname } from 'node:path'; import { createRuntimeBoundaryCursor, runtimePrefixSegment, @@ -7,6 +8,7 @@ import { type WorkspaceBaselineAuthorityInput, } from '@maka/core'; import { createSqliteRuntimeStore } from '../../sqlite-runtime-store.js'; +import { acquireOperationalStateDatabase } from '../../operational-state-store.js'; import { commitWorkspaceBaselineInternal } from '../../workspace-version-authority-internal.js'; const mode = requiredEnv('MAKA_SQLITE_RECOVERY_CONCURRENCY_MODE'); @@ -20,28 +22,33 @@ while (!existsSync(startPath)) { } let store: ReturnType | undefined; +let operationalLease: ReturnType | undefined; try { - store = createSqliteRuntimeStore(dbPath); + if (mode === 'operational_open_only') { + operationalLease = acquireOperationalStateDatabase(dirname(dbPath)); + } else { + store = createSqliteRuntimeStore(dbPath); + } writeSync(1, 'OPENED\n'); - if (mode === 'open_only') { + if (mode === 'open_only' || mode === 'operational_open_only') { if (!stopPath) throw new Error('Missing MAKA_SQLITE_RECOVERY_CONCURRENCY_STOP'); while (!existsSync(stopPath)) { Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 5); } } else if (mode === 'completed') { - await store.commitToolRecoveryBundle(completedBundle()); + await store!.commitToolRecoveryBundle(completedBundle()); } else if (mode === 'parked') { - await store.commitToolRecoveryBundle(parkedBundle()); + await store!.commitToolRecoveryBundle(parkedBundle()); } else if (mode === 'rebuild') { - await store.rebuildToolProjectionsFromRuntimeEvents(); + await store!.rebuildToolProjectionsFromRuntimeEvents(); } else if (mode === 'workspace_baseline_a' || mode === 'workspace_baseline_b') { const result = await commitWorkspaceBaselineInternal( - store, + store!, workspaceBaselineInput(mode === 'workspace_baseline_b' ? 'b' : 'a'), ); writeSync(1, `BASELINE ${result.created ? 'created' : 'existing'}\n`); } else if (mode === 'append_source') { - await store.ensureTerminalRuntimeEventDurable('session-1', 'run-1', { + await store!.ensureTerminalRuntimeEventDurable('session-1', 'run-1', { ...baseEvent('concurrent-source-terminal', 3), status: 'failed', actions: { @@ -51,7 +58,7 @@ try { }); writeSync(1, 'APPEND committed\n'); } else if (mode === 'append_target') { - await store.appendRuntimeEvent('session-1', 'fixed-target-run', { + await store!.appendRuntimeEvent('session-1', 'fixed-target-run', { id: `target-event-${process.pid}`, sessionId: 'session-1', invocationId: 'fixed-target-invocation', @@ -66,7 +73,7 @@ try { writeSync(1, 'TARGET_APPEND committed\n'); } else if (mode === 'claim' || mode === 'claim_fixed_target' || mode === 'claim_nonterminal') { const sourceRunId = mode === 'claim_nonterminal' ? 'run-1' : 'continuation-source-run'; - const prefix = await store.readImmutableRuntimePrefix({ + const prefix = await store!.readImmutableRuntimePrefix({ sessionId: 'session-1', runId: sourceRunId, ...(mode === 'claim_nonterminal' ? { upToEventSeq: 2 } : {}), @@ -87,7 +94,7 @@ try { runId: `run-${process.pid}`, turnId: `turn-${process.pid}`, }; - const result = await store.claimContinuation({ + const result = await store!.claimContinuation({ claim: { protocol: 'continuation_claim_v1', claimId: `claim-${process.pid}`, @@ -137,6 +144,7 @@ try { process.exitCode = 2; } finally { store?.close(); + operationalLease?.close(); } function completedBundle() { diff --git a/packages/storage/src/__tests__/sqlite-recovery-concurrency.test.ts b/packages/storage/src/__tests__/sqlite-recovery-concurrency.test.ts index 38ff83f867..512407c15f 100644 --- a/packages/storage/src/__tests__/sqlite-recovery-concurrency.test.ts +++ b/packages/storage/src/__tests__/sqlite-recovery-concurrency.test.ts @@ -162,6 +162,40 @@ describe('SQLite recovery authority multi-process races', () => { }); }); + it('allows concurrent operational owners to initialize the same fresh WAL database', async () => { + for (let round = 0; round < 12; round += 1) { + const root = await mkdtemp(join(tmpdir(), 'maka-operational-fresh-open-race-')); + const dbPath = join(root, 'runtime.sqlite'); + const startPath = join(root, 'start'); + try { + const results = await runOpenWorkers(dbPath, startPath, 'operational_open_only'); + assert.deepEqual( + results.map(({ code }) => code), + [0, 0], + `fresh concurrent operational open failed in round ${round + 1}: ${JSON.stringify(results)}`, + ); + + const database = new DatabaseSync(dbPath, { readOnly: true }); + try { + assert.equal( + (database.prepare('PRAGMA journal_mode').get() as { journal_mode: string }) + .journal_mode, + 'wal', + ); + assert.equal( + (database.prepare('PRAGMA user_version').get() as { user_version: number }) + .user_version, + SQLITE_RUNTIME_SCHEMA_VERSION, + ); + } finally { + database.close(); + } + } finally { + await rm(root, { recursive: true, force: true }); + } + } + }); + it('makes an exact concurrent workspace baseline open idempotent', async () => { await withPreparedDatabase(async ({ dbPath, startPath }) => { const results = await runWorkers(dbPath, startPath, [ @@ -323,10 +357,14 @@ async function runWorkers( } } -async function runOpenWorkers(dbPath: string, startPath: string): Promise { +async function runOpenWorkers( + dbPath: string, + startPath: string, + mode = 'open_only', +): Promise { const stopPath = `${startPath}.stop`; - const workers = ['open_only', 'open_only'].map((mode) => - startWorker(dbPath, startPath, mode, stopPath), + const workers = [mode, mode].map((workerMode) => + startWorker(dbPath, startPath, workerMode, stopPath), ); try { await withTimeout( diff --git a/packages/storage/src/sqlite-runtime-schema.ts b/packages/storage/src/sqlite-runtime-schema.ts index cc6ae4d37c..20392ff87b 100644 --- a/packages/storage/src/sqlite-runtime-schema.ts +++ b/packages/storage/src/sqlite-runtime-schema.ts @@ -7,6 +7,9 @@ export const RUNTIME_CONTINUATION_AUTHORITY_CAPABILITY = 'runtime_continuation_a export const RUNTIME_CONTINUATION_AUTHORITY_CAPABILITY_VERSION = 1; export const RUNTIME_WORKSPACE_VERSION_AUTHORITY_CAPABILITY = 'runtime_workspace_version_authority'; export const RUNTIME_WORKSPACE_VERSION_AUTHORITY_CAPABILITY_VERSION = 1; +const SQLITE_INITIALIZATION_BUSY_TIMEOUT_MS = 5_000; +const SQLITE_INITIALIZATION_RETRY_DELAY_MS = 10; +const initializationRetryGate = new Int32Array(new SharedArrayBuffer(4)); const MIGRATIONS: ReadonlyMap = new Map([ [ @@ -247,11 +250,8 @@ export function configureSqliteRuntimeDatabase(db: DatabaseSync): void { // Bound lock acquisition before touching persistent journal state. WAL mode is // database-persistent, so established workspaces only need to verify it rather // than making every concurrent opener execute the setting form of the pragma. - db.exec('PRAGMA busy_timeout = 5000'); - const journalMode = readJournalMode(db); - if (journalMode !== 'wal') { - db.exec('PRAGMA journal_mode = WAL'); - } + db.exec(`PRAGMA busy_timeout = ${SQLITE_INITIALIZATION_BUSY_TIMEOUT_MS}`); + ensureWalJournalMode(db); db.exec('PRAGMA synchronous = FULL'); db.exec('PRAGMA foreign_keys = ON'); } @@ -307,6 +307,45 @@ function readJournalMode(db: DatabaseSync): string { return row.journal_mode.toLowerCase(); } +function ensureWalJournalMode(db: DatabaseSync): void { + const deadline = Date.now() + SQLITE_INITIALIZATION_BUSY_TIMEOUT_MS; + while (true) { + try { + const journalMode = readJournalMode(db); + if (journalMode === 'wal' || journalMode === 'memory') return; + db.exec('PRAGMA journal_mode = WAL'); + const configuredMode = readJournalMode(db); + if (configuredMode !== 'wal') { + throw new Error(`SQLite runtime requires WAL journal mode, received ${configuredMode}`); + } + return; + } catch (error) { + if (!isSqliteBusy(error) || Date.now() >= deadline) throw error; + Atomics.wait( + initializationRetryGate, + 0, + 0, + Math.min(SQLITE_INITIALIZATION_RETRY_DELAY_MS, Math.max(1, deadline - Date.now())), + ); + } + } +} + +function isSqliteBusy(error: unknown): boolean { + if (!(error instanceof Error)) return false; + const sqliteError = error as Error & { + code?: unknown; + errcode?: unknown; + errstr?: unknown; + }; + return ( + sqliteError.errcode === 5 || + sqliteError.code === 'SQLITE_BUSY' || + sqliteError.errstr === 'database is locked' || + /database (?:is )?(?:locked|busy)/i.test(sqliteError.message) + ); +} + function rollback(db: DatabaseSync): void { try { db.exec('ROLLBACK');