Implement the full ADR-0006 plan: three-layer separation of transaction
correctness, retry policy, and cross-process writer coordination.
Layer 1 — scripts/tx.ts (transaction correctness):
- runWriteTransaction executes work exactly once; no internal retry.
- BEGIN IMMEDIATE takes the write lock up front (avoids SQLITE_BUSY_SNAPSHOT).
- Guarded rollback: checks inTransaction() via adapter before attempting
ROLLBACK; never masks the primary exception.
- WriteTxDiagnostics attached to errors: phase, code, label,
rollbackSucceeded, rollbackError, transactionActive.
- Binding adapters (betterSqliteTransactionAdapter, nodeSqliteTransactionAdapter)
mapping better-sqlite3's `.inTransaction` and node:sqlite's `.isTransaction`.
- configureConnection centralizes WAL + synchronous + busy_timeout.
Layer 2 — scripts/write-coordinator.ts (retry policy):
- runRetryableWriteTransaction: bounded retry with total time budget.
- Only retries when the transaction confirmed ended (transactionActive=false)
and the error is SQLITE_BUSY during work/commit phase.
- BEGIN-phase BUSY = abort entire build (isBeginBusyFailure); the caller
returns `{ deferred: true, reason: 'writer_busy' }` instead of waiting.
- hasUnusableTransaction detects a still-active transaction after failure;
aborts the build immediately, never retries.
Layer 3 — scripts/writer-lease.ts (cross-process coordination):
- acquireWriterLease: dedicated writer.lock.sqlite with busy_timeout=0 +
BEGIN IMMEDIATE. Non-blocking attempt; bounded wait with retryDelayMs.
- writerLockPathFor derives lock path from the target DB path.
- Lease held for the entire build; released on completion or failure.
- Lock DB uses DELETE journal (not WAL); crash/close auto-releases.
- All consumers obey: skill acquires at build start (returns deferred if
unavailable); app daemon (via worker) acquires for its build cycle.
Build semantics changes:
- affectedSessionIds updated only after successful commit.
- BuildIndexResult gains skipped/skippedFiles for observability.
- Skill finalize failure now fails the build (was silently warned).
- Checkpoint changed to PASSIVE (TRUNCATE reserved for maintenance/exit).
- Skill buildIndex returns { deferred, reason } on lease contention;
indexer-service reschedules the build (deferredRetryMs) without publishing
a heartbeat (so the build-deferred state is visible to cross-process
arbitration).
- Service publishes heartbeat immediately on start() for correct arbitration.
Tests:
- tests/write-transaction.test.mjs: single-shot execution, diagnostics
propagation, auto-rolled-back transaction detected, rollback failure
captured as metadata, BEGIN IMMEDIATE semantics.
- tests/writer-lease.test.mjs: acquire/release, contention returns null,
bounded wait with release during budget.
- tests/app-writer-lease.test.mjs: better-sqlite3 adapter integration.
- tests/app-rollback-guard.test.mjs: rewritten — transient BUSY recovered
by coordinator, persistent BUSY skips file, begin-busy aborts build,
live-transaction aborts build, phantom affectedSessionIds prevented.
- tests/daemon-arbitration.test.mjs: skill defers to fresh app heartbeat,
builds when heartbeat is stale.
- tests/app-indexer-service.test.mjs: new cases for deferred-retry
scheduling and immediate heartbeat on start.
- app/tests/electron-concurrency.mjs + child: dual-child IPC structure for
real better-sqlite3 contention (holder acquires lock → build child starts
→ delayed release → result collected; persistent contention bounded).
ADR-0006 updated to reflect the implemented design.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
78 lines
2.9 KiB
JavaScript
78 lines
2.9 KiB
JavaScript
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');
|
|
const os = require('node:os');
|
|
const { DatabaseSync } = require('node:sqlite');
|
|
|
|
const OBELISK_DIR = path.join(os.homedir(), '.obelisk');
|
|
const LEGACY_DB_PATH = path.join(CLAUDE_DIR, 'obelisk.sqlite');
|
|
const DB_PATH = path.join(OBELISK_DIR, 'obelisk.sqlite');
|
|
const SCHEMA = fs.readFileSync(new URL('./schema.sql', import.meta.url), 'utf8');
|
|
|
|
function migrateLegacyDbIfNeeded() {
|
|
if (fs.existsSync(DB_PATH)) return;
|
|
if (!fs.existsSync(LEGACY_DB_PATH)) return;
|
|
fs.mkdirSync(path.dirname(DB_PATH), { recursive: true });
|
|
fs.copyFileSync(LEGACY_DB_PATH, DB_PATH);
|
|
}
|
|
|
|
function openDb() {
|
|
migrateLegacyDbIfNeeded();
|
|
fs.mkdirSync(path.dirname(DB_PATH), { recursive: true });
|
|
const db = new DatabaseSync(DB_PATH);
|
|
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}`);
|
|
}
|
|
|
|
function tableExists(db, table) {
|
|
return Boolean(db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name=?").get(table));
|
|
}
|
|
|
|
function migrateExistingColumns(db) {
|
|
if (tableExists(db, 'sessions')) ensureColumn(db, 'sessions', 'source', "TEXT DEFAULT 'claude'");
|
|
if (tableExists(db, 'messages')) {
|
|
ensureColumn(db, 'messages', 'content_type', 'TEXT');
|
|
ensureColumn(db, 'messages', 'is_meta', 'INTEGER DEFAULT 0');
|
|
ensureColumn(db, 'messages', 'source', "TEXT DEFAULT 'claude'");
|
|
}
|
|
if (tableExists(db, 'memories')) {
|
|
ensureColumn(db, 'memories', 'anchors', 'TEXT');
|
|
ensureColumn(db, 'memories', 'deleted_at', 'TEXT');
|
|
ensureColumn(db, 'memories', 'deleted_reason', 'TEXT');
|
|
}
|
|
}
|
|
|
|
function migrateDb(db) {
|
|
migrateExistingColumns(db);
|
|
}
|
|
|
|
function rebuildMemoryFts(db) {
|
|
db.exec("INSERT INTO memories_fts(memories_fts) VALUES('rebuild')");
|
|
}
|
|
|
|
|
|
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 };
|