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>
348 lines
9.3 KiB
JavaScript
348 lines
9.3 KiB
JavaScript
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';
|
|
|
|
const require = createRequire(import.meta.url);
|
|
import { createIndexerService } from '../app/src/main/indexer-service.ts';
|
|
|
|
function manualTimers() {
|
|
const timers = new Set();
|
|
return {
|
|
setTimeout(fn) {
|
|
timers.add(fn);
|
|
return fn;
|
|
},
|
|
clearTimeout(fn) {
|
|
timers.delete(fn);
|
|
},
|
|
flush() {
|
|
const pending = [...timers];
|
|
timers.clear();
|
|
for (const fn of pending) fn();
|
|
},
|
|
};
|
|
}
|
|
|
|
test('indexer service debounces repeated build requests', async () => {
|
|
const timers = manualTimers();
|
|
const calls = [];
|
|
const service = createIndexerService({
|
|
buildIndex: async ({ reason }) => calls.push(reason),
|
|
watchProjects: () => null,
|
|
writeHeartbeat: () => {},
|
|
timers,
|
|
stabilityMs: 0,
|
|
});
|
|
|
|
service.scheduleBuild('first');
|
|
service.scheduleBuild('second');
|
|
service.scheduleBuild('third');
|
|
timers.flush();
|
|
await service.idle();
|
|
|
|
assert.deepEqual(calls, ['third']);
|
|
});
|
|
|
|
test('indexer service runs one pending build after an in-flight build finishes', async () => {
|
|
const timers = manualTimers();
|
|
const calls = [];
|
|
let finishFirst;
|
|
const service = createIndexerService({
|
|
buildIndex: async ({ reason }) => {
|
|
calls.push(reason);
|
|
if (reason === 'first') await new Promise(resolve => { finishFirst = resolve; });
|
|
},
|
|
watchProjects: () => null,
|
|
writeHeartbeat: () => {},
|
|
timers,
|
|
stabilityMs: 0,
|
|
});
|
|
|
|
const first = service.runBuildNow('first');
|
|
service.scheduleBuild('second');
|
|
timers.flush();
|
|
|
|
assert.deepEqual(calls, ['first']);
|
|
finishFirst();
|
|
await first;
|
|
await service.idle();
|
|
|
|
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 = [];
|
|
let rejectBuild;
|
|
const service = createIndexerService({
|
|
buildIndex: () => new Promise((_resolve, reject) => { rejectBuild = reject; }),
|
|
watchProjects: () => null,
|
|
writeHeartbeat: () => {},
|
|
timers,
|
|
stabilityMs: 0,
|
|
logger: { warn: (msg) => warnings.push(msg) },
|
|
});
|
|
|
|
const build = service.runBuildNow('startup');
|
|
service.stop(); // manual rebuild path tears the worker down mid-build
|
|
rejectBuild(new Error('Indexer worker stopped'));
|
|
await build;
|
|
|
|
assert.deepEqual(warnings, []);
|
|
});
|
|
|
|
test('indexer service logs a build that fails while running', async () => {
|
|
const timers = manualTimers();
|
|
const warnings = [];
|
|
let rejectBuild;
|
|
const service = createIndexerService({
|
|
buildIndex: () => new Promise((_resolve, reject) => { rejectBuild = reject; }),
|
|
watchProjects: () => null,
|
|
writeHeartbeat: () => {},
|
|
timers,
|
|
stabilityMs: 0,
|
|
logger: { warn: (msg) => warnings.push(msg) },
|
|
});
|
|
|
|
const build = service.runBuildNow('watch');
|
|
rejectBuild(new Error('disk on fire'));
|
|
await build;
|
|
|
|
assert.equal(warnings.length, 1);
|
|
assert.match(warnings[0], /Obelisk index build failed: disk on fire/);
|
|
});
|
|
|
|
test('indexer service waits for a stability window before building', async () => {
|
|
const timers = manualTimers();
|
|
const calls = [];
|
|
const service = createIndexerService({
|
|
buildIndex: async ({ reason }) => calls.push(reason),
|
|
watchProjects: () => null,
|
|
writeHeartbeat: () => {},
|
|
timers,
|
|
stabilityMs: 500,
|
|
});
|
|
|
|
service.scheduleBuild('jsonl-change');
|
|
timers.flush();
|
|
await service.idle();
|
|
assert.deepEqual(calls, []);
|
|
|
|
timers.flush();
|
|
await service.idle();
|
|
assert.deepEqual(calls, ['jsonl-change']);
|
|
});
|
|
|
|
test('indexer service retries watcher setup when the projects directory is missing', () => {
|
|
const timers = manualTimers();
|
|
let attempts = 0;
|
|
const service = createIndexerService({
|
|
buildIndex: async () => {},
|
|
watchProjects: () => {
|
|
attempts++;
|
|
return attempts === 1 ? null : { close() {} };
|
|
},
|
|
writeHeartbeat: () => {},
|
|
timers,
|
|
stabilityMs: 0,
|
|
});
|
|
|
|
service.start({ buildOnStart: false });
|
|
assert.equal(attempts, 1);
|
|
|
|
timers.flush();
|
|
assert.equal(attempts, 2);
|
|
|
|
timers.flush();
|
|
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();
|
|
const calls = [];
|
|
let watchArgs = null;
|
|
const handlers = {};
|
|
const watcher = {
|
|
on(event, handler) {
|
|
handlers[event] = handler;
|
|
return watcher;
|
|
},
|
|
closeCalled: false,
|
|
close() {
|
|
watcher.closeCalled = true;
|
|
},
|
|
};
|
|
const chokidar = {
|
|
watch(paths, options) {
|
|
watchArgs = { paths, options };
|
|
return watcher;
|
|
},
|
|
};
|
|
|
|
const service = createIndexerService({
|
|
projectsDir,
|
|
buildIndex: async ({ reason }) => calls.push(reason),
|
|
chokidar,
|
|
writeHeartbeat: () => {},
|
|
timers,
|
|
stabilityMs: 0,
|
|
debounceMs: 0,
|
|
});
|
|
|
|
try {
|
|
service.start({ buildOnStart: false });
|
|
assert.equal(watchArgs.paths, projectsDir);
|
|
assert.equal(watchArgs.options.cwd, projectsDir);
|
|
assert.equal(watchArgs.options.ignoreInitial, true);
|
|
assert.ok(watchArgs.options.awaitWriteFinish);
|
|
|
|
handlers.change('session.jsonl');
|
|
timers.flush();
|
|
await service.idle();
|
|
assert.deepEqual(calls, ['watch']);
|
|
} finally {
|
|
service.stop();
|
|
}
|
|
|
|
assert.equal(watcher.closeCalled, true);
|
|
});
|
|
|
|
test('indexer service passes changed JSONL paths to the build worker', async () => {
|
|
const projectsDir = mkdtempSync(join(tmpdir(), 'obelisk-changed-paths-'));
|
|
const timers = manualTimers();
|
|
const calls = [];
|
|
const handlers = {};
|
|
const watcher = {
|
|
on(event, handler) {
|
|
handlers[event] = handler;
|
|
return watcher;
|
|
},
|
|
close() {},
|
|
};
|
|
const chokidar = {
|
|
watch() {
|
|
return watcher;
|
|
},
|
|
};
|
|
|
|
const service = createIndexerService({
|
|
projectsDir,
|
|
buildIndex: async (args) => calls.push(args),
|
|
chokidar,
|
|
writeHeartbeat: () => {},
|
|
timers,
|
|
stabilityMs: 0,
|
|
debounceMs: 0,
|
|
});
|
|
|
|
service.start({ buildOnStart: false });
|
|
handlers.change('project-a/session-1.jsonl');
|
|
handlers.add('project-a/session-2.json');
|
|
timers.flush();
|
|
await service.idle();
|
|
|
|
assert.equal(calls.length, 1);
|
|
assert.equal(calls[0].reason, 'watch');
|
|
assert.deepEqual(calls[0].changedPaths, [
|
|
'project-a/session-1.jsonl',
|
|
'project-a/session-2.json',
|
|
]);
|
|
});
|
|
|
|
test('indexer service watches Claude projects and Codex sessions for app-side indexing', async () => {
|
|
const claudeProjectsDir = mkdtempSync(join(tmpdir(), 'obelisk-watch-claude-'));
|
|
const codexSessionsDir = mkdtempSync(join(tmpdir(), 'obelisk-watch-codex-sessions-'));
|
|
const timers = manualTimers();
|
|
const calls = [];
|
|
const watchers = [];
|
|
const watchArgs = [];
|
|
const chokidar = {
|
|
watch(paths, options) {
|
|
const handlers = {};
|
|
const watcher = {
|
|
handlers,
|
|
on(event, handler) {
|
|
handlers[event] = handler;
|
|
return watcher;
|
|
},
|
|
close() {},
|
|
};
|
|
watchers.push(watcher);
|
|
watchArgs.push({ paths, options });
|
|
return watcher;
|
|
},
|
|
};
|
|
|
|
const service = createIndexerService({
|
|
projectsDir: claudeProjectsDir,
|
|
watchDirs: [claudeProjectsDir, codexSessionsDir],
|
|
buildIndex: async (args) => calls.push(args),
|
|
chokidar,
|
|
writeHeartbeat: () => {},
|
|
timers,
|
|
stabilityMs: 0,
|
|
debounceMs: 0,
|
|
});
|
|
|
|
service.start({ buildOnStart: false });
|
|
assert.deepEqual(watchArgs.map(arg => arg.paths), [claudeProjectsDir, codexSessionsDir]);
|
|
assert.deepEqual(watchArgs.map(arg => arg.options.cwd), [claudeProjectsDir, codexSessionsDir]);
|
|
|
|
watchers[1].handlers.change('2026/06/15/rollout-2026-06-15T00-00-00-codex.jsonl');
|
|
timers.flush();
|
|
await service.idle();
|
|
|
|
assert.equal(calls.length, 1);
|
|
assert.deepEqual(calls[0].changedPaths, [
|
|
'2026/06/15/rollout-2026-06-15T00-00-00-codex.jsonl',
|
|
]);
|
|
});
|