diff --git a/CONTEXT.md b/CONTEXT.md index c4babc4..04cca63 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -67,10 +67,18 @@ incremental indexing), plus heartbeat/last-build markers used for daemon arbitration. **Daemon arbitration**: -The mechanism by which the passive pull mode detects a fresh daemon (via -`__app_heartbeat__` and `__app_last_successful_build__` markers in `index_state`) -and skips its own indexing. Because a fresh daemon owns writes, the two persist -layers never write concurrently. +The policy by which the passive pull mode detects a fresh daemon from the +`__app_heartbeat__` marker and skips every skill-side mutation, including schema +setup, indexing, checkpointing, and `attune`. The heartbeat alone means “the +daemon should write”; `__app_last_successful_build__` records coverage/freshness, +not ownership. Both indexing modes use the same persist layer. + +**Writer lease**: +The hard cross-process safety mutex behind daemon arbitration. A writer holds +`BEGIN IMMEDIATE` on `.obelisk/writer.lock.sqlite` for the complete mutation; +manual rebuild holds it through build, target-database replacement, and reopen. +The heartbeat expresses policy, while the writer lease prevents overlapping +writes during races, stale heartbeats, or processes from different versions. ## Memory diff --git a/README.md b/README.md index 55403dd..dce23b7 100644 --- a/README.md +++ b/README.md @@ -191,9 +191,10 @@ Full-text search via FTS5 covers message text across every session layer and ran The index rebuilds incrementally — only new or modified JSONL files are re-parsed. When the optional app is running, it is the active indexer: it watches Claude -project files, builds in a worker thread, writes `__app_heartbeat__` plus -`__app_last_successful_build__` into `index_state`, and the skill-side lazy -build skips work only while both markers are fresh. +project files and builds in a worker thread. A fresh `__app_heartbeat__` alone +means the daemon owns writes, so the skill remains read-only; a separate SQLite +writer lease prevents cross-process writes from overlapping. The +`__app_last_successful_build__` marker records index freshness, not ownership. Zero npm dependencies. Uses Node 22's built-in node:sqlite with FTS5. The entire runtime is ~400 lines. diff --git a/app/package.json b/app/package.json index 49e9a16..433caa8 100644 --- a/app/package.json +++ b/app/package.json @@ -11,7 +11,8 @@ "dist": "electron-vite build && electron-builder --mac --win --linux", "dist:mac": "electron-vite build && electron-builder --mac", "dist:win": "electron-vite build && electron-builder --win", - "dist:linux": "electron-vite build && electron-builder --linux" + "dist:linux": "electron-vite build && electron-builder --linux", + "test:electron": "electron-vite build && electron --no-sandbox tests/electron-concurrency.mjs" }, "build": { "appId": "com.obelisk.app", diff --git a/app/src/main/index.ts b/app/src/main/index.ts index ebdb5fb..8f35ad4 100644 --- a/app/src/main/index.ts +++ b/app/src/main/index.ts @@ -9,6 +9,7 @@ import { writeHeartbeat } from './indexer.ts'; import { createIndexerService } from './indexer-service.ts'; import { createWorkerBuildIndex } from './indexer-worker-client.ts'; import { buildRecapExportQuery } from './recap-capture-query.ts'; +import { acquireWriterLease, writerLockPathFor } from '../../../scripts/writer-lease.ts'; const __dirname = path.dirname(fileURLToPath(import.meta.url)); @@ -41,6 +42,16 @@ let db; let indexerService; let indexerWorker; +type WriterLeaseMode = 'acquire' | 'caller-held'; + +function acquireAppWriterLease(dbPath: string, waitMs = 0) { + return acquireWriterLease({ + lockPath: writerLockPathFor(dbPath), + openDb: lockPath => new Database(lockPath), + waitMs, + }); +} + function getConfiguredClaudeDir() { const persisted = loadPersistedSettings(); return persisted.claudeDir || DEFAULT_CLAUDE_DIR; @@ -60,15 +71,25 @@ function getPathsForClaudeDir(claudeDir = getConfiguredClaudeDir(), codexDir = g }; } -function migrateLegacyDbIfNeeded(paths = getPathsForClaudeDir()) { +function migrateLegacyDbIfNeeded( + paths = getPathsForClaudeDir(), + { writerLeaseMode = 'acquire' }: { writerLeaseMode?: WriterLeaseMode } = {}, +) { if (fs.existsSync(paths.dbPath)) return; const legacyDbPath = path.join(paths.claudeDir, 'obelisk.sqlite'); if (!fs.existsSync(legacyDbPath)) return; + const lease = writerLeaseMode === 'acquire' ? acquireAppWriterLease(paths.dbPath) : null; + if (writerLeaseMode === 'acquire' && !lease) return false; try { + if (fs.existsSync(paths.dbPath)) return true; fs.mkdirSync(path.dirname(paths.dbPath), { recursive: true }); fs.copyFileSync(legacyDbPath, paths.dbPath); + return true; } catch (error) { console.warn?.(`Obelisk legacy DB migration skipped: ${(error as Error).message}`); + return false; + } finally { + lease?.release(); } } @@ -151,15 +172,40 @@ function closeDb() { db = null; } -function openDb(dbPath = getPathsForClaudeDir().dbPath) { +function openDb( + dbPath = getPathsForClaudeDir().dbPath, + { writerLeaseMode = 'acquire' }: { writerLeaseMode?: WriterLeaseMode } = {}, +) { closeDb(); if (!fs.existsSync(dbPath)) return null; db = new Database(dbPath, { readonly: false }); - db.pragma('journal_mode = WAL'); - migrateDb(db); + db.pragma('busy_timeout = 5000'); + const lease = writerLeaseMode === 'acquire' ? acquireAppWriterLease(dbPath) : null; + if (writerLeaseMode === 'caller-held' || lease) { + try { + db.pragma('journal_mode = WAL'); + migrateDb(db); + } finally { + lease?.release(); + } + } return db; } +function runAppDbWrite(work: () => void): boolean { + if (!db) return false; + const lease = acquireAppWriterLease(getPathsForClaudeDir().dbPath, 250); + if (!lease) { + throw new Error('Obelisk index writer is busy; memory change was not applied'); + } + try { + work(); + return true; + } finally { + lease.release(); + } +} + function notifyIndexUpdated(result: { affectedSessionIds?: unknown } = {}) { const affectedSessionIds = Array.isArray(result.affectedSessionIds) ? [...new Set(result.affectedSessionIds.filter(Boolean))] @@ -200,8 +246,14 @@ function startIndexerService({ buildOnStart = false } = {}) { projectsDir: paths.projectsDir, dbPath: paths.dbPath, }); - openDb(paths.dbPath); - notifyIndexUpdated(result); + if (result?.deferred) { + if (Array.isArray(result.affectedSessionIds) && result.affectedSessionIds.length) { + notifyIndexUpdated(result); + } + } else { + openDb(paths.dbPath); + notifyIndexUpdated(result); + } return result; }, writeHeartbeat: () => writeHeartbeat({ dbPath: paths.dbPath }), @@ -520,16 +572,16 @@ ipcMain.handle('db:readMemoryFile', (_, filePath) => { }); ipcMain.handle('db:archiveMemory', (_, id, reason) => { - if (!db) return false; - db.prepare(`UPDATE memories SET deleted_at = ?, deleted_reason = ? WHERE id = ?`) - .run(new Date().toISOString(), reason || 'Archived via panel', id); - return true; + return runAppDbWrite(() => { + db.prepare(`UPDATE memories SET deleted_at = ?, deleted_reason = ? WHERE id = ?`) + .run(new Date().toISOString(), reason || 'Archived via panel', id); + }); }); ipcMain.handle('db:restoreMemory', (_, id) => { - if (!db) return false; - db.prepare(`UPDATE memories SET deleted_at = NULL, deleted_reason = NULL WHERE id = ?`).run(id); - return true; + return runAppDbWrite(() => { + db.prepare(`UPDATE memories SET deleted_at = NULL, deleted_reason = NULL WHERE id = ?`).run(id); + }); }); ipcMain.handle('db:getProjects', (_, opts = {}) => { @@ -830,8 +882,27 @@ ipcMain.handle('settings:rebuildIndex', async () => { indexerWorker = createWorkerBuildIndex(); } cleanupDbFiles(tempDbPath); + let writerLease: ReturnType = null; try { - migrateLegacyDbIfNeeded(paths); + const writerLeasePath = writerLockPathFor(paths.dbPath); + writerLease = acquireWriterLease({ + lockPath: writerLeasePath, + openDb: lockPath => new Database(lockPath), + waitMs: 2000, + }); + if (!writerLease) { + return { + files: 0, + latestSourceMtime: 0, + affectedSessionIds: [], + ftsRebuilt: false, + skipped: 0, + skippedFiles: [], + deferred: true, + reason: 'writer_busy', + }; + } + migrateLegacyDbIfNeeded(paths, { writerLeaseMode: 'caller-held' }); const result = await indexerWorker.buildIndex({ reason: 'manual-rebuild', force: true, @@ -840,21 +911,30 @@ ipcMain.handle('settings:rebuildIndex', async () => { projectsDir: paths.projectsDir, dbPath: tempDbPath, preserveDbPath: fs.existsSync(paths.dbPath) ? paths.dbPath : null, + writerLeasePath, + writerLeaseMode: 'caller-held', }); + if (result?.deferred) return result; closeDb(); replaceDbWithTemp(tempDbPath, paths.dbPath); - openDb(paths.dbPath); + openDb(paths.dbPath, { writerLeaseMode: 'caller-held' }); notifyIndexUpdated(result); return result; } finally { - cleanupDbFiles(tempDbPath); - if (!db) { - try { - openDb(paths.dbPath); - } catch (error) { - console.warn?.(`Obelisk DB reopen after rebuild failed: ${(error as Error).message}`); + try { + cleanupDbFiles(tempDbPath); + if (!db) { + try { + openDb(paths.dbPath, { + writerLeaseMode: writerLease ? 'caller-held' : 'acquire', + }); + } catch (error) { + console.warn?.(`Obelisk DB reopen after rebuild failed: ${(error as Error).message}`); + } } + } finally { + writerLease?.release(); + if (shouldRestartWatcher) startIndexerService({ buildOnStart: false }); } - if (shouldRestartWatcher) startIndexerService({ buildOnStart: false }); } }); diff --git a/app/src/main/indexer-service.ts b/app/src/main/indexer-service.ts index 407b310..78021ac 100644 --- a/app/src/main/indexer-service.ts +++ b/app/src/main/indexer-service.ts @@ -8,6 +8,7 @@ const DEFAULT_DEBOUNCE_MS = 2000; const DEFAULT_STABILITY_MS = 500; const DEFAULT_HEARTBEAT_MS = 30000; const DEFAULT_WATCH_RETRY_MS = 5000; +const DEFAULT_DEFERRED_RETRY_MS = 250; type TimerHandle = ReturnType; @@ -22,6 +23,15 @@ interface Watcher { close(): unknown; } +interface IndexerBuildResult { + deferred?: boolean; +} + +type IndexerBuild = (args: { + reason?: string; + changedPaths?: string[]; +}) => IndexerBuildResult | void | Promise; + interface IndexerServiceOptions { projectsDir?: string; watchDirs?: string | string[]; @@ -29,8 +39,9 @@ interface IndexerServiceOptions { stabilityMs?: number; heartbeatMs?: number; watchRetryMs?: number; - buildIndex?: (args: { reason?: string; changedPaths?: string[] }) => unknown; - writeHeartbeat?: () => void; + deferredRetryMs?: number; + buildIndex?: IndexerBuild; + writeHeartbeat?: () => unknown; watchProjects?: (onChange: (changedPath: string) => void) => Watcher | null; chokidar?: any; timers?: Timers; @@ -44,6 +55,7 @@ function createIndexerService({ stabilityMs = DEFAULT_STABILITY_MS, heartbeatMs = DEFAULT_HEARTBEAT_MS, watchRetryMs = DEFAULT_WATCH_RETRY_MS, + deferredRetryMs = DEFAULT_DEFERRED_RETRY_MS, buildIndex, writeHeartbeat = () => {}, watchProjects, @@ -100,6 +112,7 @@ function createIndexerService({ let stabilityTimer: TimerHandle | null = null; let heartbeatTimer: TimerHandle | null = null; let watchRetryTimer: TimerHandle | null = null; + let deferredRetryTimer: TimerHandle | null = null; let watcher: Watcher | null = null; let stopped = false; let running = false; @@ -124,6 +137,15 @@ function createIndexerService({ return paths; }; + const publishHeartbeat = () => { + try { + return writeHeartbeat(); + } catch (error) { + logger.warn?.(`Obelisk heartbeat failed: ${(error as Error).message}`); + return false; + } + }; + const runBuildNow = (reason = "manual", paths: string[] | undefined = undefined) => { addChangedPath(paths); if (stopped) return idlePromise; @@ -135,8 +157,18 @@ function createIndexerService({ pending = false; const buildChangedPaths = takeChangedPaths(); idlePromise = (async () => { - await buildIndex({ reason, changedPaths: buildChangedPaths }); - writeHeartbeat(); + const result = await buildIndex({ reason, changedPaths: buildChangedPaths }); + if (result?.deferred) { + addChangedPath(buildChangedPaths); + if (!stopped && !deferredRetryTimer) { + deferredRetryTimer = timers.setTimeout(() => { + deferredRetryTimer = null; + runBuildNow('writer-lease'); + }, deferredRetryMs); + } + return; + } + publishHeartbeat(); })() .catch((error) => { // A build in flight when the service is stopped (e.g. a manual rebuild @@ -158,6 +190,8 @@ function createIndexerService({ addChangedPath(changedPath); lastReason = reason; if (running) pending = true; + if (deferredRetryTimer) timers.clearTimeout(deferredRetryTimer); + deferredRetryTimer = null; if (buildTimer) timers.clearTimeout(buildTimer); if (stabilityTimer) timers.clearTimeout(stabilityTimer); buildTimer = timers.setTimeout(() => { @@ -186,15 +220,12 @@ function createIndexerService({ const start = ({ buildOnStart = true } = {}) => { stopped = false; + publishHeartbeat(); if (buildOnStart) scheduleBuild('startup'); startWatching(); if (typeof timers.setInterval === 'function') { heartbeatTimer = timers.setInterval(() => { - try { - writeHeartbeat(); - } catch (error) { - logger.warn?.(`Obelisk heartbeat failed: ${(error as Error).message}`); - } + publishHeartbeat(); }, heartbeatMs); } }; @@ -208,6 +239,8 @@ function createIndexerService({ stabilityTimer = null; if (watchRetryTimer) timers.clearTimeout(watchRetryTimer); watchRetryTimer = null; + if (deferredRetryTimer) timers.clearTimeout(deferredRetryTimer); + deferredRetryTimer = null; if (heartbeatTimer && typeof timers.clearInterval === 'function') timers.clearInterval(heartbeatTimer); heartbeatTimer = null; if (watcher?.close) watcher.close(); diff --git a/app/src/main/indexer.ts b/app/src/main/indexer.ts index d767952..c6211bb 100644 --- a/app/src/main/indexer.ts +++ b/app/src/main/indexer.ts @@ -6,6 +6,9 @@ import Database from 'better-sqlite3'; import { parse as claudeParse } from '../../../scripts/providers/claude.ts'; import { parse as codexParse } from '../../../scripts/providers/codex.ts'; import { persist } from '../../../scripts/persist.ts'; +import { runWriteTransaction, configureConnection, betterSqliteTransactionAdapter } from '../../../scripts/tx.ts'; +import { acquireWriterLease, writerLockPathFor } from '../../../scripts/writer-lease.ts'; +import { runRetryableWriteTransaction, isBeginBusyFailure, hasUnusableTransaction } from '../../../scripts/write-coordinator.ts'; import { inferProjectPath, isDir, @@ -35,15 +38,6 @@ interface FileInfo { source?: string; } -// A rollback inside a catch must never throw over the real error. SQLite -// auto-rolls back certain failures (SQLITE_BUSY, disk full, ...); a following -// explicit ROLLBACK then throws "cannot rollback - no transaction is active", -// which would both mask the true cause and turn a skippable per-file error into -// a whole-build failure. Swallow only the rollback's own error. -function safeRollback(db: { exec: (sql: string) => unknown }) { - try { db.exec('ROLLBACK'); } catch { /* no active transaction */ } -} - function resolveSchemaPath() { const candidates = [ path.join(__dirname, 'schema.sql'), @@ -63,8 +57,7 @@ function installSchema(db, schemaPath = resolveSchemaPath()) { function openIndexDb({ dbPath = DEFAULT_DB_PATH, schemaPath = resolveSchemaPath(), DatabaseImpl = Database }: { dbPath?: string; schemaPath?: string; DatabaseImpl?: new (dbPath: string) => any } = {}) { fs.mkdirSync(path.dirname(dbPath), { recursive: true }); const db = new DatabaseImpl(dbPath); - db.pragma('journal_mode = WAL'); - db.pragma('synchronous = NORMAL'); + configureConnection(db, { busyTimeoutMs: 250 }); installSchema(db, schemaPath); return db; } @@ -138,7 +131,10 @@ function normalizeChangedPath(projectsDir, changedPath) { } function jsonlFileInfoFromPath(projectsDir, changedPath) { - const fp = normalizeChangedPath(projectsDir, changedPath); + let fp = normalizeChangedPath(projectsDir, changedPath); + if (fp?.toLowerCase().endsWith('.meta.json')) { + fp = fp.slice(0, -'.meta.json'.length) + '.jsonl'; + } if (!fp || !fp.endsWith('.jsonl')) return null; if (!fs.existsSync(fp)) return null; const rel = path.relative(projectsDir, fp); @@ -365,14 +361,16 @@ function indexCodexSessionIndex(db, { codexDir = DEFAULT_CODEX_DIR } = {}) { const indexPath = path.join(codexDir, 'session_index.jsonl'); if (!fs.existsSync(indexPath)) return; readLines(indexPath, (line) => { + let item; try { - const item = JSON.parse(line); - if (!item.id || !item.thread_name) return; - db.prepare('UPDATE sessions SET title=COALESCE(title, ?), ended_at=COALESCE(ended_at, ?) WHERE id=? AND source=?') - .run(item.thread_name, item.updated_at || null, codexDbId(item.id), 'codex'); + item = JSON.parse(line); } catch (error) { console.warn(`Warning: malformed Codex session index line: ${(error as Error).message}`); + return; } + if (!item.id || !item.thread_name) return; + db.prepare('UPDATE sessions SET title=COALESCE(title, ?), ended_at=COALESCE(ended_at, ?) WHERE id=? AND source=?') + .run(item.thread_name, item.updated_at || null, codexDbId(item.id), 'codex'); }); } @@ -392,22 +390,25 @@ function refreshSessionProjectPaths(db) { } function indexSubagentMeta(db, fi) { - if (!fi.isSubagent) return; + if (!fi.isSubagent) return false; const mp = fi.path.replace('.jsonl', '.meta.json'); - if (!fs.existsSync(mp)) return; + if (!fs.existsSync(mp)) return false; + let meta; try { - const meta = JSON.parse(fs.readFileSync(mp, 'utf8')); - const tok = db.prepare('SELECT COALESCE(SUM(input_tokens),0)+COALESCE(SUM(output_tokens),0) as t FROM messages WHERE agent_id=?').get(fi.agentId); - const ts = db.prepare('SELECT MIN(timestamp) as t0, MAX(timestamp) as t1 FROM messages WHERE agent_id=?').get(fi.agentId); - const dur = ts?.t0 && ts?.t1 ? new Date(ts.t1).getTime() - new Date(ts.t0).getTime() : null; - if (fi.workflowRunId) { - db.prepare('INSERT OR REPLACE INTO workflow_agents (agent_id,run_id,session_id,agent_type,description) VALUES(?,?,?,?,?)').run(fi.agentId, fi.workflowRunId, fi.sessionId, meta.agentType||null, meta.description||null); - } else { - db.prepare('INSERT OR REPLACE INTO subagents VALUES(?,?,?,?,?,?,?)').run(fi.agentId, fi.sessionId, meta.toolUseId||null, meta.agentType||null, meta.description||null, dur, tok?.t||0); - } + meta = JSON.parse(fs.readFileSync(mp, 'utf8')); } catch (error) { console.warn(`Warning: failed to read subagent meta ${mp}: ${(error as Error).message}`); + return false; } + const tok = db.prepare('SELECT COALESCE(SUM(input_tokens),0)+COALESCE(SUM(output_tokens),0) as t FROM messages WHERE agent_id=?').get(fi.agentId); + const ts = db.prepare('SELECT MIN(timestamp) as t0, MAX(timestamp) as t1 FROM messages WHERE agent_id=?').get(fi.agentId); + const dur = ts?.t0 && ts?.t1 ? new Date(ts.t1).getTime() - new Date(ts.t0).getTime() : null; + if (fi.workflowRunId) { + db.prepare('INSERT OR REPLACE INTO workflow_agents (agent_id,run_id,session_id,agent_type,description) VALUES(?,?,?,?,?)').run(fi.agentId, fi.workflowRunId, fi.sessionId, meta.agentType||null, meta.description||null); + } else { + db.prepare('INSERT OR REPLACE INTO subagents VALUES(?,?,?,?,?,?,?)').run(fi.agentId, fi.sessionId, meta.toolUseId||null, meta.agentType||null, meta.description||null, dur, tok?.t||0); + } + return true; } function indexWorkflows(db, { projectsDir = DEFAULT_PROJECTS_DIR } = {}) { @@ -426,23 +427,25 @@ function indexWorkflows(db, { projectsDir = DEFAULT_PROJECTS_DIR } = {}) { try { wfFiles = fs.readdirSync(wd); } catch { continue; } for (const f of wfFiles) { if (!f.endsWith('.json')) continue; + let wf; try { - const wf = JSON.parse(fs.readFileSync(path.join(wd, f), 'utf8')); - if (!wf.runId) continue; - const ac = db.prepare('SELECT COUNT(*) as c FROM workflow_agents WHERE run_id=?').get(wf.runId); - db.prepare('INSERT OR REPLACE INTO workflows (run_id,session_id,task_id,script,result_json,timestamp,agent_count,duration_ms,total_tokens,status,workflow_name) VALUES(?,?,?,?,?,?,?,?,?,?,?)').run( - wf.runId, sd, wf.taskId||null, wf.script||null, - wf.result ? JSON.stringify(wf.result) : null, wf.timestamp||null, ac?.c||0, - wf.durationMs||null, wf.totalTokens||null, wf.status||null, wf.workflowName||null); - const progress = wf.workflowProgress || []; - for (const item of progress) { - if (item.type !== 'workflow_agent' || !item.agentId) continue; - db.prepare('UPDATE workflow_agents SET phase=?, label=?, model=?, state=?, duration_ms=?, tokens=?, tool_calls=? WHERE agent_id=?').run( - item.phaseTitle||null, item.label||null, item.model||null, item.state||null, - item.durationMs||null, item.tokens||null, item.toolCalls||null, 'agent-' + item.agentId); - } + wf = JSON.parse(fs.readFileSync(path.join(wd, f), 'utf8')); } catch (error) { - console.warn(`Warning: failed to index workflow ${f}: ${(error as Error).message}`); + console.warn(`Warning: failed to read workflow ${f}: ${(error as Error).message}`); + continue; + } + if (!wf.runId) continue; + const ac = db.prepare('SELECT COUNT(*) as c FROM workflow_agents WHERE run_id=?').get(wf.runId); + db.prepare('INSERT OR REPLACE INTO workflows (run_id,session_id,task_id,script,result_json,timestamp,agent_count,duration_ms,total_tokens,status,workflow_name) VALUES(?,?,?,?,?,?,?,?,?,?,?)').run( + wf.runId, sd, wf.taskId||null, wf.script||null, + wf.result ? JSON.stringify(wf.result) : null, wf.timestamp||null, ac?.c||0, + wf.durationMs||null, wf.totalTokens||null, wf.status||null, wf.workflowName||null); + const progress = wf.workflowProgress || []; + for (const item of progress) { + if (item.type !== 'workflow_agent' || !item.agentId) continue; + db.prepare('UPDATE workflow_agents SET phase=?, label=?, model=?, state=?, duration_ms=?, tokens=?, tool_calls=? WHERE agent_id=?').run( + item.phaseTitle||null, item.label||null, item.model||null, item.state||null, + item.durationMs||null, item.tokens||null, item.toolCalls||null, 'agent-' + item.agentId); } } } @@ -452,12 +455,14 @@ function indexWorkflows(db, { projectsDir = DEFAULT_PROJECTS_DIR } = {}) { function indexHistory(db, { historyPath = DEFAULT_HISTORY_PATH } = {}) { if (!fs.existsSync(historyPath)) return; readLines(historyPath, (line) => { + let item; try { - const o = JSON.parse(line); - if (o.sessionId && o.title) db.prepare('UPDATE sessions SET title=? WHERE id=? AND title IS NULL').run(o.title, o.sessionId); + item = JSON.parse(line); } catch (error) { console.warn(`Warning: malformed history line: ${(error as Error).message}`); + return; } + if (item.sessionId && item.title) db.prepare('UPDATE sessions SET title=? WHERE id=? AND title IS NULL').run(item.title, item.sessionId); }); } @@ -466,9 +471,13 @@ function rebuildFts(db) { db.exec("INSERT INTO memories_fts(memories_fts) VALUES('rebuild')"); } -function checkpointDb(db) { +// PASSIVE by default: it checkpoints what it can without blocking concurrent +// readers/writers, so it is safe to run after every build. A blocking TRUNCATE +// (which reclaims the -wal file but needs exclusive access and can contend with +// the daemon + queries) is reserved for maintenance/exit — pass mode explicitly. +function checkpointDb(db, mode = 'PASSIVE') { try { - db.pragma('wal_checkpoint(TRUNCATE)'); + db.pragma(`wal_checkpoint(${mode})`); } catch {} } @@ -497,13 +506,30 @@ function writeIndexMarker(db, key, value = Date.now()) { db.prepare('INSERT OR REPLACE INTO index_state (jsonl_path, mtime, lines_processed) VALUES (?, ?, 0)').run(key, value); } -function writeHeartbeat({ dbPath = DEFAULT_DB_PATH, DatabaseImpl = Database } = {}) { +function writeHeartbeat({ + dbPath = DEFAULT_DB_PATH, + writerLeasePath = writerLockPathFor(dbPath), + DatabaseImpl = Database, + LockDatabaseImpl = DatabaseImpl, +} = {}) { if (!fs.existsSync(dbPath)) return; - const db = new DatabaseImpl(dbPath); + const lease = acquireWriterLease({ + lockPath: writerLeasePath, + openDb: lockPath => new LockDatabaseImpl(lockPath), + }); + if (!lease) return false; try { - writeIndexMarker(db, '__app_heartbeat__'); + const db = new DatabaseImpl(dbPath); + configureConnection(db, { busyTimeoutMs: 0 }); + const txDb = betterSqliteTransactionAdapter(db); + try { + runWriteTransaction(txDb, () => writeIndexMarker(db, '__app_heartbeat__'), { label: 'heartbeat' }); + return true; + } finally { + db.close(); + } } finally { - db.close(); + lease.release(); } } @@ -515,9 +541,19 @@ interface BuildIndexOptions { dbPath?: string; schemaPath?: string; DatabaseImpl?: new (dbPath: string) => any; + LockDatabaseImpl?: new (dbPath: string) => any; force?: boolean; changedPaths?: string[]; preserveDbPath?: string | null; + writerLeasePath?: string; + writerLeaseWaitMs?: number; + writerLeaseMode?: 'acquire' | 'caller-held'; +} + +interface SkippedFile { + path: string; + error: string; + diagnostics?: unknown; } interface BuildIndexResult { @@ -525,6 +561,27 @@ interface BuildIndexResult { latestSourceMtime: number; affectedSessionIds: string[]; ftsRebuilt: boolean; + skipped: number; + skippedFiles: SkippedFile[]; + deferred: boolean; + reason?: string; +} + +function deferredBuildResult( + reason: string, + overrides: Partial> = {}, +): BuildIndexResult { + return { + files: 0, + latestSourceMtime: 0, + affectedSessionIds: [], + ftsRebuilt: false, + skipped: 0, + skippedFiles: [], + ...overrides, + deferred: true, + reason, + }; } function buildIndex({ @@ -535,90 +592,175 @@ function buildIndex({ dbPath = DEFAULT_DB_PATH, schemaPath = resolveSchemaPath(), DatabaseImpl = Database, + LockDatabaseImpl = DatabaseImpl, force = false, changedPaths = undefined, preserveDbPath = null, + writerLeasePath = writerLockPathFor(dbPath), + writerLeaseWaitMs = 2000, + writerLeaseMode = 'acquire', }: BuildIndexOptions = {}): BuildIndexResult { - const db = openIndexDb({ dbPath, schemaPath, DatabaseImpl }); - let messageFtsTriggersDropped = false; - if (preserveDbPath && path.resolve(preserveDbPath) !== path.resolve(dbPath)) { - copyMemoriesFromDb(db, preserveDbPath); + if (writerLeaseMode !== 'acquire' && writerLeaseMode !== 'caller-held') { + throw new Error(`Unknown writer lease mode: ${writerLeaseMode}`); } - const files = [ - ...discoverJsonlFiles({ projectsDir, changedPaths: force ? undefined : changedPaths }), - ...discoverCodexJsonlFiles({ codexDir, changedPaths: force ? undefined : changedPaths }), - ]; - const latestSourceMtime = files.reduce((latest, file) => { - try { - return Math.max(latest, fs.statSync(file.path).mtimeMs); - } catch { - return latest; + let lease: ReturnType = null; + if (writerLeaseMode === 'acquire') { + lease = acquireWriterLease({ + lockPath: writerLeasePath, + openDb: lockPath => new LockDatabaseImpl(lockPath), + waitMs: writerLeaseWaitMs, + }); + if (!lease) { + return deferredBuildResult('writer_busy'); } - }, 0); - + } try { - if (force) { - dropMessageFtsTriggers(db); - messageFtsTriggersDropped = true; - db.prepare("DELETE FROM index_state WHERE substr(jsonl_path, 1, 2) != '__'").run(); - db.prepare("DELETE FROM messages").run(); - db.prepare("DELETE FROM tool_calls").run(); - db.prepare("DELETE FROM tool_results").run(); - db.prepare("DELETE FROM sessions").run(); - db.prepare("DELETE FROM summaries").run(); - db.prepare("DELETE FROM subagents").run(); - db.prepare("DELETE FROM workflows").run(); - db.prepare("DELETE FROM workflow_agents").run(); - } - const affectedSessionIds = new Set(); - if (Array.isArray(changedPaths)) { - for (const changedPath of changedPaths) { - const sessionId = sessionIdFromChangedPath(projectsDir, changedPath); - if (sessionId) affectedSessionIds.add(sessionId); - } - } - for (const file of files) { - db.exec('BEGIN'); - try { - const indexed = file.source === 'codex' ? indexCodexFile(db, file) : indexClaudeFile(db, file); - if (indexed?.sessionId) affectedSessionIds.add(indexed.sessionId); - if (file.source !== 'codex') indexSubagentMeta(db, file); - db.exec('COMMIT'); - } catch (error) { - safeRollback(db); - console.warn(`Warning: failed to index ${file.path}: ${(error as Error).message}`); - } - } - db.exec('BEGIN'); - let ftsRebuilt = false; + const db = openIndexDb({ dbPath, schemaPath, DatabaseImpl }); + const txDb = betterSqliteTransactionAdapter(db); + let messageFtsTriggersDropped = false; try { - indexWorkflows(db, { projectsDir }); - refreshSessionProjectPaths(db); - indexHistory(db, { historyPath }); - indexCodexSessionIndex(db, { codexDir }); - if (messageFtsTriggersDropped) installSchema(db, schemaPath); - ftsRebuilt = ensureFtsReady(db, { force }); - writeIndexMarker(db, '__last_build__'); - writeIndexMarker(db, '__app_heartbeat__'); - writeIndexMarker(db, '__app_last_successful_build__'); - writeIndexMarker(db, '__indexer_owner_app__'); - if (latestSourceMtime) writeIndexMarker(db, '__last_source_mtime__', latestSourceMtime); - db.exec('COMMIT'); - } catch (error) { - safeRollback(db); - throw error; - } - return { files: files.length, latestSourceMtime, affectedSessionIds: [...affectedSessionIds], ftsRebuilt }; - } finally { - if (messageFtsTriggersDropped) { - try { - installSchema(db, schemaPath); - } catch (error) { - console.warn(`Warning: failed to restore message FTS triggers: ${(error as Error).message}`); + if (preserveDbPath && path.resolve(preserveDbPath) !== path.resolve(dbPath)) { + copyMemoriesFromDb(db, preserveDbPath); } + const files = [ + ...discoverJsonlFiles({ projectsDir, changedPaths: force ? undefined : changedPaths }), + ...discoverCodexJsonlFiles({ codexDir, changedPaths: force ? undefined : changedPaths }), + ]; + const latestSourceMtime = files.reduce((latest, file) => { + try { + return Math.max(latest, fs.statSync(file.path).mtimeMs); + } catch { + return latest; + } + }, 0); + + try { + if (force) { + runRetryableWriteTransaction(txDb, () => { + dropMessageFtsTriggers(db); + db.prepare("DELETE FROM index_state WHERE substr(jsonl_path, 1, 2) != '__'").run(); + db.prepare("DELETE FROM messages").run(); + db.prepare("DELETE FROM tool_calls").run(); + db.prepare("DELETE FROM tool_results").run(); + db.prepare("DELETE FROM sessions").run(); + db.prepare("DELETE FROM summaries").run(); + db.prepare("DELETE FROM subagents").run(); + db.prepare("DELETE FROM workflows").run(); + db.prepare("DELETE FROM workflow_agents").run(); + }, { label: 'force-cleanup' }); + messageFtsTriggersDropped = true; + } + } catch (error) { + if (isBeginBusyFailure(error)) { + return deferredBuildResult('database_busy', { + files: files.length, + latestSourceMtime, + }); + } + throw error; + } + const affectedSessionIds = new Set(); + const finalizeAffectedSessionIds = new Set(); + const changedMetaJsonlPaths = new Set(); + if (Array.isArray(changedPaths)) { + for (const changedPath of changedPaths) { + const sessionId = sessionIdFromChangedPath(projectsDir, changedPath); + const normalizedChangedPath = normalizeChangedPath(projectsDir, changedPath); + const isMetaChange = normalizedChangedPath?.toLowerCase().endsWith('.meta.json'); + if (isMetaChange && normalizedChangedPath) { + changedMetaJsonlPaths.add( + normalizedChangedPath.slice(0, -'.meta.json'.length) + '.jsonl', + ); + } + // Transcript files report their session only after their own transaction + // commits. Workflow changes are applied during finalize, so stage those + // IDs until the finalize transaction commits. Meta files map back to their + // transcript transaction and are reported only after that commit. + if (sessionId && !changedPath.toLowerCase().endsWith('.jsonl') && !isMetaChange) { + finalizeAffectedSessionIds.add(sessionId); + } + } + } + const skipped: SkippedFile[] = []; + for (const file of files) { + try { + // The write is committed before affectedSessionIds is updated, so a + // failed/rolled-back file never reports a phantom updated session. + const indexed = runRetryableWriteTransaction(txDb, () => { + const result = file.source === 'codex' ? indexCodexFile(db, file) : indexClaudeFile(db, file); + const metaIndexed = file.source !== 'codex' && indexSubagentMeta(db, file); + if (!result?.sessionId && metaIndexed && changedMetaJsonlPaths.has(file.path)) { + return { sessionId: file.sessionId, path: file.path }; + } + return result; + }, { label: `file:${file.path}` }); + if (indexed?.sessionId) affectedSessionIds.add(indexed.sessionId); + } catch (error) { + if (isBeginBusyFailure(error)) { + return deferredBuildResult('database_busy', { + files: files.length, + latestSourceMtime, + affectedSessionIds: [...affectedSessionIds], + skipped: skipped.length, + skippedFiles: skipped, + }); + } + if (hasUnusableTransaction(error)) throw error; + skipped.push({ path: file.path, error: (error as Error).message, diagnostics: (error as { obelisk?: unknown }).obelisk }); + console.warn(`Warning: failed to index ${file.path}: ${(error as Error).message}`); + } + } + let ftsRebuilt = false; + // Finalize is one transaction; a failure here fails the whole build (the + // index would otherwise be left inconsistent). + try { + runRetryableWriteTransaction(txDb, () => { + indexWorkflows(db, { projectsDir }); + refreshSessionProjectPaths(db); + indexHistory(db, { historyPath }); + indexCodexSessionIndex(db, { codexDir }); + if (messageFtsTriggersDropped) installSchema(db, schemaPath); + ftsRebuilt = ensureFtsReady(db, { force }); + writeIndexMarker(db, '__last_build__'); + writeIndexMarker(db, '__app_last_successful_build__'); + writeIndexMarker(db, '__indexer_owner_app__'); + if (latestSourceMtime) writeIndexMarker(db, '__last_source_mtime__', latestSourceMtime); + }, { label: 'finalize' }); + } catch (error) { + if (isBeginBusyFailure(error)) { + return deferredBuildResult('database_busy', { + files: files.length, + latestSourceMtime, + affectedSessionIds: [...affectedSessionIds], + skipped: skipped.length, + skippedFiles: skipped, + }); + } + throw error; + } + for (const sessionId of finalizeAffectedSessionIds) affectedSessionIds.add(sessionId); + return { + files: files.length, + latestSourceMtime, + affectedSessionIds: [...affectedSessionIds], + ftsRebuilt, + skipped: skipped.length, + skippedFiles: skipped, + deferred: false, + }; + } finally { + if (messageFtsTriggersDropped) { + try { + installSchema(db, schemaPath); + } catch (error) { + console.warn(`Warning: failed to restore message FTS triggers: ${(error as Error).message}`); + } + } + checkpointDb(db); + db.close(); } - checkpointDb(db); - db.close(); + } finally { + lease?.release(); } } diff --git a/app/tests/electron-concurrency-child.mjs b/app/tests/electron-concurrency-child.mjs new file mode 100644 index 0000000..75cc68a --- /dev/null +++ b/app/tests/electron-concurrency-child.mjs @@ -0,0 +1,27 @@ +import { createRequire } from 'node:module'; +import { createInterface } from 'node:readline'; + +const require = createRequire(import.meta.url); +const Database = require('better-sqlite3'); + +const [mode, payloadJson] = process.argv.slice(2); +const payload = JSON.parse(payloadJson || '{}'); + +if (mode === 'holder') { + const db = new Database(payload.lockPath); + db.pragma('busy_timeout = 0'); + db.exec('BEGIN IMMEDIATE'); + process.stdout.write('READY\n'); + const input = createInterface({ input: process.stdin }); + await new Promise(resolve => input.once('line', resolve)); + db.exec('ROLLBACK'); + db.close(); + input.close(); +} else if (mode === 'build') { + const { buildIndex } = await import('../out/main/indexer.js'); + process.stdout.write('STARTING\n'); + const result = buildIndex(payload.options); + process.stdout.write(`RESULT ${JSON.stringify(result)}\n`); +} else { + throw new Error(`Unknown concurrency child mode: ${mode}`); +} diff --git a/app/tests/electron-concurrency.mjs b/app/tests/electron-concurrency.mjs new file mode 100644 index 0000000..42bc264 --- /dev/null +++ b/app/tests/electron-concurrency.mjs @@ -0,0 +1,158 @@ +// Real Electron/better-sqlite3 concurrency test (docs/adr/0006 Phase 2). +// Run: cd app && npx electron tests/electron-concurrency.mjs +// +// Exercises actual dual-connection contention against a WAL database using the +// Electron-ABI better-sqlite3 that the app uses in production. +import { app } from 'electron'; +import { spawn } from 'node:child_process'; +import { mkdirSync, writeFileSync, rmSync, mkdtempSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { dirname, join } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { createInterface } from 'node:readline'; +import { once } from 'node:events'; +import { setTimeout as delay } from 'node:timers/promises'; +import Database from 'better-sqlite3'; +import { buildIndex } from '../out/main/indexer.js'; + +let failures = 0; +const childScript = join(dirname(fileURLToPath(import.meta.url)), 'electron-concurrency-child.mjs'); +function assert(condition, msg) { + if (!condition) { console.error('FAIL:', msg); failures++; } + else console.log('PASS:', msg); +} + +function spawnChild(mode, payload) { + return spawn(process.execPath, [childScript, mode, JSON.stringify(payload)], { + env: { ...process.env, ELECTRON_RUN_AS_NODE: '1' }, + stdio: ['pipe', 'pipe', 'inherit'], + }); +} + +function lineReader(child) { + const lines = []; + const waiters = []; + createInterface({ input: child.stdout }).on('line', line => { + lines.push(line); + for (const waiter of [...waiters]) { + if (!line.startsWith(waiter.prefix)) continue; + waiters.splice(waiters.indexOf(waiter), 1); + waiter.resolve(line); + } + }); + return { + waitFor(prefix) { + const existing = lines.find(line => line.startsWith(prefix)); + if (existing) return Promise.resolve(existing); + return new Promise(resolve => waiters.push({ prefix, resolve })); + }, + }; +} + +async function waitForSuccess(child) { + let code = child.exitCode; + if (code === null) [code] = await once(child, 'exit'); + assert(code === 0, `child exited successfully, code=${code}`); +} + +async function run() { + const home = mkdtempSync(join(tmpdir(), 'obelisk-electron-concurrency-')); + const dbPath = join(home, '.obelisk', 'obelisk.sqlite'); + const projectsDir = join(home, '.claude', 'projects'); + const projDir = join(projectsDir, '-proj'); + mkdirSync(join(home, '.obelisk'), { recursive: true }); + mkdirSync(projDir, { recursive: true }); + + function msg(uuid) { + return JSON.stringify({ + uuid, type: 'user', timestamp: '2026-06-10T10:00:00Z', cwd: '/tmp', + message: { role: 'user', content: `concurrent ${uuid}` }, + }) + '\n'; + } + for (let i = 0; i < 20; i++) { + writeFileSync(join(projDir, `s${i}.jsonl`), msg(`m${i}`)); + } + + console.log('--- Test 1: buildIndex with real better-sqlite3 ---'); + const result = buildIndex({ + force: true, + claudeDir: join(home, '.claude'), + codexDir: join(home, '.codex'), + projectsDir, + dbPath, + DatabaseImpl: Database, + }); + assert(result.files === 20, `indexed 20 files, got ${result.files}`); + assert(result.skipped === 0, `no files skipped, got ${result.skipped}`); + + console.log('--- Test 2: concurrent reader during write ---'); + const reader = new Database(dbPath, { readonly: true }); + reader.pragma('journal_mode = WAL'); + const readStmt = reader.prepare('SELECT COUNT(*) AS c FROM sessions'); + const beforeCount = readStmt.get().c; + assert(beforeCount === 20, `reader sees 20 sessions, got ${beforeCount}`); + // Incremental build concurrent with open reader + const result2 = buildIndex({ + force: false, + claudeDir: join(home, '.claude'), + codexDir: join(home, '.codex'), + projectsDir, + dbPath, + DatabaseImpl: Database, + }); + assert(result2.skipped === 0, `concurrent build no skips, got ${result2.skipped}`); + const duringCount = readStmt.get().c; + assert(duringCount === 20, `reader snapshot stable, got ${duringCount}`); + reader.close(); + + console.log('--- Test 3: a real concurrent writer releases within the lease budget ---'); + const lockPath = join(dirname(dbPath), 'writer.lock.sqlite'); + const buildOptions = { + force: false, + claudeDir: join(home, '.claude'), + codexDir: join(home, '.codex'), + projectsDir, + dbPath, + writerLeaseWaitMs: 1500, + }; + const holder = spawnChild('holder', { lockPath }); + const holderLines = lineReader(holder); + await holderLines.waitFor('READY'); + const contendedBuild = spawnChild('build', { options: buildOptions }); + const buildLines = lineReader(contendedBuild); + await buildLines.waitFor('STARTING'); + const startedAt = Date.now(); + await delay(200); + holder.stdin.write('release\n'); + const resultLine = await buildLines.waitFor('RESULT '); + const result3 = JSON.parse(resultLine.slice('RESULT '.length)); + const waitedMs = Date.now() - startedAt; + assert(result3.deferred === false, `contended build completed, reason=${result3.reason}`); + assert(result3.skipped === 0, `contended build skipped no files, got ${result3.skipped}`); + assert(waitedMs >= 150, `build overlapped the held lease for ${waitedMs}ms`); + await Promise.all([waitForSuccess(holder), waitForSuccess(contendedBuild)]); + + console.log('--- Test 4: persistent writer contention is bounded ---'); + const persistentHolder = spawnChild('holder', { lockPath }); + const persistentHolderLines = lineReader(persistentHolder); + await persistentHolderLines.waitFor('READY'); + const boundedBuild = spawnChild('build', { options: { ...buildOptions, writerLeaseWaitMs: 200 } }); + const boundedLines = lineReader(boundedBuild); + await boundedLines.waitFor('STARTING'); + const boundedStartedAt = Date.now(); + const boundedResultLine = await boundedLines.waitFor('RESULT '); + const boundedResult = JSON.parse(boundedResultLine.slice('RESULT '.length)); + const boundedMs = Date.now() - boundedStartedAt; + assert(boundedResult.deferred === true, 'persistent contention returns deferred'); + assert(boundedResult.reason === 'writer_busy', `persistent contention reason=${boundedResult.reason}`); + assert(boundedMs < 1000, `persistent contention returned within budget (${boundedMs}ms)`); + persistentHolder.stdin.write('release\n'); + await Promise.all([waitForSuccess(persistentHolder), waitForSuccess(boundedBuild)]); + + rmSync(home, { recursive: true, force: true }); + console.log('---'); + console.log(failures ? `${failures} TEST(S) FAILED` : 'ALL TESTS PASSED'); + process.exitCode = failures ? 1 : 0; +} + +app.whenReady().then(run).finally(() => app.quit()); diff --git a/docs/adr/0006-write-transaction-rollback-and-concurrency.md b/docs/adr/0006-write-transaction-rollback-and-concurrency.md index 41bff9a..5fe27fb 100644 --- a/docs/adr/0006-write-transaction-rollback-and-concurrency.md +++ b/docs/adr/0006-write-transaction-rollback-and-concurrency.md @@ -1,76 +1,75 @@ # Write-transaction rollback safety and SQLite concurrency **Context.** The app surfaced `Obelisk index build failed: cannot rollback - no -transaction is active`. That message is a *secondary* error: SQLite auto-rolls -back certain failures (notably `SQLITE_BUSY` / `SQLITE_BUSY_SNAPSHOT`, also disk -full), after which the per-file build loop's unguarded `db.exec('ROLLBACK')` in -its `catch` threw over the real error and aborted the whole build instead of -skipping just the offending file. The underlying trigger is concurrency: the app -runs a daemon indexer, manual rebuilds, and read queries against one WAL -database, and the skill's passive-pull build can write the same database from a -separate process. +transaction is active`. That text was a secondary cleanup failure. SQLite had +already ended the transaction, then the catch block's unguarded `ROLLBACK` +threw over the primary exception and turned a skippable per-file failure into a +whole-build failure. The masked exception was not preserved, so contention +(`SQLITE_BUSY` / `SQLITE_BUSY_SNAPSHOT`) is the leading explanation rather than +a proven historical fact. It is plausible because daemon builds, manual +rebuilds, skill passive-pull indexing, heartbeat writes, and reads share one WAL +database. -`busy_timeout` is **not** the root-cause fix and must not be treated as one. -better-sqlite3's constructor already defaults `timeout` to 5000ms, so the app hit -`SQLITE_BUSY` *despite* a 5s wait — which points at `SQLITE_BUSY_SNAPSHOT` from -deferred (read-then-write) transactions, a snapshot conflict that `busy_timeout` -does not wait on. Only `BEGIN IMMEDIATE` plus whole-transaction retry addresses -that. +`busy_timeout` alone is not a correctness fix. In particular, +`SQLITE_BUSY_SNAPSHOT` is not made safe by waiting longer, and retrying only the +failed statement can replay part of a transaction. -**Decision.** Split the work into a cheap stopgap now and a correctness/ -concurrency fix later. +**Decision.** Use one transaction primitive plus two explicit coordination +layers. -*Stopgap (done).* Both indexers use a guarded rollback (`safeRollback`): a -cleanup rollback swallows only its own error and never masks the primary one. -Per-file failures are logged and the build continues; the finalize failure still -propagates. The skill's connection (`node:sqlite`, which has **no** default busy -timeout) gets an explicit `PRAGMA busy_timeout = 5000`; the app adds no such -pragma because better-sqlite3 already defaults to 5000ms — adding it there was -redundant and removed, precisely so nobody reads it as "the fix". +- `scripts/tx.ts` owns the binding-agnostic `runWriteTransaction(db, work)`. + Adapters expose transaction state from better-sqlite3's `inTransaction` and + node:sqlite's `isTransaction`. The primitive performs `BEGIN IMMEDIATE`, runs + `work` exactly once, commits, and attempts rollback only when the binding says + a transaction is active or its state is unknown. Cleanup never masks the + primary exception. Diagnostics record phase, SQLite code, rollback outcome, + transaction state, label, and attempts. +- Retry is an upper-layer policy in `scripts/write-coordinator.ts`, never hidden + inside the transaction primitive. Only an idempotent whole transaction that + failed during work/commit with `SQLITE_BUSY*` and is confirmed inactive may be + retried. The default is three attempts within a one-second budget with short + backoff. BEGIN contention is deferred to the build scheduler; an active or + unknown post-error transaction aborts the build. +- Per-file failures remain warnings and are reported in `skippedFiles`; finalize + failures propagate. `affectedSessionIds` is updated only after the relevant + commit. Force cleanup is one atomic, retryable transaction, and finalize is + likewise retried as a complete idempotent transaction. +- A fresh `__app_heartbeat__` is policy ownership: while it is fresh, the skill + opens no write connection and performs no migration, schema setup, checkpoint, + index build, or `attune`. `__app_last_successful_build__` remains an + observability/freshness marker and is not required for ownership. The skill + checks ownership again after acquiring the hard lease to close the TOCTOU + window. Search/query connections are read-only. +- A dedicated `.obelisk/writer.lock.sqlite` provides the cross-process safety + mutex on every platform. Acquisition is `BEGIN IMMEDIATE` with non-blocking or + bounded waiting; release is idempotent. App builds and heartbeats, skill builds + and attune, app schema/legacy migrations and memory mutations, and manual + rebuild all participate. Manual rebuild's main process owns the lease across + worker build, atomic target replacement, and database reopen; the worker uses + the explicit `caller-held` mode. +- The app's in-process indexer service permits one build at a time. A lease + deferral retains changed paths and schedules a short retry without announcing + a successful build. Service start publishes the ownership heartbeat + immediately, then refreshes it periodically. +- Index-writer and skill read connections use an explicit 250 ms SQLite busy + timeout inside the larger bounded coordination budget. The long-lived app + query connection retains a 5 s timeout; heartbeat is deliberately non-blocking + (`0 ms`) so it never stalls the Electron main thread. Builds use + `BEGIN IMMEDIATE`. Routine checkpointing is `PASSIVE`; blocking `TRUNCATE` is + reserved for explicit maintenance. -*Planned full fix (deferred, two phases).* +**Verification.** Fast tests inject auto-rollback and BUSY failures to prove the +primary error is preserved, retry replays the whole transaction, persistent +per-file failure is skipped, force cleanup is atomic, and affected-session state +is commit-aware. The Electron harness uses real Electron-ABI better-sqlite3 and +two child processes: one holds the SQLite writer lease until signalled, while +the other runs synchronous `buildIndex`. It verifies both release-within-budget +success and bounded `writer_busy` deferral. Separate arbitration tests prove a +heartbeat-only daemon marker keeps query and attune paths read-only. -Phase 1 — transaction semantics: -- A shared, binding-agnostic `runWriteTransaction(db, work)` (same injection - model as `persist`): BEGIN → work → COMMIT, guarded rollback on failure, - original exception always preserved. Both app and skill call it, so their - behaviour is identical. -- Per-file callers catch and continue; finalize does **not** swallow — a finalize - failure fails the build (the skill currently only warns; that changes). -- In-memory state such as `affectedSessionIds` is updated **only after** a - successful COMMIT (today the app adds the id before COMMIT, so a failed commit - can report a wrong affected set). -- Structured diagnostics: `phase` (begin/file-write/commit/finalize/checkpoint), - SQLite `code`, file path, whether rollback succeeded, whether a txn is still - active; surface the skipped-file count in the build result rather than only - `console.warn` (no silent coverage gaps). - -Phase 2 — concurrency: -- A stable concurrency test against real Electron `better-sqlite3` (daemon + - rebuild + skill writer), not an injected fake BUSY. Note this cannot run under - standalone `node --test` (better-sqlite3 is Electron-ABI); it needs an - Electron-hosted harness. The fast injected-BUSY unit test is kept as a - lower-level guard for `runWriteTransaction`, alongside it. -- Serialize all index writes through a single writer: an in-process - `BuildCoordinator`, plus the existing cross-process daemon arbitration - (`__app_heartbeat__` markers) so the skill defers to a live daemon. Single - writer = both layers together. -- `BEGIN IMMEDIATE` to take the write lock up front and avoid - `SQLITE_BUSY_SNAPSHOT` on read-then-write. -- Bounded, short-backoff retry of the **whole** transaction (not the single - failed statement) on `SQLITE_BUSY` / `SQLITE_BUSY_SNAPSHOT`. This relies on the - per-file work being idempotent (upsert/replace + delete-session cascade), which - it is — an invariant the retry depends on. -- Stop forcing `wal_checkpoint(TRUNCATE)` after every build; prefer a `PASSIVE` - checkpoint on idle, with `TRUNCATE` reserved for maintenance/exit. -- Centralized connection configuration (explicit 5000ms for `node:sqlite` — a - real behaviour change; explicit for better-sqlite3 too, acknowledging its - default is already 5000ms). - -**Consequences.** With only the stopgap in place, contention no longer crashes a -build, but a conflicting file is *skipped* (non-fatal) and its data is briefly -missing until the next build — acceptable as a stopgap because the guard keeps -the index consistent and self-healing. The true fix is tracked as the two-phase -plan above. A future contributor should not "fix" the concurrency by bumping or -re-adding `busy_timeout`; the direction is a shared transaction module, single -writer, `BEGIN IMMEDIATE`, and whole-transaction retry. +**Consequences.** Heartbeat and lease have deliberately different jobs: the +heartbeat decides who should write, while the lease guarantees writers cannot +overlap when policy information races or is stale. A single bad transcript can +still be skipped so the index self-heals on a later build; structural/finalize +failures remain visible. Longer timeouts must not replace the transaction and +ownership rules recorded here. diff --git a/scripts/core.ts b/scripts/core.ts index 841f6de..ee85cd2 100644 --- a/scripts/core.ts +++ b/scripts/core.ts @@ -11,9 +11,10 @@ import { createContext, runInNewContext } from 'node:vm'; -import { DB_PATH, openDb } from './db.mjs'; -import { buildIndex } from './indexer.mjs'; +import { DB_PATH, openDb, openReadDb, openWriterLeaseDb } from './db.mjs'; +import { buildIndex, shouldSkipBuild } from './indexer.mjs'; import { createQueryApi, createAttuneApi } from './query.mjs'; +import { acquireWriterLease, writerLockPathFor } from './writer-lease.ts'; export { buildIndex, DB_PATH }; @@ -33,7 +34,7 @@ function runInSandbox(api: SandboxApi, scriptContent: string): Promise // FTS search over indexed message text. Refreshes the index, then queries. export function searchText(text: string, opts?: Record): unknown { buildIndex(); - const db = openDb(); + const db = openReadDb(); try { return createQueryApi(db).search(text, opts); } finally { @@ -44,7 +45,7 @@ export function searchText(text: string, opts?: Record): unknow // Execute a read-only CodeAct query script and resolve its returned value. export async function executeQuery(scriptContent: string): Promise { buildIndex(); - const db = openDb(); + const db = openReadDb(); try { return await runInSandbox(createQueryApi(db), scriptContent); } finally { @@ -54,11 +55,37 @@ export async function executeQuery(scriptContent: string): Promise { // Execute a memory-mutation CodeAct script (remember/forget only). export async function executeAttune(scriptContent: string): Promise { - buildIndex(); - const db = openDb(); + const build = buildIndex() as { reason?: string } | undefined; + if (build?.reason === 'daemon_active') { + throw new Error('Obelisk daemon owns index writes; attune is read-only until the daemon stops'); + } + if (build?.reason === 'writer_busy' || build?.reason === 'database_busy') { + throw new Error('Obelisk index writer is busy; attune was not applied'); + } + const lease = acquireWriterLease({ + lockPath: writerLockPathFor(DB_PATH), + openDb: openWriterLeaseDb, + waitMs: 1000, + }); + if (!lease) throw new Error('Obelisk index writer is busy; attune was not applied'); try { - return await runInSandbox(createAttuneApi(db), scriptContent); + // Close the heartbeat TOCTOU window after acquiring the hard lease. + const ownershipDb = openReadDb(); + try { + const ownership = shouldSkipBuild(ownershipDb, { ignoreRecentBuild: true }); + if (ownership.reason === 'daemon_active') { + throw new Error('Obelisk daemon owns index writes; attune is read-only until the daemon stops'); + } + } finally { + ownershipDb.close(); + } + const db = openDb(); + try { + return await runInSandbox(createAttuneApi(db), scriptContent); + } finally { + db.close(); + } } finally { - db.close(); + lease.release(); } } diff --git a/scripts/db.mjs b/scripts/db.mjs index 4d175cc..4b66404 100644 --- a/scripts/db.mjs +++ b/scripts/db.mjs @@ -1,5 +1,6 @@ import { createRequire } from 'node:module'; import { CLAUDE_DIR, CODEX_DIR, TEXT_LIMIT, trunc, truncJson, extractText, extractContentType, extractMessageIsMeta, filePath, isDir, readLines } from './parsing.mjs'; +import { configureConnection } from './tx.ts'; const require = createRequire(import.meta.url); const fs = require('node:fs'); const path = require('node:path'); @@ -22,17 +23,25 @@ function openDb() { migrateLegacyDbIfNeeded(); fs.mkdirSync(path.dirname(DB_PATH), { recursive: true }); const db = new DatabaseSync(DB_PATH); - db.exec('PRAGMA journal_mode=WAL'); - db.exec('PRAGMA synchronous=NORMAL'); - // Wait for a contended lock instead of erroring with SQLITE_BUSY: a running - // app daemon may be writing the same database while the skill reads/indexes. - db.exec('PRAGMA busy_timeout=5000'); + configureConnection(db, { busyTimeoutMs: 250 }); migrateExistingColumns(db); db.exec(SCHEMA); migrateDb(db); return db; } +// Queries and daemon-arbitration checks must never migrate/configure the index. +// The caller is responsible for ensuring the database exists first. +function openReadDb() { + const db = new DatabaseSync(DB_PATH, { readOnly: true }); + db.exec('PRAGMA busy_timeout=250'); + return db; +} + +function openWriterLeaseDb(lockPath) { + return new DatabaseSync(lockPath); +} + function ensureColumn(db, table, column, definition) { const columns = db.prepare(`PRAGMA table_info(${table})`).all().map(c => c.name); if (!columns.includes(column)) db.exec(`ALTER TABLE ${table} ADD COLUMN ${column} ${definition}`); @@ -65,4 +74,4 @@ function rebuildMemoryFts(db) { } -export { CLAUDE_DIR, CODEX_DIR, OBELISK_DIR, DB_PATH, TEXT_LIMIT, openDb, rebuildMemoryFts, trunc, truncJson, extractText, extractContentType, extractMessageIsMeta, filePath, isDir, readLines, fs, path, os }; +export { CLAUDE_DIR, CODEX_DIR, OBELISK_DIR, DB_PATH, TEXT_LIMIT, openDb, openReadDb, openWriterLeaseDb, rebuildMemoryFts, trunc, truncJson, extractText, extractContentType, extractMessageIsMeta, filePath, isDir, readLines, fs, path, os }; diff --git a/scripts/indexer.mjs b/scripts/indexer.mjs index b8f8d89..d78f9b7 100644 --- a/scripts/indexer.mjs +++ b/scripts/indexer.mjs @@ -1,23 +1,17 @@ -import { openDb, rebuildMemoryFts } from './db.mjs'; +import { DB_PATH, openDb, openReadDb, openWriterLeaseDb, rebuildMemoryFts } from './db.mjs'; import { CLAUDE_DIR, CODEX_DIR, PROJECTS_DIR, fs, path, isDir, readLines, inferProjectPath, discoverJsonlFiles, discoverCodexJsonlFiles, codexDbId, readCodexGuardianThreadInfo, } from './parsing.mjs'; import { persist } from './persist.ts'; +import { nodeSqliteTransactionAdapter } from './tx.ts'; +import { acquireWriterLease, writerLockPathFor } from './writer-lease.ts'; +import { runRetryableWriteTransaction, isBeginBusyFailure, hasUnusableTransaction } from './write-coordinator.ts'; import { parse as claudeParse } from './providers/claude.ts'; import { parse as codexParse } from './providers/codex.ts'; const HISTORY_PATH = path.join(CLAUDE_DIR, 'history.jsonl'); -// A rollback inside a catch must never throw over the real error. SQLite -// auto-rolls back certain failures (SQLITE_BUSY, disk full, ...); a following -// explicit ROLLBACK then throws "cannot rollback - no transaction is active", -// which would both mask the true cause and turn a skippable per-file error into -// a whole-build failure. Swallow only the rollback's own error. -function safeRollback(db) { - try { db.exec('ROLLBACK'); } catch { /* no active transaction */ } -} - function needsReindex(db, fp) { const mt = fs.statSync(fp).mtimeMs; @@ -31,14 +25,16 @@ function indexCodexSessionIndex(db) { const indexPath = path.join(CODEX_DIR, 'session_index.jsonl'); if (!fs.existsSync(indexPath)) return; readLines(indexPath, (line) => { + let item; try { - const item = JSON.parse(line); - if (!item.id || !item.thread_name) return; - db.prepare('UPDATE sessions SET title=COALESCE(title, ?), ended_at=COALESCE(ended_at, ?) WHERE id=? AND source=?') - .run(item.thread_name, item.updated_at || null, codexDbId(item.id), 'codex'); + item = JSON.parse(line); } catch (e) { process.stderr.write(`Warning: malformed Codex session index line: ${e.message}\n`); + return; } + if (!item.id || !item.thread_name) return; + db.prepare('UPDATE sessions SET title=COALESCE(title, ?), ended_at=COALESCE(ended_at, ?) WHERE id=? AND source=?') + .run(item.thread_name, item.updated_at || null, codexDbId(item.id), 'codex'); }); } @@ -62,17 +58,21 @@ function indexSubagentMeta(db, fi) { if (!fi.isSubagent) return; const mp = fi.path.replace('.jsonl', '.meta.json'); if (!fs.existsSync(mp)) return; + let meta; try { - const meta = JSON.parse(fs.readFileSync(mp, 'utf8')); - const tok = db.prepare('SELECT COALESCE(SUM(input_tokens),0)+COALESCE(SUM(output_tokens),0) as t FROM messages WHERE agent_id=?').get(fi.agentId); - const ts = db.prepare('SELECT MIN(timestamp) as t0, MAX(timestamp) as t1 FROM messages WHERE agent_id=?').get(fi.agentId); - const dur = ts?.t0 && ts?.t1 ? new Date(ts.t1).getTime() - new Date(ts.t0).getTime() : null; - if (fi.workflowRunId) { - db.prepare('INSERT OR REPLACE INTO workflow_agents (agent_id,run_id,session_id,agent_type,description) VALUES(?,?,?,?,?)').run(fi.agentId, fi.workflowRunId, fi.sessionId, meta.agentType||null, meta.description||null); - } else { - db.prepare('INSERT OR REPLACE INTO subagents VALUES(?,?,?,?,?,?,?)').run(fi.agentId, fi.sessionId, meta.toolUseId||null, meta.agentType||null, meta.description||null, dur, tok?.t||0); - } - } catch (e) { process.stderr.write(`Warning: failed to read subagent meta ${mp}: ${e.message}\n`); } + meta = JSON.parse(fs.readFileSync(mp, 'utf8')); + } catch (e) { + process.stderr.write(`Warning: failed to read subagent meta ${mp}: ${e.message}\n`); + return; + } + const tok = db.prepare('SELECT COALESCE(SUM(input_tokens),0)+COALESCE(SUM(output_tokens),0) as t FROM messages WHERE agent_id=?').get(fi.agentId); + const ts = db.prepare('SELECT MIN(timestamp) as t0, MAX(timestamp) as t1 FROM messages WHERE agent_id=?').get(fi.agentId); + const dur = ts?.t0 && ts?.t1 ? new Date(ts.t1).getTime() - new Date(ts.t0).getTime() : null; + if (fi.workflowRunId) { + db.prepare('INSERT OR REPLACE INTO workflow_agents (agent_id,run_id,session_id,agent_type,description) VALUES(?,?,?,?,?)').run(fi.agentId, fi.workflowRunId, fi.sessionId, meta.agentType||null, meta.description||null); + } else { + db.prepare('INSERT OR REPLACE INTO subagents VALUES(?,?,?,?,?,?,?)').run(fi.agentId, fi.sessionId, meta.toolUseId||null, meta.agentType||null, meta.description||null, dur, tok?.t||0); + } } function indexWorkflows(db) { @@ -91,22 +91,26 @@ function indexWorkflows(db) { try { wfFiles = fs.readdirSync(wd); } catch { continue; } for (const f of wfFiles) { if (!f.endsWith('.json')) continue; + let wf; try { - const wf = JSON.parse(fs.readFileSync(path.join(wd, f), 'utf8')); - if (!wf.runId) continue; - const ac = db.prepare('SELECT COUNT(*) as c FROM workflow_agents WHERE run_id=?').get(wf.runId); - db.prepare('INSERT OR REPLACE INTO workflows (run_id,session_id,task_id,script,result_json,timestamp,agent_count,duration_ms,total_tokens,status,workflow_name) VALUES(?,?,?,?,?,?,?,?,?,?,?)').run( - wf.runId, sd, wf.taskId||null, wf.script||null, - wf.result ? JSON.stringify(wf.result) : null, wf.timestamp||null, ac?.c||0, - wf.durationMs||null, wf.totalTokens||null, wf.status||null, wf.workflowName||null); - const progress = wf.workflowProgress || []; - for (const item of progress) { - if (item.type !== 'workflow_agent' || !item.agentId) continue; - db.prepare('UPDATE workflow_agents SET phase=?, label=?, model=?, state=?, duration_ms=?, tokens=?, tool_calls=? WHERE agent_id=?').run( - item.phaseTitle||null, item.label||null, item.model||null, item.state||null, - item.durationMs||null, item.tokens||null, item.toolCalls||null, 'agent-' + item.agentId); - } - } catch (e) { process.stderr.write(`Warning: failed to index workflow ${f}: ${e.message}\n`); } + wf = JSON.parse(fs.readFileSync(path.join(wd, f), 'utf8')); + } catch (e) { + process.stderr.write(`Warning: failed to read workflow ${f}: ${e.message}\n`); + continue; + } + if (!wf.runId) continue; + const ac = db.prepare('SELECT COUNT(*) as c FROM workflow_agents WHERE run_id=?').get(wf.runId); + db.prepare('INSERT OR REPLACE INTO workflows (run_id,session_id,task_id,script,result_json,timestamp,agent_count,duration_ms,total_tokens,status,workflow_name) VALUES(?,?,?,?,?,?,?,?,?,?,?)').run( + wf.runId, sd, wf.taskId||null, wf.script||null, + wf.result ? JSON.stringify(wf.result) : null, wf.timestamp||null, ac?.c||0, + wf.durationMs||null, wf.totalTokens||null, wf.status||null, wf.workflowName||null); + const progress = wf.workflowProgress || []; + for (const item of progress) { + if (item.type !== 'workflow_agent' || !item.agentId) continue; + db.prepare('UPDATE workflow_agents SET phase=?, label=?, model=?, state=?, duration_ms=?, tokens=?, tool_calls=? WHERE agent_id=?').run( + item.phaseTitle||null, item.label||null, item.model||null, item.state||null, + item.durationMs||null, item.tokens||null, item.toolCalls||null, 'agent-' + item.agentId); + } } } } @@ -115,32 +119,54 @@ function indexWorkflows(db) { function indexHistory(db) { if (!fs.existsSync(HISTORY_PATH)) return; readLines(HISTORY_PATH, (line) => { + let item; try { - const o = JSON.parse(line); - if (o.sessionId && o.title) db.prepare('UPDATE sessions SET title=? WHERE id=? AND title IS NULL').run(o.title, o.sessionId); - } catch (e) { process.stderr.write(`Warning: malformed history line: ${e.message}\n`); } + item = JSON.parse(line); + } catch (e) { + process.stderr.write(`Warning: malformed history line: ${e.message}\n`); + return; + } + if (item.sessionId && item.title) db.prepare('UPDATE sessions SET title=? WHERE id=? AND title IS NULL').run(item.title, item.sessionId); }); } const BUILD_DEBOUNCE_MS = 30000; const APP_HEARTBEAT_FRESH_MS = 60000; -function shouldSkipBuild(db, { now = Date.now() } = {}) { +function shouldSkipBuild(db, { now = Date.now(), ignoreRecentBuild = false } = {}) { const appHeartbeat = db.prepare("SELECT mtime FROM index_state WHERE jsonl_path='__app_heartbeat__'").get(); - const appSuccessfulBuild = db.prepare("SELECT mtime FROM index_state WHERE jsonl_path='__app_last_successful_build__'").get(); - if ( - appHeartbeat && now - appHeartbeat.mtime < APP_HEARTBEAT_FRESH_MS && - appSuccessfulBuild && now - appSuccessfulBuild.mtime < APP_HEARTBEAT_FRESH_MS - ) { - return { skip: true, reason: 'app_successful_build' }; + if (appHeartbeat && now - appHeartbeat.mtime < APP_HEARTBEAT_FRESH_MS) { + return { skip: true, reason: 'daemon_active' }; } - const last = db.prepare("SELECT mtime FROM index_state WHERE jsonl_path='__last_build__'").get(); - if (last && now - last.mtime < BUILD_DEBOUNCE_MS) { - return { skip: true, reason: 'recent_build' }; + if (!ignoreRecentBuild) { + const last = db.prepare("SELECT mtime FROM index_state WHERE jsonl_path='__last_build__'").get(); + if (last && now - last.mtime < BUILD_DEBOUNCE_MS) { + return { skip: true, reason: 'recent_build' }; + } } return { skip: false }; } +function isMissingIndexStateTable(error) { + const message = error instanceof Error ? error.message : String(error); + return /no such table:\s*(?:main\.)?index_state\b/i.test(message); +} + +function inspectBuildOwnership({ force = false } = {}) { + if (!fs.existsSync(DB_PATH)) return { skip: false }; + const db = openReadDb(); + try { + return shouldSkipBuild(db, { ignoreRecentBuild: force }); + } catch (error) { + // A missing table means the write path must initialize a new/legacy index. + // Any other read failure leaves daemon ownership unknown, so fail closed. + if (isMissingIndexStateTable(error)) return { skip: false }; + throw error; + } finally { + db.close(); + } +} + // A one-shot record stream that retracts a session, for routing guardian sweeps // through persist (the single db writer) instead of deleting rows directly. function* guardianDelete(sessionId) { @@ -149,78 +175,112 @@ function* guardianDelete(sessionId) { } function buildIndex({ force = false } = {}) { - const db = openDb(); - if (!force) { - const skip = shouldSkipBuild(db); - if (skip.skip) { db.close(); return; } - } - - if (force) { - db.prepare("DELETE FROM index_state WHERE jsonl_path != '__last_build__'").run(); - // Clearing index_state alone re-indexes existing files but leaves rows for - // files that no longer exist on disk (stale sessions accumulate). A force - // build is a clean rebuild: drop every derived table, then re-index from the - // current files. `memories` is the durable, human-approved layer and is never - // cleared; messages_fts is repopulated by the 'rebuild' command in finalize. - for (const table of ['messages', 'tool_calls', 'tool_results', 'sessions', 'summaries', 'subagents', 'workflows', 'workflow_agents']) { - db.prepare(`DELETE FROM ${table}`).run(); - } - } - - const files = [ - ...discoverJsonlFiles(), - ...discoverCodexJsonlFiles(), - ]; - for (const f of files) { - db.exec('BEGIN'); - try { - if (f.source === 'codex') { - // Codex goes through the pure adapter + shared persist (docs/adr/0001), - // full-reparse (countMode 'total') when the file changed. An unchanged - // file is not reparsed, but is still swept for stale guardian rows: a - // guardian/auto-review thread must never linger in the index, even if it - // was indexed before guardian detection removed it. - const { needed } = needsReindex(db, f.path); - if (needed) { - persist(db, { key: f.path, sessionId: '' }, codexParse({ key: f.path, sessionId: '' }, null)); - } else { - const guardian = readCodexGuardianThreadInfo(f.path); - if (guardian) { - persist(db, { key: f.path, sessionId: '' }, guardianDelete(codexDbId(guardian.threadRawId))); - } - } - } else { - // Claude transcripts now go through the pure adapter + shared persist - // (docs/adr/0001). needsReindex keeps the "skip unchanged file" fast path; - // the cursor's line count drives incremental resume inside parse(). - const { needed, skip } = needsReindex(db, f.path); - if (needed) { - const unit = { key: f.path, sessionId: f.sessionId, project: f.project, isSubagent: f.isSubagent, agentId: f.agentId }; - persist(db, unit, claudeParse(unit, skip > 0 ? `0:${skip}` : null)); - } - indexSubagentMeta(db, f); - } - db.exec('COMMIT'); - } catch (e) { - safeRollback(db); - process.stderr.write(`Warning: failed to index ${f.path}: ${e.message}\n`); - } - } - db.exec('BEGIN'); + const ownership = inspectBuildOwnership({ force }); + if (ownership.skip) return ownership; + const lease = acquireWriterLease({ + lockPath: writerLockPathFor(DB_PATH), + openDb: openWriterLeaseDb, + }); + if (!lease) return { skip: true, reason: 'writer_busy' }; try { - indexWorkflows(db); - refreshSessionProjectPaths(db); - indexHistory(db); - indexCodexSessionIndex(db); - db.exec("INSERT INTO messages_fts(messages_fts) VALUES('rebuild')"); - rebuildMemoryFts(db); - db.prepare("INSERT OR REPLACE INTO index_state (jsonl_path, mtime, lines_processed) VALUES ('__last_build__', ?, 0)").run(Date.now()); - db.exec('COMMIT'); - } catch (e) { - safeRollback(db); - process.stderr.write(`Warning: failed to finalize index: ${e.message}\n`); + // Ownership may change between the first read and lease acquisition. + const ownershipAfterLease = inspectBuildOwnership({ force }); + if (ownershipAfterLease.skip) return ownershipAfterLease; + + const db = openDb(); + const txDb = nodeSqliteTransactionAdapter(db); + const skippedFiles = []; + try { + try { + if (force) { + runRetryableWriteTransaction(txDb, () => { + db.prepare("DELETE FROM index_state WHERE jsonl_path != '__last_build__'").run(); + // Clearing index_state alone re-indexes existing files but leaves rows for + // files that no longer exist on disk (stale sessions accumulate). A force + // build is a clean rebuild: drop every derived table, then re-index from the + // current files. `memories` is the durable, human-approved layer and is never + // cleared; messages_fts is repopulated by the 'rebuild' command in finalize. + for (const table of ['messages', 'tool_calls', 'tool_results', 'sessions', 'summaries', 'subagents', 'workflows', 'workflow_agents']) { + db.prepare(`DELETE FROM ${table}`).run(); + } + }, { label: 'force-cleanup' }); + } + } catch (error) { + if (isBeginBusyFailure(error)) { + return { skip: true, reason: 'database_busy', skipped: skippedFiles.length, skippedFiles }; + } + throw error; + } + + const files = [ + ...discoverJsonlFiles(), + ...discoverCodexJsonlFiles(), + ]; + for (const f of files) { + try { + runRetryableWriteTransaction(txDb, () => { + if (f.source === 'codex') { + // Codex goes through the pure adapter + shared persist (docs/adr/0001), + // full-reparse (countMode 'total') when the file changed. An unchanged + // file is not reparsed, but is still swept for stale guardian rows: a + // guardian/auto-review thread must never linger in the index, even if it + // was indexed before guardian detection removed it. + const { needed } = needsReindex(db, f.path); + if (needed) { + persist(db, { key: f.path, sessionId: '' }, codexParse({ key: f.path, sessionId: '' }, null)); + } else { + const guardian = readCodexGuardianThreadInfo(f.path); + if (guardian) { + persist(db, { key: f.path, sessionId: '' }, guardianDelete(codexDbId(guardian.threadRawId))); + } + } + } else { + // Claude transcripts now go through the pure adapter + shared persist + // (docs/adr/0001). needsReindex keeps the "skip unchanged file" fast path; + // the cursor's line count drives incremental resume inside parse(). + const { needed, skip } = needsReindex(db, f.path); + if (needed) { + const unit = { key: f.path, sessionId: f.sessionId, project: f.project, isSubagent: f.isSubagent, agentId: f.agentId }; + persist(db, unit, claudeParse(unit, skip > 0 ? `0:${skip}` : null)); + } + indexSubagentMeta(db, f); + } + }, { label: `file:${f.path}` }); + } catch (e) { + if (isBeginBusyFailure(e)) { + return { skip: true, reason: 'database_busy', skipped: skippedFiles.length, skippedFiles }; + } + if (hasUnusableTransaction(e)) throw e; + // A per-file failure is skippable: log and move on. + skippedFiles.push({ path: f.path, error: e.message, diagnostics: e.obelisk }); + process.stderr.write(`Warning: failed to index ${f.path}: ${e.message}\n`); + } + } + // Finalize is one transaction and is NOT swallowed: a finalize failure fails + // the build (a half-finalized index would be inconsistent). + try { + runRetryableWriteTransaction(txDb, () => { + indexWorkflows(db); + refreshSessionProjectPaths(db); + indexHistory(db); + indexCodexSessionIndex(db); + db.exec("INSERT INTO messages_fts(messages_fts) VALUES('rebuild')"); + rebuildMemoryFts(db); + db.prepare("INSERT OR REPLACE INTO index_state (jsonl_path, mtime, lines_processed) VALUES ('__last_build__', ?, 0)").run(Date.now()); + }, { label: 'finalize' }); + } catch (error) { + if (isBeginBusyFailure(error)) { + return { skip: true, reason: 'database_busy', skipped: skippedFiles.length, skippedFiles }; + } + throw error; + } + return { skip: false, skipped: skippedFiles.length, skippedFiles }; + } finally { + db.close(); + } + } finally { + lease.release(); } - db.close(); } export { buildIndex, inferProjectPath, refreshSessionProjectPaths, shouldSkipBuild }; diff --git a/scripts/tx.ts b/scripts/tx.ts new file mode 100644 index 0000000..f3d4127 --- /dev/null +++ b/scripts/tx.ts @@ -0,0 +1,138 @@ +// Binding-agnostic SQLite write plumbing shared by the skill and app indexers +// (docs/adr/0006). The injected db must expose `exec(sql)`; this works for both +// node:sqlite (skill/CLI) and better-sqlite3 (app), same injection model as +// `persist`. + +export interface WriteTxDb { + exec(sql: string): unknown; + inTransaction(): boolean; +} + +export interface SqliteConnection { + exec(sql: string): unknown; +} + +type Phase = 'begin' | 'work' | 'commit' | 'rollback'; + +export interface WriteTxDiagnostics { + phase: Phase; + code: string | null; + label?: string; + rollbackSucceeded: boolean | null; + rollbackError: string | null; + transactionActive: boolean | null; + attempts: number; +} + +export interface WriteTxOptions { + // Diagnostic label for this transaction (e.g. a file path or 'finalize'). + label?: string; +} + +const BUSY_MESSAGE = /SQLITE_BUSY|database is locked|database is busy/i; + +function busyCode(error: unknown): string | null { + const raw = error as { code?: unknown; errcode?: unknown; message?: unknown } | null; + const code = (raw?.code ?? raw?.errcode); + if (typeof code === 'string' && code.startsWith('SQLITE_BUSY')) return code; + if (typeof raw?.message === 'string' && BUSY_MESSAGE.test(raw.message)) return 'SQLITE_BUSY'; + return null; +} + +function errorCode(error: unknown): string | null { + const raw = error as { code?: unknown } | null; + return typeof raw?.code === 'string' ? raw.code : null; +} + +interface BetterSqliteHandle { + exec(sql: string): unknown; + readonly inTransaction: boolean; +} + +interface NodeSqliteHandle { + exec(sql: string): unknown; + readonly isTransaction: boolean; +} + +export function betterSqliteTransactionAdapter(db: BetterSqliteHandle): WriteTxDb { + return { + exec: sql => db.exec(sql), + inTransaction: () => db.inTransaction, + }; +} + +export function nodeSqliteTransactionAdapter(db: NodeSqliteHandle): WriteTxDb { + return { + exec: sql => db.exec(sql), + inTransaction: () => db.isTransaction, + }; +} + +function transactionState(db: WriteTxDb): boolean | null { + try { + return db.inTransaction(); + } catch { + return null; + } +} + +function attachDiagnostics(error: unknown, diagnostics: WriteTxDiagnostics): void { + if (!error || typeof error !== 'object') return; + try { + (error as { obelisk?: WriteTxDiagnostics }).obelisk = diagnostics; + } catch { + // Frozen/native errors must still be rethrown unchanged. + } +} + +// Runs `work` exactly once inside a transaction and returns its value. Retry and +// scheduling policy belongs to the build coordinator, which knows the operation's +// idempotency and total time budget. Cleanup never masks the primary exception. +export function runWriteTransaction(db: WriteTxDb, work: () => T, options: WriteTxOptions = {}): T { + const { label } = options; + let phase: Phase = 'begin'; + try { + db.exec('BEGIN IMMEDIATE'); + phase = 'work'; + const value = work(); + phase = 'commit'; + db.exec('COMMIT'); + return value; + } catch (error) { + let rollbackSucceeded: boolean | null = null; + let rollbackError: string | null = null; + const activeBeforeRollback = transactionState(db); + if (activeBeforeRollback !== false) { + try { + db.exec('ROLLBACK'); + rollbackSucceeded = true; + } catch (rollbackFailure) { + rollbackSucceeded = false; + rollbackError = rollbackFailure instanceof Error ? rollbackFailure.message : String(rollbackFailure); + } + } + const busy = busyCode(error); + const diagnostics: WriteTxDiagnostics = { + phase, + code: busy ?? errorCode(error), + label, + rollbackSucceeded, + rollbackError, + transactionActive: transactionState(db), + attempts: 1, + }; + attachDiagnostics(error, diagnostics); + throw error; + } +} + +// Applies the connection-level pragmas used by every Obelisk writer/reader. Uses +// exec (not better-sqlite3's .pragma) so one implementation covers both bindings. +// busy_timeout is a real behavior change for node:sqlite (no default); it is set +// explicitly for better-sqlite3 too, whose own default already happens to be +// 5000ms. It is NOT the concurrency fix — see docs/adr/0006. +export function configureConnection(db: SqliteConnection, { busyTimeoutMs = 5000 } = {}): void { + db.exec(`PRAGMA busy_timeout=${busyTimeoutMs}`); + db.exec('PRAGMA journal_mode=WAL'); + db.exec('PRAGMA synchronous=NORMAL'); +} diff --git a/scripts/write-coordinator.ts b/scripts/write-coordinator.ts new file mode 100644 index 0000000..ed703bc --- /dev/null +++ b/scripts/write-coordinator.ts @@ -0,0 +1,95 @@ +// Bounded retry policy above the transaction primitive. Callers opt in only for +// idempotent work; BEGIN contention and an uncertain/live transaction are never +// retried here. + +import { runWriteTransaction, type WriteTxDb, type WriteTxOptions } from './tx.ts'; + +interface TransactionDiagnostics { + phase?: string; + code?: string | null; + transactionActive?: boolean | null; + attempts?: number; +} + +export interface WriteRetryOptions { + maxAttempts?: number; + budgetMs?: number; + retryDelayMs?: number; + now?: () => number; + sleep?: (ms: number) => void; +} + +function diagnostics(error: unknown): TransactionDiagnostics | null { + if (!error || typeof error !== 'object') return null; + return (error as { obelisk?: TransactionDiagnostics }).obelisk ?? null; +} + +function isBusyCode(code: unknown): boolean { + return typeof code === 'string' && code.startsWith('SQLITE_BUSY'); +} + +function syncSleep(ms: number): void { + if (ms <= 0) return; + try { + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, ms); + } catch { + // Bounded attempts still prevent an infinite retry loop. + } +} + +export function isBeginBusyFailure(error: unknown): boolean { + const info = diagnostics(error); + return ( + info?.phase === 'begin' && + isBusyCode(info.code) && + info.transactionActive === false + ); +} + +export function hasUnusableTransaction(error: unknown): boolean { + const info = diagnostics(error); + return Boolean(info && info.transactionActive !== false); +} + +export function isRetryableWriteFailure(error: unknown): boolean { + const info = diagnostics(error); + return ( + (info?.phase === 'work' || info?.phase === 'commit') && + isBusyCode(info.code) && + info.transactionActive === false + ); +} + +export function runWithWriteRetry(operation: () => T, { + maxAttempts = 3, + budgetMs = 1000, + retryDelayMs = 25, + now = Date.now, + sleep = syncSleep, +}: WriteRetryOptions = {}): T { + const startedAt = now(); + for (let attempt = 1; ; attempt += 1) { + try { + return operation(); + } catch (error) { + const info = diagnostics(error); + if (info) info.attempts = attempt; + if (!isRetryableWriteFailure(error) || attempt >= maxAttempts) throw error; + const remaining = budgetMs - (now() - startedAt); + if (remaining <= 0) throw error; + sleep(Math.min(retryDelayMs * attempt, remaining)); + } + } +} + +export function runRetryableWriteTransaction( + db: WriteTxDb, + work: () => T, + transactionOptions: WriteTxOptions = {}, + retryOptions: WriteRetryOptions = {}, +): T { + return runWithWriteRetry( + () => runWriteTransaction(db, work, transactionOptions), + retryOptions, + ); +} diff --git a/scripts/writer-lease.ts b/scripts/writer-lease.ts new file mode 100644 index 0000000..97ae2c7 --- /dev/null +++ b/scripts/writer-lease.ts @@ -0,0 +1,91 @@ +// Cross-process single-writer lease for a complete Obelisk index build. The +// lock lives in a dedicated SQLite database so node:sqlite and better-sqlite3 +// share identical locking semantics on every supported platform. + +import { mkdirSync } from 'node:fs'; +import { dirname, join } from 'node:path'; + +export interface WriterLeaseDb { + exec(sql: string): unknown; + close(): void; +} + +export interface WriterLease { + release(): void; +} + +export interface AcquireWriterLeaseOptions { + lockPath: string; + openDb: (path: string) => WriterLeaseDb; + waitMs?: number; + retryDelayMs?: number; + now?: () => number; + sleep?: (ms: number) => void; +} + +const BUSY_MESSAGE = /SQLITE_BUSY|database is locked|database is busy/i; + +function isBusy(error: unknown): boolean { + const raw = error as { code?: unknown; errcode?: unknown; message?: unknown } | null; + const code = raw?.code ?? raw?.errcode; + return ( + (typeof code === 'string' && code.startsWith('SQLITE_BUSY')) || + (typeof raw?.message === 'string' && BUSY_MESSAGE.test(raw.message)) + ); +} + +function syncSleep(ms: number): void { + if (ms <= 0) return; + try { + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, ms); + } catch { + // If synchronous sleeping is unavailable, the bounded attempt count below + // still prevents an infinite acquisition loop. + } +} + +export function writerLockPathFor(dbPath: string): string { + return join(dirname(dbPath), 'writer.lock.sqlite'); +} + +export function acquireWriterLease({ + lockPath, + openDb, + waitMs = 0, + retryDelayMs = 25, + now = Date.now, + sleep = syncSleep, +}: AcquireWriterLeaseOptions): WriterLease | null { + mkdirSync(dirname(lockPath), { recursive: true }); + const startedAt = now(); + const maxAttempts = waitMs > 0 ? Math.ceil(waitMs / Math.max(1, retryDelayMs)) + 1 : 1; + + for (let attempt = 0; attempt < maxAttempts; attempt += 1) { + const db = openDb(lockPath); + try { + db.exec('PRAGMA busy_timeout=0'); + db.exec('BEGIN IMMEDIATE'); + let released = false; + return { + release() { + if (released) return; + released = true; + try { + db.exec('ROLLBACK'); + } catch { + // Closing the connection releases any remaining SQLite lock. + } finally { + db.close(); + } + }, + }; + } catch (error) { + db.close(); + if (!isBusy(error)) throw error; + const remaining = waitMs - (now() - startedAt); + if (remaining <= 0 || attempt + 1 >= maxAttempts) return null; + sleep(Math.min(retryDelayMs, remaining)); + } + } + return null; +} diff --git a/tests/app-indexer-service.test.mjs b/tests/app-indexer-service.test.mjs index 52b92ac..8290be6 100644 --- a/tests/app-indexer-service.test.mjs +++ b/tests/app-indexer-service.test.mjs @@ -73,6 +73,34 @@ test('indexer service runs one pending build after an in-flight build finishes', assert.deepEqual(calls, ['first', 'pending']); }); +test('indexer service reschedules a writer-lease deferral without publishing a heartbeat', async () => { + const timers = manualTimers(); + const calls = []; + let heartbeats = 0; + const service = createIndexerService({ + buildIndex: async ({ reason, changedPaths }) => { + calls.push({ reason, changedPaths }); + return calls.length === 1 ? { deferred: true, reason: 'writer_busy' } : { deferred: false }; + }, + watchProjects: () => null, + writeHeartbeat: () => { heartbeats += 1; }, + timers, + stabilityMs: 0, + }); + + await service.runBuildNow('watch', ['project/session.jsonl']); + assert.equal(heartbeats, 0); + assert.equal(calls.length, 1); + + timers.flush(); + await service.idle(); + assert.deepEqual(calls, [ + { reason: 'watch', changedPaths: ['project/session.jsonl'] }, + { reason: 'writer-lease', changedPaths: ['project/session.jsonl'] }, + ]); + assert.equal(heartbeats, 1); +}); + test('indexer service does not log a build cancelled by a service stop', async () => { const timers = manualTimers(); const warnings = []; @@ -160,6 +188,22 @@ test('indexer service retries watcher setup when the projects directory is missi assert.equal(attempts, 2); }); +test('indexer service publishes daemon ownership as soon as it starts', () => { + const timers = manualTimers(); + let heartbeats = 0; + const service = createIndexerService({ + buildIndex: async () => ({ deferred: false }), + watchProjects: () => null, + writeHeartbeat: () => { heartbeats += 1; }, + timers, + stabilityMs: 0, + }); + + service.start({ buildOnStart: false }); + assert.equal(heartbeats, 1); + service.stop(); +}); + test('indexer service watches Claude JSON files through chokidar', async () => { const projectsDir = mkdtempSync(join(tmpdir(), 'obelisk-chokidar-projects-')); const timers = manualTimers(); diff --git a/tests/app-indexer.test.mjs b/tests/app-indexer.test.mjs index 9599986..651f715 100644 --- a/tests/app-indexer.test.mjs +++ b/tests/app-indexer.test.mjs @@ -31,7 +31,7 @@ class TestDatabase { } } -test('app indexer builds the Obelisk database from Claude JSONL and records app heartbeat', () => { +test('app indexer records build success without claiming daemon ownership', () => { const home = mkdtempSync(join(tmpdir(), 'obelisk-app-indexer-')); const claudeDir = join(home, '.claude'); const projectDir = join(claudeDir, 'projects', '-tmp-obelisk-app'); @@ -57,7 +57,7 @@ test('app indexer builds the Obelisk database from Claude JSONL and records app assert.deepEqual(firstBuild.affectedSessionIds, [sessionId]); assert.equal(firstBuild.ftsRebuilt, true); assert.equal(db.prepare("SELECT uuid FROM messages_fts WHERE messages_fts MATCH 'hello'").get().uuid, 'msg-app-1'); - assert.equal(db.prepare("SELECT jsonl_path FROM index_state WHERE jsonl_path='__app_heartbeat__'").get().jsonl_path, '__app_heartbeat__'); + assert.equal(db.prepare("SELECT jsonl_path FROM index_state WHERE jsonl_path='__app_heartbeat__'").get(), undefined); assert.equal(db.prepare("SELECT jsonl_path FROM index_state WHERE jsonl_path='__app_last_successful_build__'").get().jsonl_path, '__app_last_successful_build__'); assert.equal(db.prepare('SELECT project_path FROM sessions WHERE id=?').get(sessionId).project_path, '/tmp/obelisk-app'); db.close(); diff --git a/tests/app-main-settings.test.mjs b/tests/app-main-settings.test.mjs index 2010e2e..f6a5abe 100644 --- a/tests/app-main-settings.test.mjs +++ b/tests/app-main-settings.test.mjs @@ -7,7 +7,27 @@ import { mkdirSync, rmSync, writeFileSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; +import { acquireWriterLease } from '../scripts/writer-lease.ts'; + const require = createRequire(import.meta.url); +const { DatabaseSync } = require('node:sqlite'); + +class SqliteCompatDatabase { + constructor(dbFile) { + this.db = new DatabaseSync(dbFile); + } + pragma(statement) { this.db.exec(`PRAGMA ${statement}`); } + exec(sql) { return this.db.exec(sql); } + close() { return this.db.close(); } + prepare(sql) { + const stmt = this.db.prepare(sql); + return { + all: (...params) => stmt.all(...params), + get: (...params) => stmt.get(...params), + run: (...params) => stmt.run(...params), + }; + } +} // The app main process is now an ES module. It runs side-effectfully on import // (registers ipcMain handlers, opens windows, etc.) and has no exports, so we @@ -110,6 +130,7 @@ async function loadMainForWindowFlags(flags) { class FakeDatabase { pragma() {} + exec() {} close() {} prepare() { return { get: () => null, all: () => [], run: () => ({}) }; @@ -191,6 +212,7 @@ test('main process watches Codex sessions directory instead of Codex root', asyn class FakeDatabase { pragma() {} + exec() {} close() {} prepare() { return { get: () => null, all: () => [], run: () => ({}) }; @@ -245,6 +267,77 @@ test('main process watches Codex sessions directory instead of Codex root', asyn } }); +test('main process forwards committed IDs without reopening after a deferred build', async () => { + const originalHome = process.env.HOME; + const home = join(tmpdir(), `obelisk-main-deferred-build-${Date.now()}`); + mkdirSync(join(home, '.claude', 'projects'), { recursive: true }); + mkdirSync(join(home, '.codex', 'sessions'), { recursive: true }); + mkdirSync(join(home, '.obelisk'), { recursive: true }); + writeFileSync(join(home, '.obelisk', 'obelisk.sqlite'), ''); + process.env.HOME = home; + + let databaseOpens = 0; + let serviceOptions; + let notifications = 0; + + class FakeDatabase { + constructor() { databaseOpens += 1; } + pragma() {} + exec() {} + close() {} + prepare() { return { get: () => null, all: () => [], run: () => ({}) }; } + } + + class FakeBrowserWindow { + constructor() { + this.webContents = { on() {}, setZoomLevel() {}, openDevTools() {}, send() { notifications += 1; } }; + } + loadFile() {} + loadURL() {} + close() {} + static getAllWindows() { return [{ webContents: { send() { notifications += 1; } } }]; } + static fromWebContents() { return null; } + } + + const restore = registerMocks([ + [ELECTRON_URL, { namedExports: electronNamespace({ BrowserWindow: FakeBrowserWindow }) }], + [DATABASE_URL, { defaultExport: FakeDatabase }], + [CHOKIDAR_URL, { defaultExport: noopChokidar() }], + [INDEXER_URL, { namedExports: { writeHeartbeat() {} } }], + [INDEXER_SERVICE_URL, { + namedExports: { + createIndexerService: (options) => { + serviceOptions = options; + return { start() {}, stop() {}, idle: async () => {}, runBuildNow() { return Promise.resolve(); } }; + }, + }, + }], + [INDEXER_WORKER_URL, { + namedExports: { + createWorkerBuildIndex: () => ({ + buildIndex: async () => ({ deferred: true, reason: 'database_busy', affectedSessionIds: ['session-1'] }), + stop() {}, + }), + }, + }], + ]); + + try { + await importMain(); + const opensBeforeBuild = databaseOpens; + const notificationsBeforeBuild = notifications; + const result = await serviceOptions.buildIndex({ reason: 'writer-lease' }); + + assert.equal(result.deferred, true); + assert.equal(databaseOpens, opensBeforeBuild); + assert.equal(notifications, notificationsBeforeBuild + 2); + } finally { + restore(); + process.env.HOME = originalHome; + rmSync(home, { recursive: true, force: true }); + } +}); + test('session IPC hides Codex rows by default and supports explicit source opt-in', async () => { const originalHome = process.env.HOME; const home = join(tmpdir(), `obelisk-main-source-filter-${Date.now()}`); @@ -257,6 +350,7 @@ test('session IPC hides Codex rows by default and supports explicit source opt-i class FakeDatabase { pragma() {} + exec() {} close() {} prepare(sql) { return { @@ -369,31 +463,6 @@ test('main process migrates an existing app database before source-filtered IPC const ipcHandlers = new Map(); - // better-sqlite3-compatible adapter over node:sqlite so the real migration - // logic (ALTER TABLE ADD COLUMN source, etc.) runs against a real database. - class SqliteCompatDatabase { - constructor(dbFile) { - this.db = new DatabaseSync(dbFile); - } - pragma(statement) { - this.db.exec(`PRAGMA ${statement}`); - } - exec(sql) { - return this.db.exec(sql); - } - close() { - return this.db.close(); - } - prepare(sql) { - const stmt = this.db.prepare(sql); - return { - all: (...params) => stmt.all(...params), - get: (...params) => stmt.get(...params), - run: (...params) => stmt.run(...params), - }; - } - } - class FakeBrowserWindow { constructor() { this.webContents = { on() {}, setZoomLevel() {}, openDevTools() {}, send() {} }; @@ -441,6 +510,70 @@ test('main process migrates an existing app database before source-filtered IPC } }); +test('main process keeps schema and memory mutations behind the writer lease', async () => { + const originalHome = process.env.HOME; + const home = join(tmpdir(), `obelisk-main-migration-lease-${Date.now()}`); + const obeliskDir = join(home, '.obelisk'); + const dbPath = join(obeliskDir, 'obelisk.sqlite'); + mkdirSync(obeliskDir, { recursive: true }); + process.env.HOME = home; + + const legacy = new DatabaseSync(dbPath); + legacy.exec('CREATE TABLE sessions (id TEXT PRIMARY KEY)'); + legacy.close(); + + const holder = acquireWriterLease({ + lockPath: join(obeliskDir, 'writer.lock.sqlite'), + openDb: lockPath => new DatabaseSync(lockPath), + }); + assert.ok(holder); + const ipcHandlers = new Map(); + + class FakeBrowserWindow { + constructor() { + this.webContents = { on() {}, setZoomLevel() {}, openDevTools() {}, send() {} }; + } + loadFile() {} + loadURL() {} + close() {} + static getAllWindows() { return []; } + static fromWebContents() { return null; } + } + + const restore = registerMocks([ + [ELECTRON_URL, { + namedExports: electronNamespace({ + BrowserWindow: FakeBrowserWindow, + ipcMain: { + handle(channel, handler) { ipcHandlers.set(channel, handler); }, + }, + }), + }], + [DATABASE_URL, { defaultExport: SqliteCompatDatabase }], + [CHOKIDAR_URL, { defaultExport: noopChokidar() }], + [INDEXER_URL, { namedExports: { writeHeartbeat() {} } }], + [INDEXER_SERVICE_URL, { namedExports: defaultIndexerService() }], + [INDEXER_WORKER_URL, { namedExports: defaultIndexerWorkerClient() }], + ]); + + try { + await importMain(); + const check = new DatabaseSync(dbPath, { readOnly: true }); + const columns = check.prepare('PRAGMA table_info(sessions)').all().map(column => column.name); + check.close(); + assert.deepEqual(columns, ['id']); + assert.throws( + () => ipcHandlers.get('db:archiveMemory')(null, 'memory-1', 'test'), + /writer is busy/i, + ); + } finally { + restore(); + holder.release(); + process.env.HOME = originalHome; + rmSync(home, { recursive: true, force: true }); + } +}); + test('closing the last macOS window releases background resources until activation', async () => { const originalPlatform = Object.getOwnPropertyDescriptor(process, 'platform'); const originalHome = process.env.HOME; @@ -459,6 +592,7 @@ test('closing the last macOS window releases background resources until activati class FakeDatabase { pragma() {} + exec() {} close() { serviceEvents.push('db-close'); } prepare() { return { get: () => null, all: () => [], run: () => ({}) }; @@ -575,13 +709,16 @@ test('settings rebuild reopens the database from the configured Claude path', as const openedDbPaths = []; const buildCalls = []; const serviceEvents = []; + let competingLeaseDuringBuild; class FakeDatabase { constructor(dbPath) { + this.lockDb = dbPath.endsWith('writer.lock.sqlite') ? new DatabaseSync(dbPath) : null; openedDbPaths.push(dbPath); } pragma() {} - close() {} + exec(sql) { return this.lockDb?.exec(sql); } + close() { this.lockDb?.close(); } prepare() { return { get: () => null, all: () => [], run: () => ({}) }; } @@ -628,6 +765,12 @@ test('settings rebuild reopens the database from the configured Claude path', as buildIndex: async (args) => { serviceEvents.push('build'); buildCalls.push(args); + const competingLease = acquireWriterLease({ + lockPath: args.writerLeasePath, + openDb: lockPath => new DatabaseSync(lockPath), + }); + competingLeaseDuringBuild = Boolean(competingLease); + competingLease?.release(); writeFileSync(args.dbPath, 'rebuilt temp db'); return { files: 2, affectedSessionIds: ['session-1', 'session-2'] }; }, @@ -649,12 +792,21 @@ test('settings rebuild reopens the database from the configured Claude path', as assert.equal(buildCalls.at(-1).codexDir, customCodexDir); assert.notEqual(buildCalls.at(-1).dbPath, join(home, '.obelisk', 'obelisk.sqlite')); assert.equal(buildCalls.at(-1).preserveDbPath, join(home, '.obelisk', 'obelisk.sqlite')); + assert.equal(buildCalls.at(-1).writerLeasePath, join(home, '.obelisk', 'writer.lock.sqlite')); + assert.equal(buildCalls.at(-1).writerLeaseMode, 'caller-held'); + assert.equal(competingLeaseDuringBuild, false); assert.equal(openedDbPaths.at(-1), join(home, '.obelisk', 'obelisk.sqlite')); assert.equal( require('node:fs').readFileSync(join(home, '.obelisk', 'obelisk.sqlite'), 'utf8'), 'rebuilt temp db', ); assert.ok(serviceEvents.indexOf('build') > serviceEvents.indexOf('stop')); + const postRebuildLease = acquireWriterLease({ + lockPath: join(home, '.obelisk', 'writer.lock.sqlite'), + openDb: lockPath => new DatabaseSync(lockPath), + }); + assert.ok(postRebuildLease); + postRebuildLease.release(); } finally { restore(); process.env.HOME = originalHome; @@ -686,10 +838,15 @@ test('settings rebuild keeps the existing database after a worker failure', asyn class FakeDatabase { constructor(dbPath) { this.dbPath = dbPath; - openedDbPaths.push(dbPath); + this.lockDb = dbPath.endsWith('writer.lock.sqlite') ? new DatabaseSync(dbPath) : null; + if (!this.lockDb) openedDbPaths.push(dbPath); } pragma() {} - close() { closedDbPaths.push(this.dbPath); } + exec(sql) { return this.lockDb?.exec(sql); } + close() { + if (this.lockDb) this.lockDb.close(); + else closedDbPaths.push(this.dbPath); + } prepare() { return { get: () => null, all: () => [], run: () => ({}) }; } @@ -788,8 +945,12 @@ test('settings rebuild cancels an in-flight background build instead of waiting let buildIndexCalls = 0; class FakeDatabase { + constructor(dbPath) { + this.lockDb = dbPath.endsWith('writer.lock.sqlite') ? new DatabaseSync(dbPath) : null; + } pragma() {} - close() {} + exec(sql) { return this.lockDb?.exec(sql); } + close() { this.lockDb?.close(); } prepare() { return { get: () => null, all: () => [], run: () => ({}) }; } diff --git a/tests/app-rollback-guard.test.mjs b/tests/app-rollback-guard.test.mjs index 5fa74cd..b276eae 100644 --- a/tests/app-rollback-guard.test.mjs +++ b/tests/app-rollback-guard.test.mjs @@ -1,9 +1,7 @@ -// Regression test for the "cannot rollback - no transaction is active" crash. -// SQLite auto-rolls back certain failures (SQLITE_BUSY, disk full, ...). The -// per-file build loop then ran an explicit ROLLBACK in its catch, which threw a -// SECOND error over the real one and aborted the whole build instead of skipping -// just the bad file. buildIndex now uses a guarded rollback; this test injects a -// DB that faithfully reproduces the condition and asserts the build survives. +// Regression tests for the write-transaction runner (docs/adr/0006): +// - a transient BUSY (auto-rolled-back txn) is retried and recovers; +// - a persistent BUSY exhausts retries, and that file is SKIPPED, not fatal; +// - the guarded rollback never masks the real error ("cannot rollback ..."). import { test } from 'node:test'; import assert from 'node:assert/strict'; import { createRequire } from 'node:module'; @@ -15,78 +13,284 @@ const require = createRequire(import.meta.url); import { buildIndex } from '../app/src/main/indexer.ts'; const { DatabaseSync } = require('node:sqlite'); -// Wraps node:sqlite and simulates SQLite's auto-rollback-on-error: the first -// write inside a transaction throws a BUSY-like error AND ends the real -// transaction, so a following explicit ROLLBACK errors with "no transaction". -class AutoRollbackOnceDatabase { - constructor(dbPath) { - this.db = new DatabaseSync(dbPath); - this.inTxn = false; - this.fired = false; - } - pragma(statement) { this.db.exec(`PRAGMA ${statement}`); } - exec(sql) { - const head = sql.trim().slice(0, 8).toUpperCase(); - if (head.startsWith('BEGIN')) { this.inTxn = true; return this.db.exec('BEGIN'); } - if (head.startsWith('COMMIT')) { this.inTxn = false; return this.db.exec('COMMIT'); } - if (head.startsWith('ROLLBACK')) { - if (!this.inTxn) throw new Error('cannot rollback - no transaction is active'); +// Wraps node:sqlite and simulates SQLite auto-rollback-on-error: a poisoned write +// throws a BUSY-like error AND ends the real transaction, so a following explicit +// ROLLBACK errors with "no transaction is active". `shouldPoison(args)` decides +// which writes are poisoned. +function makeDbClass(shouldPoison) { + return class PoisonDatabase { + constructor(dbPath) { + this.db = new DatabaseSync(dbPath); this.inTxn = false; - return this.db.exec('ROLLBACK'); } - return this.db.exec(sql); - } - prepare(sql) { - const stmt = this.db.prepare(sql); - const self = this; - return { - get: (...args) => stmt.get(...args), - all: (...args) => stmt.all(...args), - run: (...args) => { - if (!self.fired && self.inTxn) { - self.fired = true; - self.db.exec('ROLLBACK'); // SQLite auto-rolled the txn back on the error - self.inTxn = false; - throw new Error('SQLITE_BUSY: database is locked'); - } - return stmt.run(...args); - }, - }; - } - close() { return this.db.close(); } + get inTransaction() { return this.inTxn; } + pragma(statement) { this.db.exec(`PRAGMA ${statement}`); } + exec(sql) { + const head = sql.trim().slice(0, 8).toUpperCase(); + if (head.startsWith('BEGIN')) { this.inTxn = true; return this.db.exec('BEGIN'); } + if (head.startsWith('COMMIT')) { this.inTxn = false; return this.db.exec('COMMIT'); } + if (head.startsWith('ROLLBACK')) { + if (!this.inTxn) throw new Error('cannot rollback - no transaction is active'); + this.inTxn = false; + return this.db.exec('ROLLBACK'); + } + return this.db.exec(sql); + } + prepare(sql) { + const stmt = this.db.prepare(sql); + const self = this; + return { + get: (...args) => stmt.get(...args), + all: (...args) => stmt.all(...args), + run: (...args) => { + if (self.inTxn && shouldPoison(args)) { + self.db.exec('ROLLBACK'); // SQLite auto-rolled the txn back on the error + self.inTxn = false; + throw new Error('SQLITE_BUSY: database is locked'); + } + return stmt.run(...args); + }, + }; + } + close() { return this.db.close(); } + }; } -test('a per-file write that auto-rolls-back the transaction is skipped, not fatal', () => { - const home = mkdtempSync(join(tmpdir(), 'obelisk-rollback-guard-')); +function twoFileHome(alphaContent, betaContent) { + const home = mkdtempSync(join(tmpdir(), 'obelisk-tx-')); const projectDir = join(home, '.claude', 'projects', '-tmp-proj'); mkdirSync(projectDir, { recursive: true }); - const msg = (uuid) => JSON.stringify({ - uuid, type: 'user', timestamp: '2026-06-10T10:00:00Z', cwd: '/tmp/proj', - message: { role: 'user', content: `hello ${uuid}` }, - }) + '\n'; - writeFileSync(join(projectDir, 'alpha.jsonl'), msg('a1')); - writeFileSync(join(projectDir, 'beta.jsonl'), msg('b1')); - - const dbPath = join(home, '.obelisk', 'obelisk.sqlite'); mkdirSync(join(home, '.obelisk'), { recursive: true }); + const msg = (uuid, content) => JSON.stringify({ + uuid, type: 'user', timestamp: '2026-06-10T10:00:00Z', cwd: '/tmp/proj', + message: { role: 'user', content }, + }) + '\n'; + writeFileSync(join(projectDir, 'alpha.jsonl'), msg('a1', alphaContent)); + writeFileSync(join(projectDir, 'beta.jsonl'), msg('b1', betaContent)); + return { home, dbPath: join(home, '.obelisk', 'obelisk.sqlite'), projectsDir: join(home, '.claude', 'projects') }; +} + +function subagentHome(description = 'first description') { + const home = mkdtempSync(join(tmpdir(), 'obelisk-meta-tx-')); + const projectDir = join(home, '.claude', 'projects', '-tmp-proj'); + const subagentDir = join(projectDir, 'session', 'subagents'); + const dbPath = join(home, '.obelisk', 'obelisk.sqlite'); + mkdirSync(subagentDir, { recursive: true }); + mkdirSync(join(home, '.obelisk'), { recursive: true }); + const message = uuid => JSON.stringify({ + uuid, + type: 'user', + timestamp: '2026-06-10T10:00:00Z', + cwd: '/tmp/proj', + message: { role: 'user', content: `message ${uuid}` }, + }) + '\n'; + writeFileSync(join(projectDir, 'session.jsonl'), message('main-message')); + writeFileSync(join(subagentDir, 'agent.jsonl'), message('agent-message')); + const metaPath = join(subagentDir, 'agent.meta.json'); + writeFileSync(metaPath, JSON.stringify({ agentType: 'Explore', description })); + return { + home, + dbPath, + projectsDir: join(home, '.claude', 'projects'), + metaPath, + changedMetaPath: join('-tmp-proj', 'session', 'subagents', 'agent.meta.json'), + }; +} + +function run(home, dbPath, projectsDir, DatabaseImpl) { + return buildIndex({ + force: true, + claudeDir: join(home, '.claude'), + codexDir: join(home, '.codex'), + projectsDir, + dbPath, + DatabaseImpl, + }); +} + +function makeBeginBusyDbClass(shouldFail) { + const Base = makeDbClass(() => false); + return class BeginBusyDatabase extends Base { + constructor(dbPath) { + super(dbPath); + this.isWriterLease = dbPath.endsWith('writer.lock.sqlite'); + this.beginCalls = 0; + } + exec(sql) { + if (!this.isWriterLease && sql.trim().toUpperCase().startsWith('BEGIN')) { + this.beginCalls += 1; + if (shouldFail(this.beginCalls)) { + throw Object.assign(new Error('database is locked'), { code: 'SQLITE_BUSY' }); + } + } + return super.exec(sql); + } + }; +} + +test('a failed changed file is not reported as an affected session', () => { + const { home, dbPath, projectsDir } = twoFileHome('POISON alpha', 'hello beta'); + const Db = makeDbClass((args) => args.some(a => typeof a === 'string' && a.includes('POISON'))); + + const result = buildIndex({ + force: false, + claudeDir: join(home, '.claude'), + codexDir: join(home, '.codex'), + projectsDir, + dbPath, + DatabaseImpl: Db, + changedPaths: ['-tmp-proj/alpha.jsonl'], + }); + + assert.deepEqual(result.affectedSessionIds, []); + assert.equal(result.skipped, 1); +}); + +test('a transient BUSY during force cleanup is retried and recovers', () => { + const { home, dbPath, projectsDir } = twoFileHome('hello alpha', 'hello beta'); + // Fire exactly once, on the first write inside a transaction, then never again. + let fired = false; + const Db = makeDbClass(() => (fired ? false : (fired = true))); - // The unguarded ROLLBACK used to make this throw the masking rollback error. let result; - assert.doesNotThrow(() => { - result = buildIndex({ - force: true, - claudeDir: join(home, '.claude'), - codexDir: join(home, '.codex'), - projectsDir: join(home, '.claude', 'projects'), - dbPath, - DatabaseImpl: AutoRollbackOnceDatabase, - }); - }, 'build must not abort on a transaction-aborting per-file error'); + assert.doesNotThrow(() => { result = run(home, dbPath, projectsDir, Db); }); - // Exactly one file's write was poisoned; the other indexed cleanly. const check = new DatabaseSync(dbPath); const sessions = check.prepare('SELECT COUNT(*) AS c FROM sessions').get().c; check.close(); - assert.equal(sessions, 1, 'the surviving file is indexed; the poisoned one is skipped'); - assert.ok(result.files >= 2, 'both files were discovered'); + assert.equal(sessions, 2, 'the retried file recovered; both files indexed'); + assert.equal(result.skipped, 0, 'nothing was skipped'); +}); + +test('a transient BUSY during a file transaction is retried and recovers', () => { + const { home, dbPath, projectsDir } = twoFileHome('hello alpha', 'hello beta'); + let fired = false; + const Db = makeDbClass((args) => { + const isAlphaWrite = args.some(arg => typeof arg === 'string' && arg.includes('hello alpha')); + if (!isAlphaWrite || fired) return false; + fired = true; + return true; + }); + + const result = run(home, dbPath, projectsDir, Db); + const check = new DatabaseSync(dbPath); + const sessions = check.prepare('SELECT COUNT(*) AS c FROM sessions').get().c; + check.close(); + assert.equal(sessions, 2); + assert.equal(result.skipped, 0); +}); + +test('a persistent BUSY exhausts retries and skips just that file, not the build', () => { + const { home, dbPath, projectsDir } = twoFileHome('POISON alpha', 'hello beta'); + // Always poison writes that carry alpha's marker text; beta is untouched. + const Db = makeDbClass((args) => args.some(a => typeof a === 'string' && a.includes('POISON'))); + + let result; + assert.doesNotThrow(() => { result = run(home, dbPath, projectsDir, Db); }); + + const check = new DatabaseSync(dbPath); + const sessions = check.prepare('SELECT id FROM sessions ORDER BY id').all().map(r => r.id); + check.close(); + assert.deepEqual(sessions, ['beta'], 'the persistently-failing file is skipped; the other indexes'); + assert.equal(result.skipped, 1, 'the skipped file is reported in the build result'); + assert.equal(result.skippedFiles[0].diagnostics?.phase, 'work', 'diagnostics record the failing phase'); +}); + +test('BEGIN contention during force cleanup defers the build', () => { + const { home, dbPath, projectsDir } = twoFileHome('hello alpha', 'hello beta'); + const Db = makeBeginBusyDbClass(beginCall => beginCall === 1); + + const result = run(home, dbPath, projectsDir, Db); + assert.equal(result.deferred, true); + assert.equal(result.reason, 'database_busy'); +}); + +test('BEGIN contention during finalize defers the build', () => { + const { home, dbPath, projectsDir } = twoFileHome('hello alpha', 'hello beta'); + const Db = makeBeginBusyDbClass(beginCall => beginCall === 3); + + const result = buildIndex({ + force: false, + claudeDir: join(home, '.claude'), + codexDir: join(home, '.codex'), + projectsDir, + dbPath, + DatabaseImpl: Db, + }); + assert.equal(result.deferred, true); + assert.equal(result.reason, 'database_busy'); +}); + +test('a finalize database error is propagated instead of swallowed as malformed input', () => { + const { home, dbPath, projectsDir } = twoFileHome('hello alpha', 'hello beta'); + const workflowDir = join(projectsDir, '-tmp-proj', 'alpha', 'workflows'); + mkdirSync(workflowDir, { recursive: true }); + writeFileSync(join(workflowDir, 'run.json'), JSON.stringify({ + runId: 'workflow-1', + workflowName: 'POISON WORKFLOW', + })); + const Db = makeDbClass(args => args.some(arg => typeof arg === 'string' && arg.includes('POISON WORKFLOW'))); + + assert.throws(() => buildIndex({ + force: false, + claudeDir: join(home, '.claude'), + codexDir: join(home, '.codex'), + projectsDir, + dbPath, + DatabaseImpl: Db, + }), /SQLITE_BUSY/); +}); + +test('a changed subagent meta file is applied and reported only after its file transaction commits', () => { + const { home, dbPath, projectsDir, metaPath, changedMetaPath } = subagentHome(); + const Db = makeDbClass(() => false); + buildIndex({ + force: true, + claudeDir: join(home, '.claude'), + codexDir: join(home, '.codex'), + projectsDir, + dbPath, + DatabaseImpl: Db, + }); + writeFileSync(metaPath, JSON.stringify({ agentType: 'Explore', description: 'updated description' })); + + const result = buildIndex({ + force: false, + changedPaths: [changedMetaPath], + claudeDir: join(home, '.claude'), + codexDir: join(home, '.codex'), + projectsDir, + dbPath, + DatabaseImpl: Db, + }); + const check = new DatabaseSync(dbPath, { readOnly: true }); + assert.equal(check.prepare('SELECT description FROM subagents WHERE agent_id=?').get('agent').description, 'updated description'); + check.close(); + assert.deepEqual(result.affectedSessionIds, ['session']); +}); + +test('a failed subagent meta transaction does not report its session as affected', () => { + const { home, dbPath, projectsDir, metaPath, changedMetaPath } = subagentHome(); + buildIndex({ + force: true, + claudeDir: join(home, '.claude'), + codexDir: join(home, '.codex'), + projectsDir, + dbPath, + DatabaseImpl: makeDbClass(() => false), + }); + writeFileSync(metaPath, JSON.stringify({ agentType: 'Explore', description: 'POISON META' })); + const FailingDb = makeDbClass(args => args.some(arg => typeof arg === 'string' && arg.includes('POISON META'))); + + const result = buildIndex({ + force: false, + changedPaths: [changedMetaPath], + claudeDir: join(home, '.claude'), + codexDir: join(home, '.codex'), + projectsDir, + dbPath, + DatabaseImpl: FailingDb, + }); + assert.deepEqual(result.affectedSessionIds, []); + assert.equal(result.skipped, 1); }); diff --git a/tests/app-writer-lease.test.mjs b/tests/app-writer-lease.test.mjs new file mode 100644 index 0000000..f535560 --- /dev/null +++ b/tests/app-writer-lease.test.mjs @@ -0,0 +1,97 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { createRequire } from 'node:module'; +import { existsSync, mkdirSync, mkdtempSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +import { buildIndex } from '../app/src/main/indexer.ts'; +import { acquireWriterLease, writerLockPathFor } from '../scripts/writer-lease.ts'; + +const require = createRequire(import.meta.url); +const { DatabaseSync } = require('node:sqlite'); + +class TestDatabase { + constructor(dbPath) { + this.db = new DatabaseSync(dbPath); + } + get inTransaction() { return this.db.isTransaction; } + exec(sql) { return this.db.exec(sql); } + pragma(statement) { return this.db.exec(`PRAGMA ${statement}`); } + prepare(sql) { return this.db.prepare(sql); } + close() { return this.db.close(); } +} + +test('an app build defers without opening the target database when another writer owns the lease', () => { + const home = mkdtempSync(join(tmpdir(), 'obelisk-app-writer-lease-')); + const claudeDir = join(home, '.claude'); + const projectsDir = join(claudeDir, 'projects'); + const projectDir = join(projectsDir, '-tmp-project'); + const dbPath = join(home, '.obelisk', 'obelisk.sqlite'); + mkdirSync(projectDir, { recursive: true }); + mkdirSync(join(home, '.obelisk'), { recursive: true }); + writeFileSync(join(projectDir, 'session.jsonl'), JSON.stringify({ + uuid: 'message-1', + type: 'user', + timestamp: '2026-07-10T00:00:00Z', + message: { role: 'user', content: 'hello' }, + }) + '\n'); + + const lease = acquireWriterLease({ + lockPath: writerLockPathFor(dbPath), + openDb: path => new DatabaseSync(path), + }); + assert.ok(lease); + try { + const result = buildIndex({ + claudeDir, + projectsDir, + dbPath, + DatabaseImpl: TestDatabase, + writerLeaseWaitMs: 0, + }); + assert.equal(result.deferred, true); + assert.equal(result.reason, 'writer_busy'); + assert.equal(existsSync(dbPath), false); + } finally { + lease.release(); + } +}); + +test('a failed force cleanup leaves the existing index intact', () => { + const home = mkdtempSync(join(tmpdir(), 'obelisk-app-force-atomic-')); + const claudeDir = join(home, '.claude'); + const projectsDir = join(claudeDir, 'projects'); + const projectDir = join(projectsDir, '-tmp-project'); + const dbPath = join(home, '.obelisk', 'obelisk.sqlite'); + mkdirSync(projectDir, { recursive: true }); + mkdirSync(join(home, '.obelisk'), { recursive: true }); + writeFileSync(join(projectDir, 'session.jsonl'), JSON.stringify({ + uuid: 'message-1', + type: 'user', + timestamp: '2026-07-10T00:00:00Z', + message: { role: 'user', content: 'hello' }, + }) + '\n'); + buildIndex({ claudeDir, projectsDir, dbPath, DatabaseImpl: TestDatabase }); + + class FailingCleanupDatabase extends TestDatabase { + prepare(sql) { + const statement = super.prepare(sql); + if (!sql.includes('DELETE FROM sessions')) return statement; + return { + get: (...args) => statement.get(...args), + all: (...args) => statement.all(...args), + run: () => { throw new Error('cleanup interrupted'); }, + }; + } + } + + assert.throws( + () => buildIndex({ claudeDir, projectsDir, dbPath, DatabaseImpl: FailingCleanupDatabase, force: true }), + /cleanup interrupted/, + ); + const check = new DatabaseSync(dbPath, { readOnly: true }); + assert.equal(check.prepare('SELECT COUNT(*) AS count FROM sessions').get().count, 1); + assert.equal(check.prepare('SELECT COUNT(*) AS count FROM messages').get().count, 1); + check.close(); +}); diff --git a/tests/daemon-arbitration.test.mjs b/tests/daemon-arbitration.test.mjs new file mode 100644 index 0000000..c2f04cc --- /dev/null +++ b/tests/daemon-arbitration.test.mjs @@ -0,0 +1,133 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { createRequire } from 'node:module'; +import { mkdirSync, mkdtempSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join, resolve } from 'node:path'; +import { spawnSync } from 'node:child_process'; + +import { acquireWriterLease, writerLockPathFor } from '../scripts/writer-lease.ts'; + +const require = createRequire(import.meta.url); +const { DatabaseSync } = require('node:sqlite'); +const repoRoot = resolve(new URL('..', import.meta.url).pathname); + +test('a passive query does not mutate the index while a fresh daemon owns writes', () => { + const home = mkdtempSync(join(tmpdir(), 'obelisk-daemon-arbitration-')); + const obeliskDir = join(home, '.obelisk'); + const dbPath = join(obeliskDir, 'obelisk.sqlite'); + mkdirSync(obeliskDir, { recursive: true }); + + const db = new DatabaseSync(dbPath); + db.exec('CREATE TABLE index_state (jsonl_path TEXT PRIMARY KEY, mtime REAL, lines_processed INTEGER)'); + const marker = db.prepare('INSERT INTO index_state (jsonl_path, mtime, lines_processed) VALUES (?, ?, 0)'); + const now = Date.now(); + marker.run('__app_heartbeat__', now); + db.close(); + + const queryPath = join(home, 'query.mjs'); + writeFileSync(queryPath, "return 'read-only';"); + const result = spawnSync(process.execPath, ['scripts/runtime.mjs', '--query', queryPath], { + cwd: repoRoot, + env: { ...process.env, HOME: home }, + encoding: 'utf8', + }); + assert.equal(result.status, 0, result.stderr || result.stdout); + assert.equal(JSON.parse(result.stdout), 'read-only'); + + const check = new DatabaseSync(dbPath, { readOnly: true }); + const tables = check.prepare("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name").all().map(row => row.name); + check.close(); + assert.deepEqual(tables, ['index_state']); +}); + +test('attune refuses to mutate the index while a fresh daemon owns writes', () => { + const home = mkdtempSync(join(tmpdir(), 'obelisk-daemon-attune-')); + const obeliskDir = join(home, '.obelisk'); + const dbPath = join(obeliskDir, 'obelisk.sqlite'); + mkdirSync(obeliskDir, { recursive: true }); + const db = new DatabaseSync(dbPath); + db.exec('CREATE TABLE index_state (jsonl_path TEXT PRIMARY KEY, mtime REAL, lines_processed INTEGER)'); + const marker = db.prepare('INSERT INTO index_state (jsonl_path, mtime, lines_processed) VALUES (?, ?, 0)'); + const now = Date.now(); + marker.run('__app_heartbeat__', now); + db.close(); + + const attunePath = join(home, 'attune.mjs'); + writeFileSync(attunePath, 'return true;'); + const result = spawnSync(process.execPath, ['scripts/runtime.mjs', '--attune', attunePath], { + cwd: repoRoot, + env: { ...process.env, HOME: home }, + encoding: 'utf8', + }); + assert.equal(result.status, 1); + assert.match(JSON.parse(result.stdout).error, /daemon owns index writes/i); + + const check = new DatabaseSync(dbPath, { readOnly: true }); + const tables = check.prepare("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name").all().map(row => row.name); + check.close(); + assert.deepEqual(tables, ['index_state']); +}); + +test('a passive query stays read-only when another process holds the writer lease', () => { + const home = mkdtempSync(join(tmpdir(), 'obelisk-writer-owned-')); + const obeliskDir = join(home, '.obelisk'); + const dbPath = join(obeliskDir, 'obelisk.sqlite'); + mkdirSync(obeliskDir, { recursive: true }); + const db = new DatabaseSync(dbPath); + db.exec('CREATE TABLE index_state (jsonl_path TEXT PRIMARY KEY, mtime REAL, lines_processed INTEGER)'); + db.close(); + + const lease = acquireWriterLease({ + lockPath: writerLockPathFor(dbPath), + openDb: path => new DatabaseSync(path), + }); + assert.ok(lease); + try { + const queryPath = join(home, 'query.mjs'); + writeFileSync(queryPath, "return 'writer-busy';"); + const result = spawnSync(process.execPath, ['scripts/runtime.mjs', '--query', queryPath], { + cwd: repoRoot, + env: { ...process.env, HOME: home }, + encoding: 'utf8', + }); + assert.equal(result.status, 0, result.stderr || result.stdout); + assert.equal(JSON.parse(result.stdout), 'writer-busy'); + } finally { + lease.release(); + } + + const check = new DatabaseSync(dbPath, { readOnly: true }); + const tables = check.prepare("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name").all().map(row => row.name); + check.close(); + assert.deepEqual(tables, ['index_state']); +}); + +test('a passive query fails closed when daemon ownership cannot be read', () => { + const home = mkdtempSync(join(tmpdir(), 'obelisk-daemon-ownership-error-')); + const obeliskDir = join(home, '.obelisk'); + const dbPath = join(obeliskDir, 'obelisk.sqlite'); + mkdirSync(obeliskDir, { recursive: true }); + const db = new DatabaseSync(dbPath); + db.exec('CREATE TABLE index_state (jsonl_path TEXT PRIMARY KEY)'); + db.close(); + + const lease = acquireWriterLease({ + lockPath: writerLockPathFor(dbPath), + openDb: path => new DatabaseSync(path), + }); + assert.ok(lease); + try { + const queryPath = join(home, 'query.mjs'); + writeFileSync(queryPath, "return 'ownership-unknown';"); + const result = spawnSync(process.execPath, ['scripts/runtime.mjs', '--query', queryPath], { + cwd: repoRoot, + env: { ...process.env, HOME: home }, + encoding: 'utf8', + }); + assert.equal(result.status, 1, result.stderr || result.stdout); + assert.match(JSON.parse(result.stdout).error, /no such column: mtime/i); + } finally { + lease.release(); + } +}); diff --git a/tests/indexer.test.mjs b/tests/indexer.test.mjs index d308283..e94f2d6 100644 --- a/tests/indexer.test.mjs +++ b/tests/indexer.test.mjs @@ -56,7 +56,7 @@ test('refreshSessionProjectPaths repairs indexed sessions from message cwd', () db.close(); }); -test('shouldSkipBuild requires both fresh app heartbeat and successful app build', () => { +test('shouldSkipBuild treats a fresh heartbeat alone as daemon write ownership', () => { const db = new DatabaseSync(':memory:'); db.exec(` CREATE TABLE index_state ( @@ -68,17 +68,19 @@ test('shouldSkipBuild requires both fresh app heartbeat and successful app build 100000, ); - assert.equal(shouldSkipBuild(db, { now: 110000 }).skip, false); + assert.deepEqual( + shouldSkipBuild(db, { now: 110000 }), + { skip: true, reason: 'daemon_active' }, + ); + + db.prepare('DELETE FROM index_state').run(); db.prepare('INSERT INTO index_state (jsonl_path, mtime, lines_processed) VALUES (?, ?, 0)').run( '__app_last_successful_build__', 100000, ); - assert.deepEqual( - shouldSkipBuild(db, { now: 110000 }), - { skip: true, reason: 'app_successful_build' }, - ); + assert.equal(shouldSkipBuild(db, { now: 110000 }).skip, false); assert.equal(shouldSkipBuild(db, { now: 200000 }).skip, false); db.close(); }); diff --git a/tests/write-transaction.test.mjs b/tests/write-transaction.test.mjs new file mode 100644 index 0000000..b455618 --- /dev/null +++ b/tests/write-transaction.test.mjs @@ -0,0 +1,166 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { createRequire } from 'node:module'; +import { mkdtempSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +import { nodeSqliteTransactionAdapter, runWriteTransaction } from '../scripts/tx.ts'; +import { hasUnusableTransaction, isBeginBusyFailure, isRetryableWriteFailure } from '../scripts/write-coordinator.ts'; + +const require = createRequire(import.meta.url); +const { DatabaseSync } = require('node:sqlite'); + +test('a failed rollback with an active transaction never retries or masks the primary error', () => { + const primary = Object.assign(new Error('database is locked'), { code: 'SQLITE_BUSY' }); + let active = false; + let workCalls = 0; + + const db = { + exec(sql) { + if (sql.startsWith('BEGIN')) { + if (active) throw new Error('cannot start a transaction within a transaction'); + active = true; + } else if (sql === 'ROLLBACK') { + throw new Error('rollback failed with I/O error'); + } else if (sql === 'COMMIT') { + active = false; + } + }, + inTransaction() { + return active; + }, + }; + + assert.throws( + () => runWriteTransaction(db, () => { + workCalls += 1; + throw primary; + }), + error => error === primary, + ); + assert.equal(workCalls, 1); + assert.equal(active, true); +}); + +test('an automatic rollback rethrows the primary error without issuing another rollback', () => { + const primary = Object.assign(new Error('database is locked'), { code: 'SQLITE_BUSY' }); + let active = false; + let rollbackCalls = 0; + const db = { + exec(sql) { + if (sql.startsWith('BEGIN')) active = true; + if (sql === 'ROLLBACK') rollbackCalls += 1; + }, + inTransaction() { + return active; + }, + }; + + assert.throws( + () => runWriteTransaction(db, () => { + active = false; + throw primary; + }), + error => error === primary, + ); + assert.equal(rollbackCalls, 0); + assert.equal(primary.obelisk.transactionActive, false); +}); + +test('an active transaction is rolled back once before the primary error is rethrown', () => { + const primary = new Error('persist failed'); + let active = false; + let rollbackCalls = 0; + const db = { + exec(sql) { + if (sql.startsWith('BEGIN')) active = true; + if (sql === 'ROLLBACK') { + rollbackCalls += 1; + active = false; + } + }, + inTransaction() { + return active; + }, + }; + + assert.throws(() => runWriteTransaction(db, () => { throw primary; }), error => error === primary); + assert.equal(rollbackCalls, 1); + assert.equal(primary.obelisk.rollbackSucceeded, true); + assert.equal(primary.obelisk.transactionActive, false); +}); + +test('an unknown post-error transaction state is unsafe for the next file', () => { + const primary = new Error('transaction state unavailable'); + const db = { + exec() {}, + inTransaction() { + throw new Error('binding cannot report transaction state'); + }, + }; + + assert.throws(() => runWriteTransaction(db, () => { throw primary; }), error => error === primary); + assert.equal(primary.obelisk.transactionActive, null); + assert.equal(hasUnusableTransaction(primary), true); +}); + +test('node:sqlite generic error codes preserve BUSY classification from the message', () => { + const primary = Object.assign(new Error('database is locked'), { code: 'ERR_SQLITE_ERROR' }); + let active = false; + const db = { + exec(sql) { + if (sql === 'BEGIN IMMEDIATE') active = true; + }, + inTransaction() { return active; }, + }; + + assert.throws(() => runWriteTransaction(db, () => { + active = false; + throw primary; + }), error => error === primary); + assert.equal(primary.obelisk.code, 'SQLITE_BUSY'); + assert.equal(isRetryableWriteFailure(primary), true); +}); + +test('a real node:sqlite BEGIN lock is classified as a deferrable BUSY', () => { + const dbPath = join(mkdtempSync(join(tmpdir(), 'obelisk-node-sqlite-busy-')), 'index.sqlite'); + const holder = new DatabaseSync(dbPath); + const contender = new DatabaseSync(dbPath); + holder.exec('PRAGMA busy_timeout=0; CREATE TABLE test (value TEXT); BEGIN IMMEDIATE'); + contender.exec('PRAGMA busy_timeout=0'); + + try { + assert.throws( + () => runWriteTransaction(nodeSqliteTransactionAdapter(contender), () => {}), + error => isBeginBusyFailure(error) && error.obelisk.code === 'SQLITE_BUSY', + ); + } finally { + holder.exec('ROLLBACK'); + holder.close(); + contender.close(); + } +}); + +test('a BEGIN failure with an active transaction is not deferrable', () => { + const primary = Object.assign(new Error('database is locked'), { code: 'SQLITE_BUSY' }); + let rollbackCalls = 0; + const db = { + exec(sql) { + if (sql === 'BEGIN IMMEDIATE') throw primary; + if (sql === 'ROLLBACK') { + rollbackCalls += 1; + throw new Error('rollback failed with I/O error'); + } + }, + inTransaction() { + return true; + }, + }; + + assert.throws(() => runWriteTransaction(db, () => {}), error => error === primary); + assert.equal(rollbackCalls, 1); + assert.equal(primary.obelisk.transactionActive, true); + assert.equal(hasUnusableTransaction(primary), true); + assert.equal(isBeginBusyFailure(primary), false); +}); diff --git a/tests/writer-lease.test.mjs b/tests/writer-lease.test.mjs new file mode 100644 index 0000000..997e260 --- /dev/null +++ b/tests/writer-lease.test.mjs @@ -0,0 +1,25 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { createRequire } from 'node:module'; +import { mkdtempSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +import { acquireWriterLease } from '../scripts/writer-lease.ts'; + +const require = createRequire(import.meta.url); +const { DatabaseSync } = require('node:sqlite'); + +test('a writer lease excludes another writer until it is released', () => { + const lockPath = join(mkdtempSync(join(tmpdir(), 'obelisk-writer-lease-')), 'writer.lock.sqlite'); + const openDb = path => new DatabaseSync(path); + + const first = acquireWriterLease({ lockPath, openDb }); + assert.ok(first); + assert.equal(acquireWriterLease({ lockPath, openDb }), null); + + first.release(); + const afterRelease = acquireWriterLease({ lockPath, openDb }); + assert.ok(afterRelease); + afterRelease.release(); +}); diff --git a/tsconfig.json b/tsconfig.json index 95cd83d..0d2929f 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -7,6 +7,7 @@ "types": ["node"], "strict": true, "noEmit": true, + "allowImportingTsExtensions": true, "allowJs": true, "checkJs": false, "erasableSyntaxOnly": true,