mirror of
https://github.com/stablyai/orca.git
synced 2026-10-23 16:03:49 +00:00
fix(ai-vault): use bounded reads for live opencode db
This commit is contained in:
@@ -97,7 +97,6 @@
|
||||
// project's serve spec; the module has no imports, so listing it pulls in nothing else.
|
||||
"../src/main/startup/serve-mode-argv.ts",
|
||||
"../src/main/runtime/runtime-metadata.ts",
|
||||
"../src/main/sqlite/live-sqlite-readonly.ts",
|
||||
"../src/main/sqlite/sync-database.ts",
|
||||
"../src/main/win32-utils.ts"
|
||||
],
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { LOCAL_EXECUTION_HOST_ID } from '../../shared/execution-host'
|
||||
import { LiveSqliteUnavailableError } from '../sqlite/live-sqlite-readonly'
|
||||
import { liveSqliteUnavailableIssue, sessionParseIssueFromError } from './session-scan-issues'
|
||||
|
||||
describe('sessionParseIssueFromError', () => {
|
||||
@@ -36,15 +35,4 @@ describe('sessionParseIssueFromError', () => {
|
||||
message: 'no such table: session'
|
||||
})
|
||||
})
|
||||
|
||||
it('maps a serialized worker unavailable error onto the same notice', () => {
|
||||
expect(
|
||||
sessionParseIssueFromError({
|
||||
executionHostId: LOCAL_EXECUTION_HOST_ID,
|
||||
agent: 'opencode',
|
||||
path: '/tmp/opencode.db',
|
||||
error: new Error(new LiveSqliteUnavailableError().message)
|
||||
}).kind
|
||||
).toBe('notice')
|
||||
})
|
||||
})
|
||||
|
||||
@@ -4,7 +4,7 @@ import {
|
||||
type AiVaultScanIssue
|
||||
} from '../../shared/ai-vault-types'
|
||||
import type { ExecutionHostId } from '../../shared/execution-host'
|
||||
import { isLiveSqliteUnavailableError } from '../sqlite/live-sqlite-readonly'
|
||||
import { isSqliteBusyError } from '../sqlite/readonly-sync-database'
|
||||
import { errorMessage } from './session-scanner-values'
|
||||
|
||||
// Why: a stalled WSL distro or an unreachable remote host fails one probe per
|
||||
@@ -53,7 +53,7 @@ export function sessionParseIssueFromError(args: {
|
||||
path: string
|
||||
error: unknown
|
||||
}): AiVaultScanIssue {
|
||||
if (isLiveSqliteUnavailableError(args.error)) {
|
||||
if (isSqliteBusyError(args.error)) {
|
||||
return liveSqliteUnavailableIssue({
|
||||
executionHostId: args.executionHostId,
|
||||
agent: args.agent,
|
||||
|
||||
@@ -5,10 +5,7 @@ import {
|
||||
} from './session-scanner-opencode-sqlite-paths'
|
||||
import type { SessionFileCandidate } from './session-scanner-types'
|
||||
import { errorMessage } from './session-scanner-values'
|
||||
import {
|
||||
isLiveSqliteUnavailableError,
|
||||
openLiveSqliteReadonly
|
||||
} from '../sqlite/live-sqlite-readonly'
|
||||
import { isSqliteBusyError, openReadonlySyncDatabase } from '../sqlite/readonly-sync-database'
|
||||
import type SyncDatabase from '../sqlite/sync-database'
|
||||
import { columnExists, tableExists } from '../opencode-usage/schema-helpers'
|
||||
import { liveSqliteUnavailableIssue, recordSessionScanIssue } from './session-scan-issues'
|
||||
@@ -100,7 +97,7 @@ export async function listOpenCodeSqliteSessions(args: {
|
||||
for (const dbPath of args.dbPaths) {
|
||||
let db: SyncDatabase | null = null
|
||||
try {
|
||||
db = openLiveSqliteReadonly(dbPath)
|
||||
db = openReadonlySyncDatabase(dbPath)
|
||||
if (!canReadOpenCodeSessions(db)) {
|
||||
continue
|
||||
}
|
||||
@@ -113,7 +110,7 @@ export async function listOpenCodeSqliteSessions(args: {
|
||||
} catch (err) {
|
||||
recordSessionScanIssue(
|
||||
args.issues,
|
||||
isLiveSqliteUnavailableError(err)
|
||||
isSqliteBusyError(err)
|
||||
? liveSqliteUnavailableIssue({ agent: 'opencode', path: dbPath })
|
||||
: { agent: 'opencode', path: dbPath, message: errorMessage(err) }
|
||||
)
|
||||
|
||||
@@ -1,145 +1,96 @@
|
||||
import { mkdtempSync, rmSync } from 'node:fs'
|
||||
import { mkdtemp, rm, writeFile } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { Worker } from 'node:worker_threads'
|
||||
import { afterEach, describe, expect, it } from 'vitest'
|
||||
import type { AiVaultScanIssue } from '../../shared/ai-vault-types'
|
||||
import {
|
||||
flushLiveSqliteSnapshotsForTests,
|
||||
setLiveSqliteFileCopyForTests
|
||||
} from '../sqlite/live-sqlite-readonly'
|
||||
import Database from '../sqlite/sync-database'
|
||||
import { buildOpenCodeSqliteCandidatePath } from './session-scanner-opencode-sqlite-paths'
|
||||
import { listOpenCodeSqliteSessions } from './session-scanner-opencode-sqlite-list'
|
||||
import { buildOpenCodeSqliteCandidatePath } from './session-scanner-opencode-sqlite-paths'
|
||||
import { parseOpenCodeSqliteSession } from './session-scanner-opencode-sqlite'
|
||||
|
||||
let tempDirs: string[] = []
|
||||
const temporaryDirectories: string[] = []
|
||||
|
||||
afterEach(() => {
|
||||
setLiveSqliteFileCopyForTests(null)
|
||||
flushLiveSqliteSnapshotsForTests()
|
||||
for (const dir of tempDirs) {
|
||||
rmSync(dir, { recursive: true, force: true })
|
||||
}
|
||||
tempDirs = []
|
||||
afterEach(async () => {
|
||||
await Promise.all(
|
||||
temporaryDirectories
|
||||
.splice(0)
|
||||
.map((directory) => rm(directory, { recursive: true, force: true }))
|
||||
)
|
||||
})
|
||||
|
||||
function createLockedOpenCodeDb(): { writer: Database.Database; path: string } {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'orca-opencode-sqlite-lock-'))
|
||||
tempDirs.push(dir)
|
||||
const path = join(dir, 'opencode.db')
|
||||
const HOLDER_SOURCE = `
|
||||
import { parentPort, workerData } from 'node:worker_threads'
|
||||
const { DatabaseSync } = process.getBuiltinModule('node:sqlite')
|
||||
const db = new DatabaseSync(workerData.dbPath)
|
||||
db.exec('BEGIN EXCLUSIVE')
|
||||
parentPort.postMessage('locked')
|
||||
parentPort.once('message', (message) => {
|
||||
if (message !== 'reader-started') throw new Error('Unexpected lock-holder message')
|
||||
setTimeout(() => {
|
||||
db.exec('COMMIT')
|
||||
db.close()
|
||||
parentPort.postMessage('committed')
|
||||
}, 200)
|
||||
})
|
||||
`
|
||||
|
||||
async function createDatabase(schema: string): Promise<{ directory: string; path: string }> {
|
||||
const directory = await mkdtemp(join(tmpdir(), 'orca-opencode-sqlite-lock-'))
|
||||
temporaryDirectories.push(directory)
|
||||
const path = join(directory, 'opencode.db')
|
||||
const writer = new Database(path)
|
||||
writer.exec(`
|
||||
CREATE TABLE session (
|
||||
id TEXT PRIMARY KEY,
|
||||
time_created INTEGER NOT NULL,
|
||||
time_updated INTEGER NOT NULL
|
||||
);
|
||||
INSERT INTO session VALUES ('ses_locked', 1777634000000, 1777634001000);
|
||||
BEGIN EXCLUSIVE;
|
||||
`)
|
||||
return { writer, path }
|
||||
writer.exec(schema)
|
||||
writer.close()
|
||||
return { directory, path }
|
||||
}
|
||||
|
||||
describe('listOpenCodeSqliteSessions under a live writer lock', () => {
|
||||
it('lists sessions from a contended database without a raw sqlite lock string', async () => {
|
||||
const { writer, path } = createLockedOpenCodeDb()
|
||||
async function holdExclusive(directory: string, dbPath: string): Promise<Worker> {
|
||||
const workerPath = join(directory, 'holder.mjs')
|
||||
await writeFile(workerPath, HOLDER_SOURCE)
|
||||
const worker = new Worker(workerPath, { workerData: { dbPath } })
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
worker.once('error', reject)
|
||||
worker.on('message', (message) => {
|
||||
if (message === 'locked') {
|
||||
resolve()
|
||||
}
|
||||
})
|
||||
})
|
||||
return worker
|
||||
}
|
||||
|
||||
describe('OpenCode session SQLite under transient writer contention', () => {
|
||||
it('lists sessions after an exclusive writer releases within the timeout', async () => {
|
||||
const { directory, path } = await createDatabase(`
|
||||
CREATE TABLE session (
|
||||
id TEXT PRIMARY KEY,
|
||||
time_created INTEGER NOT NULL,
|
||||
time_updated INTEGER NOT NULL
|
||||
);
|
||||
INSERT INTO session VALUES ('ses_locked', 1777634000000, 1777634001000);
|
||||
`)
|
||||
const worker = await holdExclusive(directory, path)
|
||||
try {
|
||||
worker.postMessage('reader-started')
|
||||
const issues: AiVaultScanIssue[] = []
|
||||
const candidates = await listOpenCodeSqliteSessions({
|
||||
dbPaths: [path],
|
||||
limit: 10,
|
||||
issues
|
||||
})
|
||||
expect(JSON.stringify(issues)).not.toMatch(/database is locked/i)
|
||||
|
||||
expect(issues).toEqual([])
|
||||
expect(candidates.map((candidate) => candidate.file.path)).toEqual([
|
||||
buildOpenCodeSqliteCandidatePath(path, 'ses_locked')
|
||||
])
|
||||
expect(issues).toEqual([])
|
||||
} finally {
|
||||
writer.exec('COMMIT')
|
||||
writer.close()
|
||||
await worker.terminate()
|
||||
}
|
||||
})
|
||||
|
||||
it('surfaces a retryable unavailable notice when the contended DB cannot be snapshotted', async () => {
|
||||
const { writer, path } = createLockedOpenCodeDb()
|
||||
setLiveSqliteFileCopyForTests(() => {
|
||||
throw new Error('EPERM: copy blocked')
|
||||
})
|
||||
try {
|
||||
const issues: AiVaultScanIssue[] = []
|
||||
const candidates = await listOpenCodeSqliteSessions({
|
||||
dbPaths: [path],
|
||||
limit: 10,
|
||||
issues
|
||||
})
|
||||
expect(candidates).toEqual([])
|
||||
expect(issues).toHaveLength(1)
|
||||
expect(issues[0]?.kind).toBe('notice')
|
||||
expect(issues[0]?.agent).toBe('opencode')
|
||||
expect(issues[0]?.path).toBe(path)
|
||||
expect(issues[0]?.message).toMatch(/temporarily unavailable/i)
|
||||
expect(issues[0]?.message).not.toMatch(/database is locked/i)
|
||||
expect(issues.filter((issue) => !issue.kind)).toEqual([])
|
||||
} finally {
|
||||
writer.exec('COMMIT')
|
||||
writer.close()
|
||||
}
|
||||
})
|
||||
|
||||
it('recovers the session list on the next scan after the writer releases', async () => {
|
||||
const { writer, path } = createLockedOpenCodeDb()
|
||||
setLiveSqliteFileCopyForTests(() => {
|
||||
throw new Error('EPERM: copy blocked')
|
||||
})
|
||||
const lockedIssues: AiVaultScanIssue[] = []
|
||||
const locked = await listOpenCodeSqliteSessions({
|
||||
dbPaths: [path],
|
||||
limit: 10,
|
||||
issues: lockedIssues
|
||||
})
|
||||
expect(locked).toEqual([])
|
||||
expect(lockedIssues[0]?.kind).toBe('notice')
|
||||
|
||||
setLiveSqliteFileCopyForTests(null)
|
||||
writer.exec('COMMIT')
|
||||
writer.close()
|
||||
|
||||
const recoveredIssues: AiVaultScanIssue[] = []
|
||||
const recovered = await listOpenCodeSqliteSessions({
|
||||
dbPaths: [path],
|
||||
limit: 10,
|
||||
issues: recoveredIssues
|
||||
})
|
||||
expect(recoveredIssues).toEqual([])
|
||||
expect(recovered.map((candidate) => candidate.file.path)).toEqual([
|
||||
buildOpenCodeSqliteCandidatePath(path, 'ses_locked')
|
||||
])
|
||||
})
|
||||
|
||||
it('still lists an uncontended readable database', async () => {
|
||||
const { writer, path } = createLockedOpenCodeDb()
|
||||
writer.exec('COMMIT')
|
||||
writer.close()
|
||||
|
||||
const issues: AiVaultScanIssue[] = []
|
||||
const candidates = await listOpenCodeSqliteSessions({
|
||||
dbPaths: [path],
|
||||
limit: 10,
|
||||
issues
|
||||
})
|
||||
expect(issues).toEqual([])
|
||||
expect(candidates).toHaveLength(1)
|
||||
})
|
||||
})
|
||||
|
||||
describe('parseOpenCodeSqliteSession under a live writer lock', () => {
|
||||
it('parses a contended session without throwing a raw sqlite lock string', async () => {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'orca-opencode-sqlite-parse-lock-'))
|
||||
tempDirs.push(dir)
|
||||
const path = join(dir, 'opencode.db')
|
||||
const writer = new Database(path)
|
||||
writer.exec(`
|
||||
it('parses a session after an exclusive writer releases within the timeout', async () => {
|
||||
const { directory, path } = await createDatabase(`
|
||||
CREATE TABLE session (
|
||||
id TEXT PRIMARY KEY,
|
||||
project_id TEXT NOT NULL,
|
||||
@@ -168,20 +119,21 @@ describe('parseOpenCodeSqliteSession under a live writer lock', () => {
|
||||
'ses_locked', 'proj-1', 'slug', '/tmp/opencode', 'Locked session', '1.0.0',
|
||||
1777634000000, 1777634001000, 0, 1, 1, 0, 0, 0
|
||||
);
|
||||
BEGIN EXCLUSIVE;
|
||||
`)
|
||||
const worker = await holdExclusive(directory, path)
|
||||
try {
|
||||
worker.postMessage('reader-started')
|
||||
const session = await parseOpenCodeSqliteSession({
|
||||
dbPath: path,
|
||||
sessionId: 'ses_locked',
|
||||
platform: 'darwin'
|
||||
})
|
||||
|
||||
expect(session).not.toBeNull()
|
||||
expect(session?.sessionId).toBe('ses_locked')
|
||||
expect(session?.title).toBe('Locked session')
|
||||
} finally {
|
||||
writer.exec('COMMIT')
|
||||
writer.close()
|
||||
await worker.terminate()
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
@@ -10,7 +10,7 @@ import {
|
||||
shouldCaptureFullFirstUserPrompt
|
||||
} from './session-scanner-first-user-prompt'
|
||||
import { normalizeTitleText } from './session-scanner-values'
|
||||
import { openLiveSqliteReadonly } from '../sqlite/live-sqlite-readonly'
|
||||
import { openReadonlySyncDatabase } from '../sqlite/readonly-sync-database'
|
||||
import type SyncDatabase from '../sqlite/sync-database'
|
||||
import { columnExists, tableExists } from '../opencode-usage/schema-helpers'
|
||||
|
||||
@@ -252,7 +252,7 @@ export async function parseOpenCodeSqliteSession(args: {
|
||||
const { dbPath, sessionId, platform } = args
|
||||
let db: SyncDatabase | null = null
|
||||
try {
|
||||
db = openLiveSqliteReadonly(dbPath)
|
||||
db = openReadonlySyncDatabase(dbPath)
|
||||
if (!canReadOpenCodeSessions(db)) {
|
||||
return null
|
||||
}
|
||||
|
||||
@@ -1,10 +1,6 @@
|
||||
import { readdirSync } from 'node:fs'
|
||||
import { join } from 'node:path'
|
||||
import {
|
||||
isLiveSqliteUnavailableError,
|
||||
openLiveSqliteReadonly
|
||||
} from '../sqlite/live-sqlite-readonly'
|
||||
import type SyncDatabase from '../sqlite/sync-database'
|
||||
import SyncDatabase from '../sqlite/sync-database'
|
||||
|
||||
const STATE_DB_FILE_PATTERN = /^state_(\d+)\.sqlite$/
|
||||
|
||||
@@ -44,7 +40,7 @@ export function readCodexStateDbBackfillStatus(codexHomePath: string): CodexStat
|
||||
}
|
||||
let db: SyncDatabase | null = null
|
||||
try {
|
||||
db = openLiveSqliteReadonly(stateDbPath)
|
||||
db = new SyncDatabase(stateDbPath, { readonly: true, fileMustExist: true })
|
||||
const table = db
|
||||
.prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'backfill_state'")
|
||||
.get()
|
||||
@@ -64,11 +60,7 @@ export function readCodexStateDbBackfillStatus(codexHomePath: string): CodexStat
|
||||
return {
|
||||
kind: 'unreadable',
|
||||
stateDbPath,
|
||||
error: isLiveSqliteUnavailableError(error)
|
||||
? 'Codex state is temporarily unavailable.'
|
||||
: error instanceof Error
|
||||
? error.message
|
||||
: String(error)
|
||||
error: error instanceof Error ? error.message : String(error)
|
||||
}
|
||||
} finally {
|
||||
try {
|
||||
|
||||
@@ -1,8 +1,5 @@
|
||||
import { yieldToEventLoop } from '../../shared/event-loop-yield'
|
||||
import {
|
||||
isLiveSqliteUnavailableError,
|
||||
openLiveSqliteReadonly
|
||||
} from '../sqlite/live-sqlite-readonly'
|
||||
import Database from '../sqlite/sync-database'
|
||||
import { createUsageEventAggregation } from '../usage/usage-event-aggregation'
|
||||
import {
|
||||
compareOpenCodeClaimPriority,
|
||||
@@ -57,9 +54,9 @@ export async function parseOpenCodeUsageDatabase(
|
||||
options: { claimSession?: (sessionId: string) => boolean } = {}
|
||||
): Promise<OpenCodeUsagePersistedDatabase> {
|
||||
const processedDatabase = await getProcessedDatabaseInfo(dbPath)
|
||||
let db: ReturnType<typeof openLiveSqliteReadonly> | null = null
|
||||
const db = new Database(dbPath, { readonly: true, fileMustExist: true })
|
||||
try {
|
||||
db = openLiveSqliteReadonly(dbPath)
|
||||
db.pragma('query_only = ON')
|
||||
const events: OpenCodeUsageAttributedEvent[] = []
|
||||
const claimedBySessionId = new Map<string, boolean>()
|
||||
let hasDeferredClaims = false
|
||||
@@ -92,19 +89,8 @@ export async function parseOpenCodeUsageDatabase(
|
||||
.map(([sessionId]) => sessionId),
|
||||
hasDeferredClaims
|
||||
}
|
||||
} catch (error) {
|
||||
if (isLiveSqliteUnavailableError(error)) {
|
||||
return {
|
||||
...processedDatabase,
|
||||
sessions: [],
|
||||
dailyAggregates: [],
|
||||
ownedSessionIds: [],
|
||||
hasDeferredClaims: false
|
||||
}
|
||||
}
|
||||
throw error
|
||||
} finally {
|
||||
db?.close()
|
||||
db.close()
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1,102 +0,0 @@
|
||||
import { existsSync, mkdtempSync, rmSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import {
|
||||
flushLiveSqliteSnapshotsForTests,
|
||||
isLiveSqliteUnavailableError,
|
||||
LiveSqliteUnavailableError,
|
||||
openLiveSqliteReadonly,
|
||||
setLiveSqliteFileCopyForTests
|
||||
} from './live-sqlite-readonly'
|
||||
import SyncDatabase from './sync-database'
|
||||
|
||||
let tempDirs: string[] = []
|
||||
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks()
|
||||
setLiveSqliteFileCopyForTests(null)
|
||||
flushLiveSqliteSnapshotsForTests()
|
||||
for (const dir of tempDirs) {
|
||||
rmSync(dir, { recursive: true, force: true })
|
||||
}
|
||||
tempDirs = []
|
||||
})
|
||||
|
||||
function createLockedDb(): { writer: SyncDatabase; path: string } {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'orca-live-sqlite-'))
|
||||
tempDirs.push(dir)
|
||||
const path = join(dir, 'sessions.db')
|
||||
const writer = new SyncDatabase(path)
|
||||
writer.exec('CREATE TABLE items (id INTEGER); INSERT INTO items VALUES (1); BEGIN EXCLUSIVE')
|
||||
return { writer, path }
|
||||
}
|
||||
|
||||
describe('openLiveSqliteReadonly', () => {
|
||||
it('reads an uncontended database in place', () => {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'orca-live-sqlite-'))
|
||||
tempDirs.push(dir)
|
||||
const path = join(dir, 'sessions.db')
|
||||
const writer = new SyncDatabase(path)
|
||||
writer.exec('CREATE TABLE items (id INTEGER); INSERT INTO items VALUES (7)')
|
||||
writer.close()
|
||||
|
||||
const db = openLiveSqliteReadonly(path)
|
||||
expect(db.pragma('query_only', { simple: true })).toBe(1)
|
||||
expect(db.prepare('SELECT id FROM items').get()).toEqual({ id: 7 })
|
||||
expect(() => db.exec('INSERT INTO items VALUES (8)')).toThrow()
|
||||
db.close()
|
||||
})
|
||||
|
||||
it('reads through a snapshot when the live database is exclusively locked', () => {
|
||||
const { writer, path } = createLockedDb()
|
||||
try {
|
||||
const db = openLiveSqliteReadonly(path)
|
||||
expect(db.prepare('SELECT id FROM items').get()).toEqual({ id: 1 })
|
||||
db.close()
|
||||
expect(existsSync(path)).toBe(true)
|
||||
expect(existsSync(`${path}-wal`)).toBe(false)
|
||||
} finally {
|
||||
writer.exec('COMMIT')
|
||||
writer.close()
|
||||
}
|
||||
})
|
||||
|
||||
it('throws a retryable unavailable error when a locked database cannot be copied', () => {
|
||||
const { writer, path } = createLockedDb()
|
||||
setLiveSqliteFileCopyForTests(() => {
|
||||
throw new Error('EPERM: copy blocked')
|
||||
})
|
||||
try {
|
||||
expect(() => openLiveSqliteReadonly(path)).toThrow(LiveSqliteUnavailableError)
|
||||
try {
|
||||
openLiveSqliteReadonly(path)
|
||||
} catch (error) {
|
||||
expect(isLiveSqliteUnavailableError(error)).toBe(true)
|
||||
expect(error).toBeInstanceOf(LiveSqliteUnavailableError)
|
||||
expect(String(error)).not.toMatch(/database is locked/i)
|
||||
}
|
||||
} finally {
|
||||
writer.exec('COMMIT')
|
||||
writer.close()
|
||||
}
|
||||
})
|
||||
|
||||
it('closes the handle when query_only setup fails on an uncontended file', () => {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'orca-live-sqlite-'))
|
||||
tempDirs.push(dir)
|
||||
const path = join(dir, 'sessions.db')
|
||||
const writer = new SyncDatabase(path)
|
||||
writer.exec('CREATE TABLE items (id INTEGER)')
|
||||
writer.close()
|
||||
|
||||
const setupError = new Error('query_only setup failed')
|
||||
vi.spyOn(SyncDatabase.prototype, 'pragma').mockImplementationOnce(() => {
|
||||
throw setupError
|
||||
})
|
||||
const closeSpy = vi.spyOn(SyncDatabase.prototype, 'close')
|
||||
|
||||
expect(() => openLiveSqliteReadonly(path)).toThrow(setupError)
|
||||
expect(closeSpy).toHaveBeenCalled()
|
||||
})
|
||||
})
|
||||
@@ -1,200 +0,0 @@
|
||||
// Why: a namespace import defers export resolution to call time. Named imports bind at
|
||||
// module evaluation, so any suite that partially mocks `fs` fails the moment this module
|
||||
// is pulled into its graph - even when it never touches sqlite.
|
||||
import * as nodeFs from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import SyncDatabase from './sync-database'
|
||||
|
||||
// Why: a live agent writer can hold an exclusive lock for the whole process.
|
||||
// Waiting on busy_timeout just stalls the scan; snapshot the files instead.
|
||||
const SNAPSHOT_IDLE_MS = 15_000
|
||||
const SQLITE_BUSY_PATTERN =
|
||||
/database is locked|database is busy|database table is locked|SQLITE_BUSY|SQLITE_LOCKED|EPERM|EBUSY/i
|
||||
|
||||
export const LIVE_SQLITE_UNAVAILABLE_MESSAGE =
|
||||
'Session history is temporarily unavailable. It will refresh on the next scan.'
|
||||
|
||||
export class LiveSqliteUnavailableError extends Error {
|
||||
constructor(cause?: unknown) {
|
||||
super(LIVE_SQLITE_UNAVAILABLE_MESSAGE)
|
||||
this.name = 'LiveSqliteUnavailableError'
|
||||
if (cause instanceof Error) {
|
||||
this.cause = cause
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export function isSqliteBusyError(error: unknown): boolean {
|
||||
const message = error instanceof Error ? error.message : String(error)
|
||||
return SQLITE_BUSY_PATTERN.test(message)
|
||||
}
|
||||
|
||||
export function isLiveSqliteUnavailableError(error: unknown): boolean {
|
||||
if (error instanceof LiveSqliteUnavailableError) {
|
||||
return true
|
||||
}
|
||||
const message = error instanceof Error ? error.message : String(error)
|
||||
return /temporarily unavailable/i.test(message) || isSqliteBusyError(error)
|
||||
}
|
||||
|
||||
type LiveSqliteFileCopy = (source: string, dest: string) => void
|
||||
|
||||
const DEFAULT_LIVE_SQLITE_FILE_COPY: LiveSqliteFileCopy = (src, dest) =>
|
||||
nodeFs.copyFileSync(src, dest)
|
||||
|
||||
let copyLiveSqliteFile: LiveSqliteFileCopy = DEFAULT_LIVE_SQLITE_FILE_COPY
|
||||
|
||||
export function setLiveSqliteFileCopyForTests(copy: LiveSqliteFileCopy | null): void {
|
||||
copyLiveSqliteFile = copy ?? DEFAULT_LIVE_SQLITE_FILE_COPY
|
||||
}
|
||||
|
||||
type SnapshotLease = {
|
||||
dir: string
|
||||
dbFile: string
|
||||
refs: number
|
||||
idleTimer: ReturnType<typeof setTimeout> | null
|
||||
}
|
||||
|
||||
const snapshotLeases = new Map<string, SnapshotLease>()
|
||||
|
||||
class SnapshotReadonlyDatabase extends SyncDatabase {
|
||||
constructor(
|
||||
snapshotDbPath: string,
|
||||
private readonly releaseSnapshot: () => void
|
||||
) {
|
||||
super(snapshotDbPath, { readonly: true, fileMustExist: true, timeout: 0 })
|
||||
}
|
||||
|
||||
close(): void {
|
||||
try {
|
||||
super.close()
|
||||
} finally {
|
||||
this.releaseSnapshot()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function configureQueryOnly(db: SyncDatabase): SyncDatabase {
|
||||
try {
|
||||
db.pragma('query_only = ON')
|
||||
// Open itself can succeed while the first real read still takes SHARED and
|
||||
// throws SQLITE_BUSY, so probe before handing the handle to callers.
|
||||
db.prepare('SELECT 1 AS ok FROM sqlite_master LIMIT 1').get()
|
||||
return db
|
||||
} catch (error) {
|
||||
try {
|
||||
db.close()
|
||||
} catch {
|
||||
// Preserve the open/probe failure.
|
||||
}
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
function copyLiveSqliteSnapshot(sourcePath: string): SnapshotLease {
|
||||
const dir = nodeFs.mkdtempSync(join(tmpdir(), 'orca-live-sqlite-'))
|
||||
const dbFile = join(dir, 'db.sqlite')
|
||||
try {
|
||||
copyLiveSqliteFile(sourcePath, dbFile)
|
||||
const walPath = `${sourcePath}-wal`
|
||||
if (nodeFs.existsSync(walPath)) {
|
||||
copyLiveSqliteFile(walPath, `${dbFile}-wal`)
|
||||
}
|
||||
} catch (error) {
|
||||
nodeFs.rmSync(dir, { recursive: true, force: true })
|
||||
throw error
|
||||
}
|
||||
return { dir, dbFile, refs: 0, idleTimer: null }
|
||||
}
|
||||
|
||||
function clearSnapshotIdle(lease: SnapshotLease): void {
|
||||
if (lease.idleTimer) {
|
||||
clearTimeout(lease.idleTimer)
|
||||
lease.idleTimer = null
|
||||
}
|
||||
}
|
||||
|
||||
function disposeSnapshot(sourcePath: string, lease: SnapshotLease): void {
|
||||
clearSnapshotIdle(lease)
|
||||
snapshotLeases.delete(sourcePath)
|
||||
nodeFs.rmSync(lease.dir, { recursive: true, force: true })
|
||||
}
|
||||
|
||||
function acquireSnapshot(sourcePath: string): { dbFile: string; release: () => void } {
|
||||
let lease = snapshotLeases.get(sourcePath)
|
||||
if (!lease) {
|
||||
lease = copyLiveSqliteSnapshot(sourcePath)
|
||||
snapshotLeases.set(sourcePath, lease)
|
||||
} else {
|
||||
clearSnapshotIdle(lease)
|
||||
}
|
||||
lease.refs += 1
|
||||
let released = false
|
||||
return {
|
||||
dbFile: lease.dbFile,
|
||||
release: () => {
|
||||
if (released) {
|
||||
return
|
||||
}
|
||||
released = true
|
||||
const current = snapshotLeases.get(sourcePath)
|
||||
if (!current) {
|
||||
return
|
||||
}
|
||||
current.refs = Math.max(0, current.refs - 1)
|
||||
if (current.refs > 0) {
|
||||
return
|
||||
}
|
||||
clearSnapshotIdle(current)
|
||||
current.idleTimer = setTimeout(() => {
|
||||
const idle = snapshotLeases.get(sourcePath)
|
||||
if (idle && idle.refs === 0) {
|
||||
disposeSnapshot(sourcePath, idle)
|
||||
}
|
||||
}, SNAPSHOT_IDLE_MS)
|
||||
current.idleTimer.unref?.()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function openSnapshotReadonly(sourcePath: string): SyncDatabase {
|
||||
const snapshot = acquireSnapshot(sourcePath)
|
||||
try {
|
||||
return configureQueryOnly(new SnapshotReadonlyDatabase(snapshot.dbFile, snapshot.release))
|
||||
} catch (error) {
|
||||
snapshot.release()
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
export function flushLiveSqliteSnapshotsForTests(): void {
|
||||
for (const [sourcePath, lease] of snapshotLeases) {
|
||||
disposeSnapshot(sourcePath, lease)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Open another process's SQLite database for reading without waiting on its
|
||||
* write lock. Uncontended files open in place; a busy database is copied
|
||||
* (db + WAL only) and the copy is read. The original files are never mutated,
|
||||
* checkpointed, or deleted.
|
||||
*/
|
||||
export function openLiveSqliteReadonly(dbPath: string): SyncDatabase {
|
||||
try {
|
||||
return configureQueryOnly(
|
||||
new SyncDatabase(dbPath, { readonly: true, fileMustExist: true, timeout: 0 })
|
||||
)
|
||||
} catch (error) {
|
||||
if (!isSqliteBusyError(error)) {
|
||||
throw error
|
||||
}
|
||||
}
|
||||
try {
|
||||
return openSnapshotReadonly(dbPath)
|
||||
} catch (error) {
|
||||
throw error instanceof LiveSqliteUnavailableError
|
||||
? error
|
||||
: new LiveSqliteUnavailableError(error)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,124 @@
|
||||
import { mkdtemp, rm, writeFile } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { Worker } from 'node:worker_threads'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { openReadonlySyncDatabase } from './readonly-sync-database'
|
||||
import SyncDatabase from './sync-database'
|
||||
|
||||
const temporaryDirectories: string[] = []
|
||||
|
||||
afterEach(async () => {
|
||||
vi.restoreAllMocks()
|
||||
await Promise.all(
|
||||
temporaryDirectories
|
||||
.splice(0)
|
||||
.map((directory) => rm(directory, { recursive: true, force: true }))
|
||||
)
|
||||
})
|
||||
|
||||
const HOLDER_SOURCE = `
|
||||
import { parentPort, workerData } from 'node:worker_threads'
|
||||
const { DatabaseSync } = process.getBuiltinModule('node:sqlite')
|
||||
const db = new DatabaseSync(workerData.dbPath)
|
||||
db.exec('BEGIN EXCLUSIVE')
|
||||
parentPort.postMessage('locked')
|
||||
parentPort.once('message', (message) => {
|
||||
if (message !== 'reader-started') throw new Error('Unexpected lock-holder message')
|
||||
setTimeout(() => {
|
||||
db.exec('COMMIT')
|
||||
db.close()
|
||||
parentPort.postMessage('committed')
|
||||
}, workerData.releaseDelayMs)
|
||||
})
|
||||
`
|
||||
|
||||
async function createDatabase(): Promise<{ directory: string; path: string }> {
|
||||
const directory = await mkdtemp(join(tmpdir(), 'orca-readonly-sqlite-'))
|
||||
temporaryDirectories.push(directory)
|
||||
const path = join(directory, 'sessions.db')
|
||||
const writer = new SyncDatabase(path)
|
||||
writer.exec('CREATE TABLE items (id INTEGER); INSERT INTO items VALUES (1)')
|
||||
writer.close()
|
||||
return { directory, path }
|
||||
}
|
||||
|
||||
async function holdExclusive(
|
||||
directory: string,
|
||||
dbPath: string,
|
||||
releaseDelayMs: number
|
||||
): Promise<Worker> {
|
||||
const workerPath = join(directory, 'holder.mjs')
|
||||
await writeFile(workerPath, HOLDER_SOURCE)
|
||||
const worker = new Worker(workerPath, { workerData: { dbPath, releaseDelayMs } })
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
worker.once('error', reject)
|
||||
worker.on('message', (message) => {
|
||||
if (message === 'locked') {
|
||||
resolve()
|
||||
}
|
||||
})
|
||||
})
|
||||
return worker
|
||||
}
|
||||
|
||||
describe('openReadonlySyncDatabase', () => {
|
||||
it('reads after a concurrent exclusive writer releases within the busy timeout', async () => {
|
||||
const { directory, path } = await createDatabase()
|
||||
const worker = await holdExclusive(directory, path, 200)
|
||||
try {
|
||||
worker.postMessage('reader-started')
|
||||
const db = openReadonlySyncDatabase(path, { timeoutMs: 2_000 })
|
||||
expect(db.prepare('SELECT id FROM items').get()).toEqual({ id: 1 })
|
||||
db.close()
|
||||
} finally {
|
||||
await worker.terminate()
|
||||
}
|
||||
})
|
||||
|
||||
it('fails when the writer outlives the busy timeout', async () => {
|
||||
const { directory, path } = await createDatabase()
|
||||
const worker = await holdExclusive(directory, path, 2_000)
|
||||
try {
|
||||
worker.postMessage('reader-started')
|
||||
expect(() => openReadonlySyncDatabase(path, { timeoutMs: 50 })).toThrow(/database is locked/i)
|
||||
} finally {
|
||||
await worker.terminate()
|
||||
}
|
||||
})
|
||||
|
||||
it('keeps the handle query-only', async () => {
|
||||
const { path } = await createDatabase()
|
||||
const db = openReadonlySyncDatabase(path)
|
||||
expect(db.pragma('query_only', { simple: true })).toBe(1)
|
||||
expect(() => db.exec('INSERT INTO items VALUES (2)')).toThrow()
|
||||
db.close()
|
||||
})
|
||||
|
||||
it('closes the handle when setup fails', async () => {
|
||||
const { path } = await createDatabase()
|
||||
const setupError = new Error('query_only setup failed')
|
||||
vi.spyOn(SyncDatabase.prototype, 'pragma').mockImplementationOnce(() => {
|
||||
throw setupError
|
||||
})
|
||||
const closeSpy = vi.spyOn(SyncDatabase.prototype, 'close')
|
||||
|
||||
expect(() => openReadonlySyncDatabase(path)).toThrow(setupError)
|
||||
expect(closeSpy).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('preserves the setup error when close also fails', async () => {
|
||||
const { path } = await createDatabase()
|
||||
const setupError = new Error('query_only setup failed')
|
||||
const originalClose = SyncDatabase.prototype.close
|
||||
vi.spyOn(SyncDatabase.prototype, 'pragma').mockImplementationOnce(() => {
|
||||
throw setupError
|
||||
})
|
||||
vi.spyOn(SyncDatabase.prototype, 'close').mockImplementationOnce(function (this: SyncDatabase) {
|
||||
originalClose.call(this)
|
||||
throw new Error('close failed')
|
||||
})
|
||||
|
||||
expect(() => openReadonlySyncDatabase(path)).toThrow(setupError)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,36 @@
|
||||
import SyncDatabase from './sync-database'
|
||||
|
||||
// Why: OpenCode uses WAL, but brief exclusive transitions can still race a reader on Windows.
|
||||
export const READONLY_SQLITE_BUSY_TIMEOUT_MS = 5_000
|
||||
|
||||
const SQLITE_BUSY_PATTERN =
|
||||
/database is locked|database is busy|database table is locked|SQLITE_BUSY|SQLITE_LOCKED/i
|
||||
|
||||
export function isSqliteBusyError(error: unknown): boolean {
|
||||
const message = error instanceof Error ? error.message : String(error)
|
||||
return SQLITE_BUSY_PATTERN.test(message)
|
||||
}
|
||||
|
||||
export function openReadonlySyncDatabase(
|
||||
dbPath: string,
|
||||
options: { fileMustExist?: boolean; timeoutMs?: number } = {}
|
||||
): SyncDatabase {
|
||||
const db = new SyncDatabase(dbPath, {
|
||||
readonly: true,
|
||||
fileMustExist: options.fileMustExist ?? true,
|
||||
timeout: options.timeoutMs ?? READONLY_SQLITE_BUSY_TIMEOUT_MS
|
||||
})
|
||||
try {
|
||||
db.pragma('query_only = ON')
|
||||
// Open can succeed before the first read takes a shared lock.
|
||||
db.prepare('SELECT 1 AS ok FROM sqlite_master LIMIT 1').get()
|
||||
return db
|
||||
} catch (error) {
|
||||
try {
|
||||
db.close()
|
||||
} catch {
|
||||
// Preserve the initialization failure.
|
||||
}
|
||||
throw error
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user