From 73f3e5d06ae9330dc4cb1efc747db2d63809026c Mon Sep 17 00:00:00 2001 From: chenhao Date: Wed, 23 Sep 2026 02:49:40 +0800 Subject: [PATCH] fix(storage): retry contended writes across the whole Turn path Every mcode process opens the same runtime-state.sqlite in WAL mode, so a concurrent process holding the single write lock made a foreign writer's SQLITE_BUSY surface as "database is locked" and aborted the Turn. PR #287 covered only the v2 message-upsert path; turn admission, session state projection, settlement, queue writes and the v1 Goal store still failed after the five-second native busy timeout. Move the retry policy into @mavis/shared/sqlite-write-retry and apply it to every turn-critical write. Each attempt takes only a 50ms native stall and the rest of the ten-second budget is asynchronous backoff, so a foreign writer can no longer block the event loop for the whole budget. Only lock acquisition is replayed: a callback that already started rolls back and propagates unchanged. Reads keep their synchronous path because WAL readers are not blocked by a foreign writer. Cron orchestration stays on the old policy - its scheduler handler is synchronous end to end and a failed persist is already reported as CRON_PERSIST_FAILED. Related to #282. --- .../src/infra/db/write-transaction.ts | 120 +++++++------- .../session-system/messages/repo/drizzle.ts | 66 ++++---- .../session-system/queue/repo/drizzle.ts | 21 ++- .../session-system/sessions/repo/drizzle.ts | 51 +++--- .../persistence/turn.repository.ts | 51 +++--- .../src/thread-goal/db-write-lock.ts | 52 ++++++ .../local-runtime/src/thread-goal/store.ts | 35 ++-- .../test/unit/local-thread-goal-store.test.ts | 77 ++++++++- packages/shared/package.json | 4 + packages/shared/src/sqlite-write-retry.ts | 113 +++++++++++++ release/public-source.json | 3 + test/sqlite-turn-contention.test.ts | 156 ++++++++++++++++++ test/vitest-suites.json | 1 + tsconfig.standalone.json | 3 + 14 files changed, 587 insertions(+), 166 deletions(-) create mode 100644 packages/local-runtime/src/thread-goal/db-write-lock.ts create mode 100644 packages/shared/src/sqlite-write-retry.ts create mode 100644 test/sqlite-turn-contention.test.ts diff --git a/packages/local-runtime-v2/src/infra/db/write-transaction.ts b/packages/local-runtime-v2/src/infra/db/write-transaction.ts index 5146ca9b..e6b5ac3d 100644 --- a/packages/local-runtime-v2/src/infra/db/write-transaction.ts +++ b/packages/local-runtime-v2/src/infra/db/write-transaction.ts @@ -1,8 +1,12 @@ -import { setTimeout as delay } from 'node:timers/promises'; import { sql } from 'drizzle-orm'; +import { + retrySqliteWrite, + SqliteWriteAttemptFailedError, + SqliteWriteRetryAbortedError, +} from '@mavis/shared/sqlite-write-retry'; + import type { AppDb } from './client.js'; -const WRITE_LOCK_BUDGET_MS = 10_000; const WRITE_LOCK_ATTEMPT_MS = 50; /** Cancellation before the mutation callback starts; no write needs to be replayed. */ @@ -15,72 +19,66 @@ export class WriteLockWaitAbortedError extends Error { } /** - * Retry only transaction admission: a callback that has started is never replayed. - * The signal cancels contention waits, not an immediately available write. This - * lets post-cancellation tool completion and cleanup messages remain durable. + * The transaction handle drizzle hands to a `transaction()` callback. Callers + * thread this through their `*InTransaction` helpers, so the retrying call has + * to expose the same type the raw drizzle call used to. + */ +type DrizzleWriteTransaction = Parameters[0]>[0]; + +/** + * Run a `BEGIN IMMEDIATE` mutation under the shared bounded-stall retry policy + * (see `@mavis/shared/sqlite-write-retry`). Only lock acquisition is retried: + * once the callback has started, any error — busy or not — rolls the + * transaction back and propagates without a replay. + * + * The optional signal cancels contention waits only. An immediately available + * write still commits, which keeps post-cancellation tool completion and + * cleanup messages durable. */ export async function runWithWriteLock( db: AppDb, - mutation: (tx: AppDb) => T, + mutation: (tx: DrizzleWriteTransaction) => T, options: { readonly signal?: AbortSignal; readonly timeoutMs?: number } = {}, ): Promise { - const budget = options.timeoutMs ?? WRITE_LOCK_BUDGET_MS; - if (!Number.isFinite(budget) || budget <= 0) - throw new RangeError('Invalid write lock budget'); - const deadline = performance.now() + budget; - let attempt = 0; - let hasContended = false; - let lastBusy: unknown = new Error('SQLite write lock wait exceeded its deadline'); - for (;;) { - if (hasContended) throwIfWaitAborted(options.signal); - const remaining = deadline - performance.now(); - if (remaining <= 0) throw lastBusy; - const previous = db.get<{ timeout: number }>(sql`PRAGMA busy_timeout`).timeout; - let entered = false; - try { - const nativeWaitMs = options.signal?.aborted - ? 0 - : Math.ceil(Math.min(WRITE_LOCK_ATTEMPT_MS, remaining)); - db.run(sql.raw(`PRAGMA busy_timeout = ${nativeWaitMs}`)); - return db.transaction( - (tx) => { - entered = true; - // Only lock acquisition gets a short timeout. Restore the connection's - // policy before callbacks (including nested transactions) can use it. + try { + return await retrySqliteWrite( + ({ remainingMs }) => { + const previous = db.get<{ timeout: number }>(sql`PRAGMA busy_timeout`).timeout; + let entered = false; + try { + // Only lock acquisition gets a short native wait; the callback below + // restores the connection policy before any statement runs. + db.run( + sql.raw( + `PRAGMA busy_timeout = ${Math.min(WRITE_LOCK_ATTEMPT_MS, Math.ceil(remainingMs))}`, + ), + ); + return db.transaction( + (tx) => { + entered = true; + db.run(sql.raw(`PRAGMA busy_timeout = ${previous}`)); + return mutation(tx); + }, + { behavior: 'immediate' }, + ); + } catch (error) { + // A callback that already started owns its own rollback; only a + // failure to acquire the write lock is safe to replay. + if (entered) throw new SqliteWriteAttemptFailedError(error); + throw error; + } finally { + // No await occurs while the shared connection has a temporary timeout. db.run(sql.raw(`PRAGMA busy_timeout = ${previous}`)); - return mutation(tx); - }, - { behavior: 'immediate' }, - ); - } catch (error) { - if (entered || !isBusy(error)) throw error; - lastBusy = error; - } finally { - // No await occurs while the shared connection has a temporary timeout. - db.run(sql.raw(`PRAGMA busy_timeout = ${previous}`)); - } - hasContended = true; - throwIfWaitAborted(options.signal); - const wait = Math.min( - deadline - performance.now(), - 25 * 2 ** Math.min(attempt++, 3) + Math.random() * 25, + } + }, + { timeoutMs: options.timeoutMs, signal: options.signal }, ); - if (wait <= 0) throw lastBusy; - try { - await delay(wait, undefined, { signal: options.signal }); - } catch (error) { - if (options.signal?.aborted && error instanceof Error && error.name === 'AbortError') { - throw new WriteLockWaitAbortedError(options.signal); - } - throw error; + } catch (error) { + // Only an abandoned wait becomes a Turn-facing cancellation; an error the + // callback raised after it started keeps its own identity. + if (error instanceof SqliteWriteRetryAbortedError) { + throw new WriteLockWaitAbortedError(options.signal as AbortSignal); } + throw error; } } - -function throwIfWaitAborted(signal: AbortSignal | undefined): void { - if (signal?.aborted) throw new WriteLockWaitAbortedError(signal); -} - -function isBusy(error: unknown): boolean { - return error instanceof Error && Reflect.get(error, 'code') === 'SQLITE_BUSY'; -} diff --git a/packages/local-runtime-v2/src/service/session-system/messages/repo/drizzle.ts b/packages/local-runtime-v2/src/service/session-system/messages/repo/drizzle.ts index eb62483a..fc03aec5 100644 --- a/packages/local-runtime-v2/src/service/session-system/messages/repo/drizzle.ts +++ b/packages/local-runtime-v2/src/service/session-system/messages/repo/drizzle.ts @@ -74,7 +74,7 @@ class DrizzleMessageRepository implements MessageRepository { } async get(sessionId: string, msgId: string): Promise { - this.ensureReady(sessionId); + await this.ensureReady(sessionId); const db = this.options.db; let query = messageReads.get(db); if (!query) { @@ -86,7 +86,7 @@ class DrizzleMessageRepository implements MessageRepository { } async list(sessionId: string, options: ListMessagesOptions = {}): Promise { - this.ensureReady(sessionId); + await this.ensureReady(sessionId); const anchorId = options.before ? this.messageRowId(sessionId, options.before) : undefined; const limit = normalizeLimit(options.limit, 0); const query = this.options.db @@ -110,7 +110,7 @@ class DrizzleMessageRepository implements MessageRepository { } async listAfter(sessionId: string, afterMsgId?: string): Promise { - this.ensureReady(sessionId); + await this.ensureReady(sessionId); const anchorId = afterMsgId ? this.messageRowId(sessionId, afterMsgId) : undefined; if (afterMsgId && anchorId === undefined) return { status: 'missing-anchor', messages: [] }; const rows = this.options.db @@ -128,7 +128,7 @@ class DrizzleMessageRepository implements MessageRepository { } async listTurn(sessionId: string, turnId: string): Promise { - this.ensureReady(sessionId); + await this.ensureReady(sessionId); const db = this.options.db; let query = turnReads.get(db); if (!query) { @@ -147,7 +147,7 @@ class DrizzleMessageRepository implements MessageRepository { }, ): Promise { if (!Number.isFinite(options.limit) || options.limit <= 0) return []; - this.ensureReady(sessionId); + await this.ensureReady(sessionId); const rows = this.options.db .select() .from(messageRows) @@ -175,7 +175,8 @@ class DrizzleMessageRepository implements MessageRepository { generatedIdDiscriminator: `user:${input.turnId}`, nowMs: this.nowMs, }); - return this.options.db.transaction( + return await runWithWriteLock( + this.options.db, (tx) => { const rejection = this.options.userMessageAdmission?.rejectionInTransaction(tx, { sessionId: input.sessionId, @@ -235,7 +236,6 @@ class DrizzleMessageRepository implements MessageRepository { message: decodeDisplayMessage(inserted), }; }, - { behavior: 'immediate' }, ); } @@ -282,7 +282,7 @@ class DrizzleMessageRepository implements MessageRepository { async replace(input: MessageReplaceInput): Promise { const normalized = this.normalizeReplacementBatch(input.messages, 0); - this.writeReplacementBatch(input.sessionId, normalized, true); + await this.writeReplacementBatch(input.sessionId, normalized, true); } async replaceStream(input: MessageReplaceStreamInput): Promise { @@ -291,15 +291,16 @@ class DrizzleMessageRepository implements MessageRepository { for await (const batch of input.batches) { if (batch.length === 0) continue; const normalized = this.normalizeReplacementBatch(batch, messageIndex); - this.writeReplacementBatch(input.sessionId, normalized, replaceExisting); + await this.writeReplacementBatch(input.sessionId, normalized, replaceExisting); messageIndex += normalized.length; replaceExisting = false; } - if (replaceExisting) this.writeReplacementBatch(input.sessionId, [], true); + if (replaceExisting) await this.writeReplacementBatch(input.sessionId, [], true); } async rewindInclusive(input: MessageRewindInclusiveInput): Promise { - return this.options.db.transaction( + return await runWithWriteLock( + this.options.db, (tx) => { this.ensureReadyInTransaction(tx, input.sessionId); if (!input.fromMessageId.startsWith('msg-user-v1-')) { @@ -366,12 +367,11 @@ class DrizzleMessageRepository implements MessageRepository { this.markSessionAssetIndexCurrent(tx, input.sessionId); return { deletedMessageIds }; }, - { behavior: 'immediate' }, ); } async rewind(input: MessageRewindInput): Promise { - this.mutationTransaction(input.sessionId, (tx) => { + await this.mutationTransaction(input.sessionId, (tx) => { const ids = new Set(input.messageIds ?? []); const anchorId = input.afterMessageId ? this.messageRowId(input.sessionId, input.afterMessageId, tx) @@ -401,7 +401,7 @@ class DrizzleMessageRepository implements MessageRepository { } async resolveTurnSource(sessionId: string, turnId: string) { - this.ensureReady(sessionId); + await this.ensureReady(sessionId); const row = this.options.db .select() .from(messageRows) @@ -419,7 +419,7 @@ class DrizzleMessageRepository implements MessageRepository { } async latestDisplayRowId(sessionId: string): Promise { - this.ensureReady(sessionId); + await this.ensureReady(sessionId); return ( this.options.db .select({ id: messageRows.id }) @@ -444,12 +444,12 @@ class DrizzleMessageRepository implements MessageRepository { targetSessionId: string; throughMessageId: string; }): Promise { - this.options.db.transaction( + await runWithWriteLock( + this.options.db, (tx) => { this.copyPrefixInTransaction(tx, input); this.appendForkOriginInTransaction(tx, input); }, - { behavior: 'immediate' }, ); } @@ -514,7 +514,8 @@ class DrizzleMessageRepository implements MessageRepository { targetSessionId: string; sourceSessionId: string; }): Promise { - this.options.db.transaction((tx) => { + await runWithWriteLock( + this.options.db,(tx) => { this.ensureReadyInTransaction(tx, input.targetSessionId); this.appendForkOriginInTransaction(tx, input); }); @@ -544,7 +545,8 @@ class DrizzleMessageRepository implements MessageRepository { } async deleteSessionData(sessionId: string): Promise { - this.options.db.transaction((tx) => { + await runWithWriteLock( + this.options.db,(tx) => { tx.delete(sessionAssets).where(eq(sessionAssets.sessionId, sessionId)).run(); tx.delete(sessionAssetIndexState) .where(eq(sessionAssetIndexState.sessionId, sessionId)) @@ -567,23 +569,23 @@ class DrizzleMessageRepository implements MessageRepository { }); } - private ensureReady(sessionId: string): void { + private async ensureReady(sessionId: string): Promise { const marker = this.options.db .select({ sessionId: messageRowMigrations.sessionId }) .from(messageRowMigrations) .where(eq(messageRowMigrations.sessionId, sessionId)) .get(); if (marker) return; - this.mutationTransaction(sessionId, () => undefined); + await this.mutationTransaction(sessionId, () => undefined); } - private mutationTransaction(sessionId: string, mutation: (db: AppDb) => T): T { - return this.options.db.transaction( - (tx) => { - this.ensureReadyInTransaction(tx, sessionId); - return mutation(tx); - }, - { behavior: 'immediate' }, - ); + private mutationTransaction( + sessionId: string, + mutation: (db: AppDb) => T, + ): Promise { + return runWithWriteLock(this.options.db, (tx) => { + this.ensureReadyInTransaction(tx, sessionId); + return mutation(tx); + }); } private ensureReadyInTransaction(db: AppDb, sessionId: string): void { ensureMessageRowsReadyInTransaction(db, sessionId, this.nowMs(), (message, index) => { @@ -653,12 +655,12 @@ class DrizzleMessageRepository implements MessageRepository { }), ); } - private writeReplacementBatch( + private async writeReplacementBatch( sessionId: string, messages: readonly NormalizedDisplayMessage[], replaceExisting: boolean, - ): void { - this.mutationTransaction(sessionId, (tx) => { + ): Promise { + await this.mutationTransaction(sessionId, (tx) => { if (replaceExisting) { tx.delete(messageRows).where(eq(messageRows.sessionId, sessionId)).run(); tx.delete(sessionAssets).where(eq(sessionAssets.sessionId, sessionId)).run(); diff --git a/packages/local-runtime-v2/src/service/session-system/queue/repo/drizzle.ts b/packages/local-runtime-v2/src/service/session-system/queue/repo/drizzle.ts index 66c7da16..9da79c9a 100644 --- a/packages/local-runtime-v2/src/service/session-system/queue/repo/drizzle.ts +++ b/packages/local-runtime-v2/src/service/session-system/queue/repo/drizzle.ts @@ -3,6 +3,7 @@ import { randomUUID } from 'node:crypto'; import { and, asc, eq, inArray, isNotNull, lte, sql } from 'drizzle-orm'; import type { AppDb } from '../../../../infra/db/client.js'; +import { runWithWriteLock } from '../../../../infra/db/write-transaction.js'; import { legacyQueues, queueItems, @@ -325,7 +326,7 @@ class DrizzleQueueRepository implements QueueRepository { readonly turnId: string; }): Promise { if (!input.turnId) throw new TypeError('Queue delivery requires a Turn identity'); - this.transaction(input.sessionId, (db) => { + await this.transaction(input.sessionId, (db) => { const items = this.claimRows(db, input.sessionId, input.claimId); if (items.length === 0) throw new QueueClaimNotFoundError(input.sessionId, input.claimId); for (const item of items) { @@ -537,7 +538,8 @@ class DrizzleQueueRepository implements QueueRepository { if (items.some((item) => item.sessionId !== sessionId || !isQueueItem(item))) { throw new TypeError('Queue replacement contains an invalid item'); } - this.options.db.transaction( + await runWithWriteLock( + this.options.db, (db) => { this.replaceRows(db, sessionId, items); db.insert(queueRowMigrations) @@ -550,7 +552,6 @@ class DrizzleQueueRepository implements QueueRepository { db.delete(legacyQueues).where(eq(legacyQueues.sessionId, sessionId)).run(); this.reconcilePause(db, sessionId); }, - { behavior: 'immediate' }, ); return committed(undefined); } @@ -570,11 +571,11 @@ class DrizzleQueueRepository implements QueueRepository { }); } - private transaction( + private async transaction( sessionId: string, operation: (db: AppDb) => CommittedQueueResult, - ): CommittedQueueResult { - return this.options.db.transaction( + ): Promise> { + return runWithWriteLock(this.options.db, (db) => { ensureQueueRowsReadyInTransaction(db, sessionId, this.nowMs()); const wasPaused = Boolean(this.pause(db, sessionId)); @@ -582,15 +583,14 @@ class DrizzleQueueRepository implements QueueRepository { this.reconcilePause(db, sessionId); return withPauseRemovalFact(result, sessionId, wasPaused && !this.pause(db, sessionId)); }, - { behavior: 'immediate' }, ); } - private mutationTransaction( + private async mutationTransaction( sessionId: string, operation: (db: AppDb) => CommittedQueueResult, - ): CommittedQueueResult { - return this.options.db.transaction( + ): Promise> { + return runWithWriteLock(this.options.db, (db) => { this.assertMutationAdmitted(db, sessionId); ensureQueueRowsReadyInTransaction(db, sessionId, this.nowMs()); @@ -599,7 +599,6 @@ class DrizzleQueueRepository implements QueueRepository { this.reconcilePause(db, sessionId); return withPauseRemovalFact(result, sessionId, wasPaused && !this.pause(db, sessionId)); }, - { behavior: 'immediate' }, ); } diff --git a/packages/local-runtime-v2/src/service/session-system/sessions/repo/drizzle.ts b/packages/local-runtime-v2/src/service/session-system/sessions/repo/drizzle.ts index 007f23af..355c32fd 100644 --- a/packages/local-runtime-v2/src/service/session-system/sessions/repo/drizzle.ts +++ b/packages/local-runtime-v2/src/service/session-system/sessions/repo/drizzle.ts @@ -20,6 +20,7 @@ import { } from 'drizzle-orm'; import type { AppDb } from '../../../../infra/db/client.js'; +import { runWithWriteLock } from '../../../../infra/db/write-transaction.js'; import { sessionAgentDefinitions, taskSessionBindings, @@ -264,7 +265,8 @@ class DrizzleSessionRepository implements SessionRepository { ? serializeTaskSessionBinding(toLegacyTaskSessionBinding(definitionInput)) : undefined; try { - this.options.db.transaction((tx) => { + await runWithWriteLock( + this.options.db,(tx) => { tx.insert(sessions) .values(encodeSessionRow(record, { projectId: input.projectId })) .run(); @@ -317,7 +319,8 @@ class DrizzleSessionRepository implements SessionRepository { backfill: SessionAgentDefinitionBackfill, ): Promise { const serialized = serializeSessionAgentDefinition(backfill.agentDefinition); - return this.options.db.transaction( + return runWithWriteLock( + this.options.db, (tx) => { const row = selectRow(tx, sessionId); if (!row) throw new Error(`Session does not exist: ${sessionId}`); @@ -336,7 +339,6 @@ class DrizzleSessionRepository implements SessionRepository { if (!definition) throw new Error(`Session Agent definition was not created: ${sessionId}`); return decodeSessionAgentDefinition(definition); }, - { behavior: 'immediate' }, ); } @@ -345,7 +347,8 @@ class DrizzleSessionRepository implements SessionRepository { next: SessionAgentDefinitionBackfill, ): Promise { const serialized = serializeSessionAgentDefinition(next.agentDefinition); - return this.options.db.transaction( + return runWithWriteLock( + this.options.db, (tx) => { const row = selectRow(tx, sessionId); if (!row) throw new Error(`Session does not exist: ${sessionId}`); @@ -366,7 +369,6 @@ class DrizzleSessionRepository implements SessionRepository { .run(); return { sessionId, definition: next.agentDefinition.definition }; }, - { behavior: 'immediate' }, ); } @@ -384,7 +386,8 @@ class DrizzleSessionRepository implements SessionRepository { backfill: SessionTaskAgentBindingBackfill, ): Promise { const serialized = serializeTaskSessionBinding(backfill.taskAgentBinding); - return this.options.db.transaction( + return runWithWriteLock( + this.options.db, (tx) => { const row = selectRow(tx, sessionId); if (!row) throw new Error(`Session does not exist: ${sessionId}`); @@ -403,7 +406,6 @@ class DrizzleSessionRepository implements SessionRepository { if (!binding) throw new Error(`Task Agent binding was not created: ${sessionId}`); return decodeTaskSessionBinding(binding); }, - { behavior: 'immediate' }, ); } @@ -413,7 +415,8 @@ class DrizzleSessionRepository implements SessionRepository { expectedModel?: SessionModelSnapshot, expectedTitle?: string | null, ): Promise { - return this.options.db.transaction( + return runWithWriteLock( + this.options.db, (tx) => { const row = selectRow(tx, sessionId); if (!row) return undefined; @@ -429,7 +432,6 @@ class DrizzleSessionRepository implements SessionRepository { syncSessionAgentDefinitionModel(tx, sessionId, modelFields); return record; }, - { behavior: 'immediate' }, ); } @@ -440,7 +442,8 @@ class DrizzleSessionRepository implements SessionRepository { const normalizedCronId = originCronId.trim(); if (!normalizedCronId) return []; const normalizedTargetSessionId = targetSessionId?.trim(); - return this.options.db.transaction( + return runWithWriteLock( + this.options.db, (tx) => { const rows = tx .select() @@ -472,20 +475,20 @@ class DrizzleSessionRepository implements SessionRepository { return detached; }); }, - { behavior: 'immediate' }, ); } async upsert(record: SessionWriteRecord): Promise { - this.writeUpsert(record, false); + await this.writeUpsert(record, false); } async upsertImportedLegacy(record: SessionWriteRecord): Promise { - this.writeUpsert(record, true); + await this.writeUpsert(record, true); } async delete(sessionId: string): Promise { - this.options.db.transaction((tx) => { + await runWithWriteLock( + this.options.db,(tx) => { deleteSessionSearchDocument(tx, sessionId); tx.delete(sessionAgentState).where(eq(sessionAgentState.sessionId, sessionId)).run(); tx.delete(sessions).where(eq(sessions.sessionId, sessionId)).run(); @@ -504,7 +507,8 @@ class DrizzleSessionRepository implements SessionRepository { sessionId: string, relativeDir: string, ): Promise { - return this.options.db.transaction( + return runWithWriteLock( + this.options.db, (tx) => { const row = tx .select({ historyRelativeDir: sessions.historyRelativeDir }) @@ -531,12 +535,12 @@ class DrizzleSessionRepository implements SessionRepository { .get()?.historyRelativeDir ?? undefined ); }, - { behavior: 'immediate' }, ); } async swapRoot(input: SessionRootSwapInput): Promise { - return this.options.db.transaction( + return runWithWriteLock( + this.options.db, (tx) => { const nextRow = selectRow(tx, input.nextRootSessionId); if (!nextRow) throw new SessionRootSwapError('next-root-not-found'); @@ -597,7 +601,6 @@ class DrizzleSessionRepository implements SessionRepository { writeRow(tx, nextRow, nextRoot); return { previousRoots, nextRoot }; }, - { behavior: 'immediate' }, ); } @@ -605,7 +608,8 @@ class DrizzleSessionRepository implements SessionRepository { parentSessionId: string, nextParentSessionId: string | null, ): Promise { - this.options.db.transaction((tx) => { + await runWithWriteLock( + this.options.db,(tx) => { const nowMs = this.nowMs(); const rows = tx .select() @@ -628,7 +632,8 @@ class DrizzleSessionRepository implements SessionRepository { async applyAgentState(input: AgentSessionStateMutation): Promise { assertAgentStateMutation(input); - return this.options.db.transaction( + return runWithWriteLock( + this.options.db, (tx) => { const row = selectRow(tx, input.sessionId); if (!row) return { status: 'not-found' }; @@ -649,7 +654,6 @@ class DrizzleSessionRepository implements SessionRepository { .run(); return { status: 'applied' }; }, - { behavior: 'immediate' }, ); } @@ -962,12 +966,13 @@ class DrizzleSessionRepository implements SessionRepository { return result; } - private writeUpsert(record: SessionWriteRecord, replaceIdentity: boolean): void { + private async writeUpsert(record: SessionWriteRecord, replaceIdentity: boolean): Promise { const normalizedRecord: SessionRecord = { ...record, sessionType: normalizeSessionType(record.sessionType), }; - this.options.db.transaction((tx) => { + await runWithWriteLock( + this.options.db,(tx) => { const existing = selectRow(tx, normalizedRecord.sessionId); if (!existing) { tx.insert(sessions).values(encodeSessionRow(normalizedRecord)).run(); diff --git a/packages/local-runtime-v2/src/service/turn-system/persistence/turn.repository.ts b/packages/local-runtime-v2/src/service/turn-system/persistence/turn.repository.ts index ab24d34d..fa909935 100644 --- a/packages/local-runtime-v2/src/service/turn-system/persistence/turn.repository.ts +++ b/packages/local-runtime-v2/src/service/turn-system/persistence/turn.repository.ts @@ -29,6 +29,7 @@ import type { TurnRepositoryOptions, } from './contracts.js'; import { createPluginHookSessionPersistence } from './plugin-hook-session.repository.js'; +import { runWithWriteLock } from '../../../infra/db/write-transaction.js'; import { persistedTurnConsumesQueuePause, publishQueueFacts, @@ -68,7 +69,8 @@ export function createTurnRepository(options: TurnRepositoryOptions): TurnReposi return { tryAcquireSessionMaintenance: async (sessionId) => - options.db.transaction( + await runWithWriteLock( + options.db, (tx) => { const now = nowMs(); const rejection = options.sessionAdmission.rejectionInTransaction(tx, { sessionId }); @@ -94,7 +96,6 @@ export function createTurnRepository(options: TurnRepositoryOptions): TurnReposi recovery.terminalFacts, ); }, - { behavior: 'immediate' }, ), renewSessionMaintenance: async (lease) => { const now = nowMs(); @@ -145,7 +146,8 @@ export function createTurnRepository(options: TurnRepositoryOptions): TurnReposi leaseId: input.leaseId, }; try { - const committed = options.db.transaction( + const committed = await runWithWriteLock( + options.db, (tx) => { const completedAtMs = nowMs(); const result = settleInTransaction(tx, input, completedAtMs); @@ -160,7 +162,6 @@ export function createTurnRepository(options: TurnRepositoryOptions): TurnReposi : []; return { result, facts }; }, - { behavior: 'immediate' }, ); if (samePendingSettlement(pendingSettlements.get(input.sessionId), pending)) { pendingSettlements.delete(input.sessionId); @@ -173,9 +174,9 @@ export function createTurnRepository(options: TurnRepositoryOptions): TurnReposi } }, recoverExpired: async (sessionId) => - options.db.transaction((tx) => recoverExpiredInTransaction(tx, sessionId, nowMs()), { - behavior: 'immediate', - }), + runWithWriteLock(options.db, (tx) => + recoverExpiredInTransaction(tx, sessionId, nowMs()), + ), recoverProcessRestart: ({ processStartedAtMs }) => recoverRepositoryProcessRestart( options, @@ -189,7 +190,8 @@ export function createTurnRepository(options: TurnRepositoryOptions): TurnReposi ), beginSessionDeletion: async (sessionId) => { const pending = pendingSettlements.get(sessionId); - const result = options.db.transaction( + const result = await runWithWriteLock( + options.db, (tx) => { const now = nowMs(); if (pending) recoverPendingSettlementInTransaction(tx, pending, now); @@ -201,7 +203,6 @@ export function createTurnRepository(options: TurnRepositoryOptions): TurnReposi isLeaseOwnerCurrent, }); }, - { behavior: 'immediate' }, ); if (pendingSettlements.get(sessionId) === pending) pendingSettlements.delete(sessionId); return result; @@ -247,7 +248,8 @@ async function admitRepositoryTurn( }, ): Promise { const pending = dependencies.pendingSettlements.get(input.sessionId); - const committed = options.db.transaction( + const committed = await runWithWriteLock( + options.db, (tx) => { const result = admitTurnInTransaction(tx, options, input, { pending, @@ -272,7 +274,6 @@ async function admitRepositoryTurn( facts: effect.facts, }; }, - { behavior: 'immediate' }, ); if (dependencies.pendingSettlements.get(input.sessionId) === pending) { dependencies.pendingSettlements.delete(input.sessionId); @@ -315,7 +316,8 @@ async function recoverRepositoryProcessRestart( throw new TypeError('processStartedAtMs must be finite'); } const observedLegacyTurnLeases = new Set(); - const committed = options.db.transaction( + const committed = await runWithWriteLock( + options.db, (tx) => { const recovery = recoverProcessRestartInTransaction(tx, { ...input, @@ -348,7 +350,6 @@ async function recoverRepositoryProcessRestart( queueFacts, }; }, - { behavior: 'immediate' }, ); for (const key of legacyTurnLeaseObservations.keys()) { if (!observedLegacyTurnLeases.has(key)) legacyTurnLeaseObservations.delete(key); @@ -394,7 +395,8 @@ async function releaseSessionMaintenanceLease( nowMs: () => number, lease: Parameters[0], ): Promise { - options.db.transaction( + await runWithWriteLock( + options.db, (tx) => { const lock = findSessionLock(tx, lease.sessionId); if (lock?.ownerId !== lease.leaseId) return; @@ -425,7 +427,6 @@ async function releaseSessionMaintenanceLease( .run(); } }, - { behavior: 'immediate' }, ); } @@ -434,7 +435,8 @@ async function deleteTurnSessionData( pendingSettlements: Map, sessionId: string, ): Promise { - options.db.transaction( + await runWithWriteLock( + options.db, (tx) => { const lock = findSessionLock(tx, sessionId); const turnIds = tx @@ -455,7 +457,6 @@ async function deleteTurnSessionData( } tx.delete(turnIngress).where(eq(turnIngress.sessionId, sessionId)).run(); }, - { behavior: 'immediate' }, ); pendingSettlements.delete(sessionId); } @@ -465,7 +466,8 @@ async function completeTurnSessionDeletion( sessionId: string, isLeaseOwnerCurrent: (ownerId: string) => boolean, ): Promise { - options.db.transaction( + await runWithWriteLock( + options.db, (tx) => { const lock = findSessionLock(tx, sessionId); if (!lock) return; @@ -487,7 +489,6 @@ async function completeTurnSessionDeletion( } tx.delete(sessionLocks).where(eq(sessionLocks.sessionId, sessionId)).run(); }, - { behavior: 'immediate' }, ); } @@ -581,7 +582,8 @@ async function reserveSteeringReceipt( options: TurnRepositoryOptions, input: Parameters[0], ): ReturnType { - return options.db.transaction( + return runWithWriteLock( + options.db, (tx) => { const existing = tx .select({ turnId: turnIngressClientRequests.turnId }) @@ -616,7 +618,6 @@ async function reserveSteeringReceipt( .run(); return { status: 'reserved' as const }; }, - { behavior: 'immediate' }, ); } @@ -624,7 +625,8 @@ async function releaseSteeringReceipt( options: TurnRepositoryOptions, input: Parameters[0], ): Promise { - options.db.transaction( + await runWithWriteLock( + options.db, (tx) => { tx.delete(turnIngressClientRequests) .where( @@ -636,7 +638,6 @@ async function releaseSteeringReceipt( ) .run(); }, - { behavior: 'immediate' }, ); } @@ -645,7 +646,8 @@ async function revokeAdmission( input: Parameters[0], restoredAtMs: number, ): Promise { - const committed = options.db.transaction( + const committed = await runWithWriteLock( + options.db, (tx) => { const receipt = tx .select({ status: turnIngress.status }) @@ -685,7 +687,6 @@ async function revokeAdmission( }), }; }, - { behavior: 'immediate' }, ); publishQueueFacts(options, committed.facts); return committed.revoked; diff --git a/packages/local-runtime/src/thread-goal/db-write-lock.ts b/packages/local-runtime/src/thread-goal/db-write-lock.ts new file mode 100644 index 00000000..992f1a5d --- /dev/null +++ b/packages/local-runtime/src/thread-goal/db-write-lock.ts @@ -0,0 +1,52 @@ +import { + retrySqliteWrite, + SqliteWriteAttemptFailedError, +} from '@mavis/shared/sqlite-write-retry'; + +import { runInImmediateTransaction, withLocalRuntimeDb, type DatabaseLike } from '../persistence/db.js'; + +/** Native wait granted to one `BEGIN IMMEDIATE`; the rest of the budget is async. */ +const GOAL_WRITE_STALL_MS = 50; + +function readBusyTimeout(db: DatabaseLike): number { + const row = db.prepare('PRAGMA busy_timeout').get() as { timeout?: number } | undefined; + return typeof row?.timeout === 'number' ? row.timeout : 0; +} + +/** + * Run one Goal store write against the shared `runtime-state.sqlite`. + * + * v1 and v2 open that file through separate connections, so a second `mcode` + * process holding the WAL write lock used to surface `SQLITE_BUSY` straight + * into the Turn (issue #282). Lock acquisition is retried with a short native + * stall plus asynchronous backoff; once the write callback has started the + * transaction rolls back and the error propagates without a replay. + * + * Reads deliberately keep the synchronous `withLocalRuntimeDb` path: WAL + * readers are not blocked by a foreign writer, so they cannot hit this. + */ +export async function runGoalStoreWrite( + dataDir: Parameters[0], + write: (db: DatabaseLike) => T, +): Promise { + return retrySqliteWrite(() => + withLocalRuntimeDb(dataDir, (db) => { + const previous = readBusyTimeout(db); + let entered = false; + try { + // Only lock acquisition gets a short native wait. The restore below + // runs before this connection can be reused by anything else. + db.exec(`PRAGMA busy_timeout = ${GOAL_WRITE_STALL_MS}`); + return runInImmediateTransaction(db, () => { + entered = true; + return write(db); + }); + } catch (error) { + if (entered) throw new SqliteWriteAttemptFailedError(error); + throw error; + } finally { + db.exec(`PRAGMA busy_timeout = ${previous}`); + } + }), + ); +} diff --git a/packages/local-runtime/src/thread-goal/store.ts b/packages/local-runtime/src/thread-goal/store.ts index 56ada8a1..cbd580d5 100644 --- a/packages/local-runtime/src/thread-goal/store.ts +++ b/packages/local-runtime/src/thread-goal/store.ts @@ -36,6 +36,7 @@ import { runInImmediateTransaction, withLocalRuntimeDb, } from '../persistence/db.js'; +import { runGoalStoreWrite } from './db-write-lock.js'; import { updateThreadGoalBreaker } from './store-breaker.js'; import { bumpThreadGoalBoundUsage, settleThreadGoalBoundTurn } from './store-bound-settlement.js'; import { @@ -86,7 +87,7 @@ export class SqliteThreadGoalStore implements ThreadGoalStore { sessionId: string, statusReason: ThreadGoalStatusReason = 'paused(user_requested)', ): Promise { - return this.withDb((db) => { + return this.withWriteDb((db) => { const now = this.nowMs(); const result = db .prepare( @@ -114,7 +115,7 @@ export class SqliteThreadGoalStore implements ThreadGoalStore { readonly statusReason: ThreadGoalStatusReason; }, ): Promise { - return this.withDb((db) => + return this.withWriteDb((db) => runInImmediateTransaction(db, () => { const nextEpoch = Math.max(this.nowMs(), expectedEpoch + 1); const result = db @@ -140,7 +141,7 @@ export class SqliteThreadGoalStore implements ThreadGoalStore { goalId: string, expectedEpoch: number, ): Promise { - return this.withDb((db) => + return this.withWriteDb((db) => runInImmediateTransaction(db, () => { const nextEpoch = Math.max(this.nowMs(), expectedEpoch + 1); const result = db @@ -161,7 +162,7 @@ export class SqliteThreadGoalStore implements ThreadGoalStore { } async create(input: ThreadGoalCreateInput): Promise { - return this.withDb((db) => + return this.withWriteDb((db) => runInImmediateTransaction(db, () => { // codex parity (`insert_thread_goal`'s `ON CONFLICT … WHERE // status = 'complete'`): silently replace ONLY a complete goal. @@ -232,7 +233,7 @@ export class SqliteThreadGoalStore implements ThreadGoalStore { } async patch(goalId: string, input: ThreadGoalPatchInput): Promise { - return this.withDb((db) => patchThreadGoal(db, this.nowMs, goalId, input)); + return this.withWriteDb((db) => patchThreadGoal(db, this.nowMs, goalId, input)); } /** User-authored Goal mutations; kept distinct from host settlement CAS. */ @@ -245,28 +246,28 @@ export class SqliteThreadGoalStore implements ThreadGoalStore { delta: ThreadGoalBoundUsageDelta, limits: ThreadGoalBudgetLimits, ): Promise { - return this.withDb((db) => bumpThreadGoalBoundUsage(db, this.nowMs, binding, delta, limits)); + return this.withWriteDb((db) => bumpThreadGoalBoundUsage(db, this.nowMs, binding, delta, limits)); } async settleBoundTurn(input: ThreadGoalSettleBoundTurnInput): Promise { - return this.withDb((db) => settleThreadGoalBoundTurn(db, this.nowMs, input)); + return this.withWriteDb((db) => settleThreadGoalBoundTurn(db, this.nowMs, input)); } async updateBreaker( goalId: string, input: ThreadGoalBreakerInput, ): Promise { - return this.withDb((db) => updateThreadGoalBreaker(db, this.nowMs, goalId, input)); + return this.withWriteDb((db) => updateThreadGoalBreaker(db, this.nowMs, goalId, input)); } async recordVerification( input: ThreadGoalRecordVerificationInput, ): Promise { - return this.withDb((db) => recordThreadGoalVerification(db, this.nowMs, input)); + return this.withWriteDb((db) => recordThreadGoalVerification(db, this.nowMs, input)); } async delete(goalId: string): Promise { - this.withDb((db) => { + await this.withWriteDb((db) => { db.prepare(`DELETE FROM local_runtime_thread_goals WHERE goal_id = ?`).run(goalId); }); } @@ -327,7 +328,7 @@ export class SqliteThreadGoalStore implements ThreadGoalStore { expected: ThreadGoalKickoffState, next: ThreadGoalKickoffState, ): Promise { - return this.withDb((db) => { + return this.withWriteDb((db) => { const result = db .prepare( `UPDATE local_runtime_thread_goals @@ -348,14 +349,14 @@ export class SqliteThreadGoalStore implements ThreadGoalStore { readonly expectedUpdatedAt: number; readonly reason: ThreadGoalWaitReason; }): Promise { - return this.withDb((db) => setThreadGoalExecutionWait(db, { ...input, nowMs: this.nowMs() })); + return this.withWriteDb((db) => setThreadGoalExecutionWait(db, { ...input, nowMs: this.nowMs() })); } async clearExecutionWaitAtEpoch(input: { readonly goalId: string; readonly expectedUpdatedAt: number; }): Promise { - return this.withDb((db) => clearThreadGoalExecutionWait(db, input)); + return this.withWriteDb((db) => clearThreadGoalExecutionWait(db, input)); } /** Goals still showing `verification` — only ever stale rows at startup. */ @@ -366,4 +367,12 @@ export class SqliteThreadGoalStore implements ThreadGoalStore { private withDb(fn: (db: DatabaseLike) => T): T { return withLocalRuntimeDb(this.dataDir, fn); } + + /** + * Writes share `runtime-state.sqlite` with every other `mcode` process, so + * they retry lock acquisition instead of failing the Turn on SQLITE_BUSY. + */ + private withWriteDb(fn: (db: DatabaseLike) => T): Promise { + return runGoalStoreWrite(this.dataDir, fn); + } } diff --git a/packages/local-runtime/test/unit/local-thread-goal-store.test.ts b/packages/local-runtime/test/unit/local-thread-goal-store.test.ts index ff781dc8..7399b54a 100644 --- a/packages/local-runtime/test/unit/local-thread-goal-store.test.ts +++ b/packages/local-runtime/test/unit/local-thread-goal-store.test.ts @@ -32,7 +32,11 @@ import { type LastVerificationV1, } from "@mavis/goal"; -import { closeLocalRuntimeDb, openLocalRuntimeDb } from "../../src/persistence/db.js"; +import { + closeLocalRuntimeDb, + openLocalRuntimeDb, + resolveLocalRuntimeDbPath, +} from "../../src/persistence/db.js"; import { SqliteThreadGoalStore } from "../../src/thread-goal/store.js"; async function withDataDir(fn: (dir: string) => Promise): Promise { @@ -2241,3 +2245,74 @@ describe("SqliteThreadGoalStore — bound settlement", () => { }); }); }); + +/** + * Issue #282: the v1 Goal store shares `runtime-state.sqlite` with every other + * `mcode` process. A foreign writer holding the WAL write lock used to surface + * `SQLITE_BUSY` out of `transaction(...).immediate()` and fail the Turn. + */ +describe("SqliteThreadGoalStore write contention", () => { + it("commits a goal after a foreign writer outlasts the native busy timeout", async () => { + await withDataDir(async (dataDir) => { + openLocalRuntimeDb(dataDir); + const holder = await holdForeignWriter(dataDir, 8_000); + try { + const store = new SqliteThreadGoalStore(dataDir, () => 1_700_000_000_000); + const goal = await store.create({ sessionId: "s-contended", objective: "survive" }); + expect(goal.status).toBe("active"); + expect((await store.getBySession("s-contended"))?.goalId).toBe(goal.goalId); + } finally { + await holder(); + } + }); + }, 20_000); + + it("leaves reads on the synchronous path while a writer holds the lock", async () => { + await withDataDir(async (dataDir) => { + openLocalRuntimeDb(dataDir); + const holder = await holdForeignWriter(dataDir, 2_000); + try { + const store = new SqliteThreadGoalStore(dataDir, () => 1_700_000_000_000); + expect(await store.getBySession("missing")).toBeUndefined(); + } finally { + await holder(); + } + }); + }, 20_000); +}); + +/** Hold the single WAL write lock from another process, like a second `mcode`. */ +async function holdForeignWriter( + dataDir: string, + durationMs: number, +): Promise<() => Promise> { + const { fork } = await import("node:child_process"); + const { once } = await import("node:events"); + const { createRequire } = await import("node:module"); + const { writeFile } = await import("node:fs/promises"); + const script = join(dataDir, "goal-lock-holder.cjs"); + await writeFile( + script, + `const Database = require(process.argv[2]); + const db = new Database(process.argv[3]); + db.exec('BEGIN IMMEDIATE'); + process.send('locked'); + setTimeout(() => { db.exec('COMMIT'); db.close(); process.disconnect(); }, Number(process.argv[4]));`, + ); + const child = fork( + script, + [createRequire(import.meta.url).resolve("better-sqlite3"), resolveLocalRuntimeDbPath(dataDir), String(durationMs)], + { execArgv: [], stdio: ["ignore", "ignore", "ignore", "ipc"] }, + ); + const exited = once(child, "exit"); + await Promise.race([ + once(child, "message"), + exited.then(() => { + throw new Error("Lock holder exited before acquiring the lock"); + }), + ]); + return async () => { + if (child.exitCode === null && child.signalCode === null) child.kill(); + await exited; + }; +} diff --git a/packages/shared/package.json b/packages/shared/package.json index 57e7636a..881bd2d6 100644 --- a/packages/shared/package.json +++ b/packages/shared/package.json @@ -191,6 +191,10 @@ "./safety-check-v2": { "types": "./src/safety-check-v2.ts", "import": "./dist/safety-check-v2.js" + }, + "./sqlite-write-retry": { + "types": "./src/sqlite-write-retry.ts", + "import": "./dist/sqlite-write-retry.js" } }, "types": "./src/index.ts", diff --git a/packages/shared/src/sqlite-write-retry.ts b/packages/shared/src/sqlite-write-retry.ts new file mode 100644 index 00000000..36e61897 --- /dev/null +++ b/packages/shared/src/sqlite-write-retry.ts @@ -0,0 +1,113 @@ +import { setTimeout as delay } from 'node:timers/promises'; + +/** + * Shared retry policy for SQLite writers that share one WAL database across + * processes (issue #282). + * + * Every `mcode` process opens the same `runtime-state.sqlite`, and WAL allows + * a single writer at a time. A foreign writer that holds the write lock longer + * than the connection `busy_timeout` otherwise surfaces `SQLITE_BUSY` to the + * caller, which for turn-critical writes aborts a live Turn. + * + * better-sqlite3 blocks the Node event loop for the whole native wait, so each + * attempt is expected to grant only a short native stall and spend the rest of + * its budget in asynchronous backoff. That keeps the TUI responsive while + * still outliving foreign transactions. + */ + +const WRITE_RETRY_BUDGET_MS = 10_000; +const FIRST_BACKOFF_MS = 25; +const MAX_BACKOFF_EXPONENT = 3; + +/** True for `SQLITE_BUSY` and the `SQLITE_BUSY_SNAPSHOT` family. */ +export function isSqliteBusyError(error: unknown): boolean { + return error instanceof Error && Reflect.get(error, 'code') === 'SQLITE_BUSY'; +} + +/** + * The attempt failed after its write callback started. The transaction rolls + * itself back, so the retry loop must surface the cause instead of replaying + * the callback. Adapters throw this to mark "not safe to retry". + */ +export class SqliteWriteAttemptFailedError extends Error { + constructor(readonly cause: unknown) { + super('SQLite write attempt failed', { cause }); + } +} + +/** + * The wait was cancelled while the loop was retrying lock acquisition. This is + * deliberately distinct from a callback error that merely happens to be the + * signal's abort reason: only a real abandoned wait may be reclassified by the + * caller. + */ +export class SqliteWriteRetryAbortedError extends Error { + constructor(readonly signal: AbortSignal) { + super('SQLite write lock wait was cancelled', { cause: signal.reason }); + } +} + +export interface SqliteWriteRetryOptions { + /** Wall-clock budget for the whole retry loop, not per attempt. */ + readonly timeoutMs?: number; + /** + * Checked before attempts that follow a contended one and around every wait, + * so an abort never sleeps. The first attempt always runs: an uncontended + * write must stay durable even when the caller already cancelled. + */ + readonly signal?: AbortSignal; +} + +export interface SqliteWriteAttemptContext { + /** Zero-based count of attempts already made. */ + readonly attemptIndex: number; + /** Milliseconds left in the budget; cap the native stall to this. */ + readonly remainingMs: number; +} + +/** + * Re-run `attempt` while it fails with `SQLITE_BUSY` and the budget lasts. + * + * Only lock acquisition may be replayed. An adapter signals that its callback + * already started by throwing {@link SqliteWriteAttemptFailedError}, which is + * unwrapped and propagated without a retry. + */ +export async function retrySqliteWrite( + attempt: (context: SqliteWriteAttemptContext) => T, + options: SqliteWriteRetryOptions = {}, +): Promise { + const budget = options.timeoutMs ?? WRITE_RETRY_BUDGET_MS; + if (!Number.isFinite(budget) || budget <= 0) throw new RangeError('Invalid write lock budget'); + const deadline = performance.now() + budget; + let attemptIndex = 0; + let contended = false; + let lastBusy: unknown = new Error('SQLite write lock wait exceeded its deadline'); + for (;;) { + if (contended && options.signal?.aborted) throw new SqliteWriteRetryAbortedError(options.signal); + const remainingMs = deadline - performance.now(); + if (remainingMs <= 0) throw lastBusy; + try { + return attempt({ attemptIndex, remainingMs }); + } catch (error) { + if (error instanceof SqliteWriteAttemptFailedError) throw error.cause; + if (!isSqliteBusyError(error)) throw error; + lastBusy = error; + } + contended = true; + attemptIndex += 1; + if (options.signal?.aborted) throw new SqliteWriteRetryAbortedError(options.signal); + const wait = Math.min( + deadline - performance.now(), + FIRST_BACKOFF_MS * 2 ** Math.min(attemptIndex - 1, MAX_BACKOFF_EXPONENT) + Math.random() * 25, + ); + if (wait <= 0) throw lastBusy; + try { + await delay(wait, undefined, options.signal ? { signal: options.signal } : undefined); + } catch (error) { + if (options.signal?.aborted && error instanceof Error && error.name === 'AbortError') { + throw new SqliteWriteRetryAbortedError(options.signal); + } + throw error; + } + } +} diff --git a/release/public-source.json b/release/public-source.json index b5fd4d50..cd9b79b7 100644 --- a/release/public-source.json +++ b/release/public-source.json @@ -2566,6 +2566,7 @@ "packages/local-runtime/src/thread-goal/continuation-rearm.ts", "packages/local-runtime/src/thread-goal/continuation.ts", "packages/local-runtime/src/thread-goal/contract.ts", + "packages/local-runtime/src/thread-goal/db-write-lock.ts", "packages/local-runtime/src/thread-goal/dependency-gates.ts", "packages/local-runtime/src/thread-goal/eval-observability.ts", "packages/local-runtime/src/thread-goal/events.ts", @@ -2782,6 +2783,7 @@ "packages/shared/src/source-citation-id.ts", "packages/shared/src/source-provenance.ts", "packages/shared/src/sqlite-driver.ts", + "packages/shared/src/sqlite-write-retry.ts", "packages/shared/src/subagent-roles.ts", "packages/shared/src/turn-identity.ts", "packages/shared/src/watch-interval.ts", @@ -3348,6 +3350,7 @@ "test/smoke.test.mjs", "test/source-sync.test.mjs", "test/sqlite-message-contention.test.ts", + "test/sqlite-turn-contention.test.ts", "test/vitest-suites.json", "test/windows-contract.test.mjs", "third_party/pi-mono/.minimax-vendor.json", diff --git a/test/sqlite-turn-contention.test.ts b/test/sqlite-turn-contention.test.ts new file mode 100644 index 00000000..e429561c --- /dev/null +++ b/test/sqlite-turn-contention.test.ts @@ -0,0 +1,156 @@ +/** + * Issue #282: concurrent `mcode` processes share one `runtime-state.sqlite` + * (WAL). A foreign writer that outlasts the five-second native `busy_timeout` + * used to make the turn-critical write paths throw `SQLITE_BUSY`, which + * aborted the whole turn at admission, state projection or settlement. + * + * These tests drive the real DatabaseClient (real better-sqlite3, real schema) + * and hold the write lock from a separate process, exactly like a second + * `mcode` session would. Message upsert is covered by + * `test/sqlite-message-contention.test.ts`; the paths here are the ones that + * still failed after that fix. + */ +import { fork } from 'node:child_process'; +import { once } from 'node:events'; +import { mkdtemp, rm, writeFile } from 'node:fs/promises'; +import { createRequire } from 'node:module'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterEach, expect, it } from 'vitest'; +import { DatabaseClient } from '../packages/local-runtime-v2/src/infra/db/client.js'; +import { initializeDatabase } from '../packages/local-runtime-v2/src/infra/db/initialize.js'; +import { createMessageRepository } from '../packages/local-runtime-v2/src/service/session-system/messages/repo/drizzle.js'; +import { createQueueTurnAdmissionPriorityFence } from '../packages/local-runtime-v2/src/service/session-system/queue/turn-priority-fence.js'; +import { createSessionRepository } from '../packages/local-runtime-v2/src/service/session-system/sessions/repo/drizzle.js'; +import { createTurnRepository } from '../packages/local-runtime-v2/src/service/turn-system/persistence/turn.repository.js'; + +const cleanup: Array<() => Promise | void> = []; +afterEach(async () => { + for (const dispose of cleanup.splice(0).reverse()) await dispose(); +}); + +async function fixture() { + const dataDir = await mkdtemp(join(tmpdir(), 'mcode-282-repro-')); + cleanup.push(() => rm(dataDir, { recursive: true, force: true })); + const client = new DatabaseClient({ dataDir }); + cleanup.push(() => client.close()); + await initializeDatabase({ database: client, dataDir }); + const sessions = createSessionRepository({ db: client.db, nowMs: () => 1_000 }); + const messages = createMessageRepository({ db: client.db, sourceProjectionEnabled: false }); + const turns = createTurnRepository({ + db: client.db, + priorityFence: createQueueTurnAdmissionPriorityFence(), + sessionAdmission: { rejectionInTransaction: () => undefined }, + nowMs: () => 1_000, + makeLeaseId: () => 'lease-282', + }); + await sessions.create({ + sessionId: 's-282', + agentName: 'mavis', + workspaceDir: join(dataDir, 'ws'), + runtime: 'pi-agent', + }); + return { client, dataDir, sessions, messages, turns }; +} + +async function holdWriter(dataDir: string, durationMs: number) { + const file = join(dataDir, 'lock-holder.cjs'); + await writeFile( + file, + ` + const Database = require(process.argv[2]); + const db = new Database(process.argv[3]); + db.exec('BEGIN IMMEDIATE'); + process.send('locked'); + setTimeout(() => { + db.exec('COMMIT'); + db.close(); + process.disconnect(); + }, Number(process.argv[4])); + `, + ); + const child = fork( + file, + [ + createRequire(import.meta.url).resolve('better-sqlite3'), + join(dataDir, 'v2', 'sqlite', 'runtime-state.sqlite'), + String(durationMs), + ], + { execArgv: [], stdio: ['ignore', 'ignore', 'pipe', 'ipc'] }, + ); + const exited = once(child, 'exit'); + cleanup.push(async () => { + if (child.exitCode === null && child.signalCode === null) child.kill(); + await exited; + }); + await Promise.race([ + once(child, 'message'), + exited.then(() => { + throw new Error('Lock holder exited before acquiring the lock'); + }), + ]); + return { exited }; +} + +function admitInput() { + return { + sessionId: 's-282', + turnId: 'turn-282', + busyReason: 'turn' as const, + inputDigest: 'digest:282', + inputMetadata: { attachmentCount: 0, hasContent: true }, + candidateCreatedAtMs: 1_000, + priority: { kind: 'retry-continuation' as const }, + }; +} + +const FOREIGN_LOCK_MS = 8_000; + +it('commits a message after a foreign writer outlasts the native busy timeout', async () => { + const { dataDir, messages } = await fixture(); + await holdWriter(dataDir, FOREIGN_LOCK_MS); + await messages.upsert({ + sessionId: 's-282', + message: { msg_id: 'assistant-1', role: 'assistant' }, + }); + expect((await messages.list('s-282')).messages.map((message) => message.msg_id)).toEqual([ + 'assistant-1', + ]); +}, 20_000); + +it('admits a turn after a foreign writer outlasts the native busy timeout', async () => { + const { dataDir, turns } = await fixture(); + await holdWriter(dataDir, FOREIGN_LOCK_MS); + await expect(turns.admit(admitInput())).resolves.toMatchObject({ status: 'accepted' }); + expect((await turns.findActiveTurn('s-282'))?.turnId).toBe('turn-282'); +}, 20_000); + +it('applies agent state after a foreign writer outlasts the native busy timeout', async () => { + const { dataDir, sessions } = await fixture(); + await holdWriter(dataDir, FOREIGN_LOCK_MS); + await expect( + sessions.applyAgentState({ + sessionId: 's-282', + turnId: 'turn-282', + turnSequence: 1, + eventId: 'evt-1', + update: { status: 'started' }, + }), + ).resolves.toEqual({ status: 'applied' }); + expect((await sessions.get('s-282'))?.status).toBe('started'); +}, 20_000); + +it('settles a turn after a foreign writer outlasts the native busy timeout', async () => { + const { dataDir, turns } = await fixture(); + await turns.admit(admitInput()); + await holdWriter(dataDir, FOREIGN_LOCK_MS); + await expect( + turns.settle({ + sessionId: 's-282', + turnId: 'turn-282', + leaseId: 'lease-282', + outcome: 'completed', + }), + ).resolves.toMatchObject({ status: 'settled', outcome: 'completed' }); + expect(await turns.findActiveTurn('s-282')).toBeUndefined(); +}, 20_000); diff --git a/test/vitest-suites.json b/test/vitest-suites.json index c8a8090f..f1047489 100644 --- a/test/vitest-suites.json +++ b/test/vitest-suites.json @@ -159,6 +159,7 @@ "packages/local-runtime/test/unit/session-diff-fallback.test.ts", "test/history-processing.test.ts", "test/sqlite-message-contention.test.ts", + "test/sqlite-turn-contention.test.ts", "packages/tui/test/unit/tui-feature-viewport-contract.test.ts", "packages/tui/test/unit/tui-session-history-flow.test.ts", "packages/tui/test/unit/tui-session-mutation-panels.test.ts", diff --git a/tsconfig.standalone.json b/tsconfig.standalone.json index 1ad29d2b..a3758343 100644 --- a/tsconfig.standalone.json +++ b/tsconfig.standalone.json @@ -358,6 +358,9 @@ "@mavis/shared/source-provenance": [ "./packages/shared/src/source-provenance.ts" ], + "@mavis/shared/sqlite-write-retry": [ + "./packages/shared/src/sqlite-write-retry.ts" + ], "@mavis/shared/subagent-roles": [ "./packages/shared/src/subagent-roles.ts" ],