fix(runtime): retry the sqlite WAL switch under the busy timeout; await MCP session admission in tests

.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)

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<void> => {
+  const afterSequence = session.trace().entries.at(-1)?.sequence ?? 0;
+  let subscription: McpSessionTraceSubscription | undefined;
+  const sent = new Promise<void>((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<void> => new Promise<void>((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<void>((resolvePromise) => {
+          const released = new Promise<void>((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: [] });

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<number> =>
+  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<TState, TEvents extends AgentStateEventSchemas> 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;

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');