diff --git a/.changeset/ci-flakes-sqlite-session.md b/.changeset/ci-flakes-sqlite-session.md new file mode 100644 index 000000000..fe210895b --- /dev/null +++ b/.changeset/ci-flakes-sqlite-session.md @@ -0,0 +1,5 @@ +--- +"@agent-bundle/runtime": patch +--- + +Configure sqlite state storage under the busy timeout: `createSqliteStateDriver` now retries the `PRAGMA journal_mode = WAL` switch on the first open of a state file for up to `busyTimeoutMs` while another process holds the write lock (SQLite never routes that read-to-write lock upgrade through `busy_timeout`), so two processes opening one workspace-durable state file no longer fail with `unavailable` / `database is locked` during setup. Extended SQLite result codes (`SQLITE_BUSY_*`, `SQLITE_CORRUPT_*`) now map to the same typed `unavailable` / `corrupt` errors as their primary codes (#567) diff --git a/packages/agent-bundle/tests/mcp-session-service.test.ts b/packages/agent-bundle/tests/mcp-session-service.test.ts index f8b3761b0..9e29a80d3 100644 --- a/packages/agent-bundle/tests/mcp-session-service.test.ts +++ b/packages/agent-bundle/tests/mcp-session-service.test.ts @@ -18,6 +18,7 @@ import { McpSession, McpSessionError, McpSessionService, + type McpSessionTraceSubscription, } from '../src/dev/mcp-session/mcp-session-service.ts'; import type { ArtifactEpoch } from '../src/dev/types.ts'; import { pathTokens, type NormalizationTargetRegistry } from '../src/core/types.ts'; @@ -69,6 +70,30 @@ const textFrom = (value: { readonly content: readonly { readonly type: string }[ return content.text; }; +const isToolCallFrame = (message: unknown, name: string): boolean => + typeof message === 'object' + && message !== null + && (message as { readonly method?: unknown }).method === 'tools/call' + && (message as { readonly params?: { readonly name?: unknown } }).params?.name === name; + +/** + * Resolves once the session puts the `tools/call` for `name` on the wire. The + * request slot is admitted before the SDK sends, so from then on `cancel()` + * finds it — an ordering a fixed sleep can only approximate. + */ +const toolCallSent = (session: McpSession, name: string): Promise => { + const afterSequence = session.trace().entries.at(-1)?.sequence ?? 0; + let subscription: McpSessionTraceSubscription | undefined; + const sent = new Promise((resolvePromise) => { + subscription = session.subscribeTrace({ afterSequence }, (entry) => { + if ('kind' in entry && entry.kind === 'frame' && entry.direction === 'client' && isToolCallFrame(entry.message, name)) { + resolvePromise(); + } + }); + }); + return sent.finally(() => subscription?.unsubscribe()); +}; + const publishFixtureEpoch = async ( root: string, id: string, @@ -268,8 +293,9 @@ it('keeps one generated server and plugin-data directory bound to the selected e expect(restartedState.root).toBe(firstState.root); expect(restartedState.pid).not.toBe(firstState.pid); + const pendingSent = toolCallSent(session, 'hang'); const pending = session.callTool({ arguments: {}, name: 'hang', requestId: 'pending-hang' }); - await new Promise((resolvePromise) => setTimeout(resolvePromise, 25)); + await pendingSent; expect(session.cancel('pending-hang')).toBe(true); await expect(pending).rejects.toBeDefined(); @@ -1089,13 +1115,21 @@ it('fails admission, lifecycle, and service misuse closed with coded McpSessionE const epochStore = await publishFixtureEpoch(root, 'epoch-1'); const releases: Array<() => void> = []; const signals: AbortSignal[] = []; + // Each call the stub receives resolves the oldest admission waiter: by + // then the session has admitted the request and registered its signal. + const admissions: Array<() => void> = []; + const nextAdmission = (): Promise => new Promise((resolvePromise) => { + admissions.push(resolvePromise); + }); const service = new McpSessionService({ createClient: () => ({ callTool: async (_params: unknown, options?: { readonly signal?: AbortSignal }) => { if (options?.signal !== undefined) signals.push(options.signal); - await new Promise((resolvePromise) => { + const released = new Promise((resolvePromise) => { releases.push(resolvePromise); }); + admissions.shift()?.(); + await released; return { content: [] }; }, close: async () => undefined, @@ -1133,8 +1167,9 @@ it('fails admission, lifecycle, and service misuse closed with coded McpSessionE 'MCP session requestId must be nonempty.', ); + const firstAdmitted = nextAdmission(); const first = session.callTool({ arguments: {}, name: 'fixture', requestId: 'shared' }); - await new Promise((resolvePromise) => setTimeout(resolvePromise, 10)); + await firstAdmitted; expect(signals).toHaveLength(1); await expectSessionError( session.callTool({ arguments: {}, name: 'fixture', requestId: 'shared' }), @@ -1146,8 +1181,9 @@ it('fails admission, lifecycle, and service misuse closed with coded McpSessionE await expect(first).resolves.toEqual({ content: [] }); // Releasing the request slot aborts its controller and frees the id. expect(signals[0]?.aborted).toBe(true); + const reusedAdmitted = nextAdmission(); const reused = session.callTool({ arguments: {}, name: 'fixture', requestId: 'shared' }); - await new Promise((resolvePromise) => setTimeout(resolvePromise, 10)); + await reusedAdmitted; expect(signals).toHaveLength(2); releases.shift()?.(); await expect(reused).resolves.toEqual({ content: [] }); diff --git a/packages/rsc-runtime/src/state/sqlite.ts b/packages/rsc-runtime/src/state/sqlite.ts index 2bcb9089e..c9b4c2292 100644 --- a/packages/rsc-runtime/src/state/sqlite.ts +++ b/packages/rsc-runtime/src/state/sqlite.ts @@ -14,9 +14,11 @@ import { DatabaseSync, type StatementSync } from 'node:sqlite'; import { Context, + Duration, Effect, Exit, Layer, + Schedule, } from 'effect'; import { @@ -105,7 +107,9 @@ export interface SqliteStateDriverOptions { /** * SQLite lock wait budget per operation in milliseconds (default 5000). * Contending cross-process writers queue on the database lock for at most - * this long before the operation fails typed `unavailable`. + * this long before the operation fails typed `unavailable`. The same budget + * bounds how long the first open of a file waits for another process that + * is switching it into WAL. */ readonly busyTimeoutMs?: number; /** @@ -133,6 +137,33 @@ const SQLITE_CORRUPT = 11; const SQLITE_NOTADB = 26; const SQLITE_BUSY = 5; +/** + * `node:sqlite` reports the extended result code (`SQLITE_BUSY_RECOVERY`, + * `SQLITE_CORRUPT_INDEX`, ...); the primary code lives in the low byte. + */ +const primaryResultCode = (error: unknown): number | undefined => { + const errcode = (error as SqliteErrorShape | undefined)?.errcode; + return typeof errcode === 'number' ? errcode & 0xff : undefined; +}; + +const isSqliteBusy = (error: unknown): boolean => primaryResultCode(error) === SQLITE_BUSY; + +/** + * Longest pause between attempts to switch a rollback-mode database into WAL + * while another process holds its write lock (see `SqliteStore#initialize`). + */ +const WAL_SWITCH_RETRY_DELAY_MS = 5; + +/** + * Retries the WAL switch every {@link WAL_SWITCH_RETRY_DELAY_MS} (or the whole + * budget when that is shorter), and only while the next attempt still starts + * inside `busyTimeoutMs`: an open never outlives its budget by a pause. + */ +const walSwitchRetries = (busyTimeoutMs: number): Schedule.Schedule => + Schedule.spaced(Duration.millis(Math.min(WAL_SWITCH_RETRY_DELAY_MS, busyTimeoutMs))).pipe( + Schedule.while(({ duration, elapsed }) => elapsed + Duration.toMillis(duration) <= busyTimeoutMs), + ); + const mapSqliteError = ( definitionId: string, action: string, @@ -141,21 +172,22 @@ const mapSqliteError = ( ): AgentStateError | undefined => { if (error instanceof AgentStateError) return error; const shape = error as SqliteErrorShape; + const errcode = primaryResultCode(error); const sqliteError = - typeof shape?.errcode === 'number' + errcode !== undefined || (typeof shape?.code === 'string' && shape.code.startsWith('ERR_SQLITE')); if (!sqliteError && !(mapSystemError && typeof shape?.code === 'string')) { return undefined; } const detail = typeof shape.errstr === 'string' ? `: ${shape.errstr}` : ''; - if (shape.errcode === SQLITE_CORRUPT || shape.errcode === SQLITE_NOTADB) { + if (errcode === SQLITE_CORRUPT || errcode === SQLITE_NOTADB) { return new AgentStateError( 'corrupt', `State '${definitionId}' storage is corrupt (${action}${detail})`, { cause: error }, ); } - if (shape.errcode === SQLITE_BUSY) { + if (errcode === SQLITE_BUSY) { return new AgentStateError( 'unavailable', `State '${definitionId}' storage stayed locked beyond the busy timeout (${action})`, @@ -927,11 +959,24 @@ class SqliteStore implements Age Effect.gen(function*() { const db = yield* SqliteConnection; yield* sqliteEffect(definitionId, 'configure storage', () => { - // busy_timeout first: switching journal modes takes the database - // lock, and two processes racing the very first open would otherwise - // fail SQLITE_BUSY with a zero retry budget. db.exec(`PRAGMA busy_timeout = ${String(busyTimeoutMs)}`); + }); + // busy_timeout does not cover this statement. Switching a rollback- + // mode database into WAL opens a read transaction and upgrades it to + // a write (SHARED -> RESERVED), the one lock transition SQLite never + // routes through the busy handler (deadlock avoidance), so two + // processes racing the very first open of a file fail SQLITE_BUSY at + // once. Retry the switch under the same budget: once the header says + // WAL the statement is a plain read and never contends again. + yield* sqliteEffect(definitionId, 'configure storage', () => { db.exec('PRAGMA journal_mode = WAL'); + }).pipe( + Effect.retry({ + schedule: walSwitchRetries(busyTimeoutMs), + while: (error) => isSqliteBusy(error.cause), + }), + ); + yield* sqliteEffect(definitionId, 'configure storage', () => { db.exec('PRAGMA synchronous = FULL'); }); yield* initializeStorage; diff --git a/packages/rsc-runtime/tests/state-sqlite.test.ts b/packages/rsc-runtime/tests/state-sqlite.test.ts index 93075199d..d61db04f3 100644 --- a/packages/rsc-runtime/tests/state-sqlite.test.ts +++ b/packages/rsc-runtime/tests/state-sqlite.test.ts @@ -382,6 +382,54 @@ describe('sqlite driver storage behavior', () => { await store.close(); })); + // A connection holding the rollback-mode write lock (RESERVED) on a fresh + // file stands in for a second process caught mid `PRAGMA journal_mode = WAL`. + // SQLite answers the driver's own switch SQLITE_BUSY at once, without + // consulting busy_timeout, so the driver must retry the switch itself. + it('waits for another process mid-WAL-switch instead of failing the first open', () => + withRoot(async (root) => { + const file = join(root, 'state.sqlite'); + const keeper = new DatabaseSync(file); + keeper.exec('BEGIN IMMEDIATE'); + try { + const opening = createSqliteStateDriver({ file }).open(counterDefinition()); + // Neither outcome is reachable while the lock is held: the switch cannot + // succeed, and the 5 s default budget cannot run out in 50 ms. + const outcome = await Promise.race([ + opening.then(() => 'opened', () => 'failed'), + new Promise<'pending'>((resolve) => setTimeout(() => resolve('pending'), 50)), + ]); + expect(outcome).toBe('pending'); + keeper.exec('ROLLBACK'); + const store = await opening; + expect(keeper.prepare('PRAGMA journal_mode').get()).toEqual({ journal_mode: 'wal' }); + await expect(store.dispatch('bumped', { by: 1 }, { idempotencyKey: 'k1' })).resolves.toMatchObject({ revision: 1 }); + await store.close(); + } finally { + keeper.close(); + } + })); + + it('fails the first open as unavailable once the WAL switch outlives busyTimeoutMs', () => + withRoot(async (root) => { + const file = join(root, 'state.sqlite'); + const keeper = new DatabaseSync(file); + keeper.exec('BEGIN IMMEDIATE'); + try { + // Budgets both above and below the retry pause fail closed; the + // shorter one proves the pause is clamped rather than outliving it. + for (const busyTimeoutMs of [100, 1]) { + await expect(createSqliteStateDriver({ busyTimeoutMs, file }).open(counterDefinition())).rejects.toMatchObject({ + code: 'unavailable', + message: expect.stringContaining('storage stayed locked beyond the busy timeout (configure storage)') as string, + name: 'AgentStateError', + }); + } + } finally { + keeper.close(); + } + })); + it('rolls back failed transactions without collapsing unexpected defects', () => withRoot(async (root) => { const defect = new Error('clock implementation defect');