fix(journal): serialize concurrent appends via lockfile

src/domain/graph/journal.ts

@@ -4,6 +4,106 @@ import { debug, warn } from '../../infrastructure/logger.js';
 
 export const JOURNAL_FILENAME = 'changes.journal';
 const HEADER_PREFIX = '# codegraph-journal v1 ';
+const LOCK_SUFFIX = '.lock';
+const LOCK_TIMEOUT_MS = 5_000;
+const LOCK_STALE_MS = 30_000;
+const LOCK_RETRY_MS = 25;
+
+function sleepSync(ms: number): void {
+  const buf = new Int32Array(new SharedArrayBuffer(4));
+  Atomics.wait(buf, 0, 0, ms);
+}
+
+function isPidAlive(pid: number): boolean {
+  if (!Number.isFinite(pid) || pid <= 0) return false;
+  try {
+    process.kill(pid, 0);
+    return true;
+  } catch (e) {
+    // EPERM means the process exists but we lack permission — still alive.
+    return (e as NodeJS.ErrnoException).code === 'EPERM';
+  }
+}
+
+function acquireJournalLock(lockPath: string): number {
+  const start = Date.now();
+  for (;;) {
+    try {
+      const fd = fs.openSync(lockPath, 'wx');
+      try {
+        fs.writeSync(fd, `${process.pid}\n`);
+      } catch {
+        /* PID stamp is advisory; fd is still exclusive */
+      }
+      return fd;
+    } catch (e) {
+      if ((e as NodeJS.ErrnoException).code !== 'EEXIST') throw e;
+    }
+
+    let holderAlive = true;
+    try {
+      const pidContent = fs.readFileSync(lockPath, 'utf-8').trim();
+      holderAlive = isPidAlive(Number(pidContent));
+    } catch {
+      /* unreadable — fall through to age check */
+    }
+
+    if (!holderAlive) {
+      try {
+        fs.unlinkSync(lockPath);
+      } catch {
+        /* another writer stole it first */
+      }
+      continue;
+    }
+
+    try {
+      const stat = fs.statSync(lockPath);
+      if (Date.now() - stat.mtimeMs > LOCK_STALE_MS) {
+        try {
+          fs.unlinkSync(lockPath);
+        } catch {
+          /* raced */
+        }
+        continue;
+      }
+    } catch {
+      /* stat failed — keep retrying */
+    }
+
+    if (Date.now() - start > LOCK_TIMEOUT_MS) {
+      throw new Error(`Failed to acquire journal lock at ${lockPath} within ${LOCK_TIMEOUT_MS}ms`);
+    }
+    sleepSync(LOCK_RETRY_MS);
+  }
+}
+
+function releaseJournalLock(lockPath: string, fd: number): void {
+  try {
+    fs.closeSync(fd);
+  } catch {
+    /* ignore */
+  }
+  try {
+    fs.unlinkSync(lockPath);
+  } catch {
+    /* ignore */
+  }
+}
+
+function withJournalLock<T>(rootDir: string, fn: () => T): T {
+  const dir = path.join(rootDir, '.codegraph');
+  if (!fs.existsSync(dir)) {
+    fs.mkdirSync(dir, { recursive: true });
+  }
+  const lockPath = path.join(dir, `${JOURNAL_FILENAME}${LOCK_SUFFIX}`);
+  const fd = acquireJournalLock(lockPath);
+  try {
+    return fn();
+  } finally {
+    releaseJournalLock(lockPath, fd);
+  }
+}
 
 interface JournalResult {
   valid: boolean;
@@ -63,43 +163,37 @@ export function appendJournalEntries(
   rootDir: string,
   entries: Array<{ file: string; deleted?: boolean }>,
 ): void {
-  const dir = path.join(rootDir, '.codegraph');
-  const journalPath = path.join(dir, JOURNAL_FILENAME);
+  withJournalLock(rootDir, () => {
+    const journalPath = path.join(rootDir, '.codegraph', JOURNAL_FILENAME);
 
-  if (!fs.existsSync(dir)) {
-    fs.mkdirSync(dir, { recursive: true });
-  }
+    if (!fs.existsSync(journalPath)) {
+      fs.writeFileSync(journalPath, `${HEADER_PREFIX}0\n`);
+    }
 
-  if (!fs.existsSync(journalPath)) {
-    fs.writeFileSync(journalPath, `${HEADER_PREFIX}0\n`);
-  }
+    const lines = entries.map((e) => {
+      if (e.deleted) return `DELETED ${e.file}`;
+      return e.file;
+    });
 
-  const lines = entries.map((e) => {
-    if (e.deleted) return `DELETED ${e.file}`;
-    return e.file;
+    fs.appendFileSync(journalPath, `${lines.join('\n')}\n`);
   });
-
-  fs.appendFileSync(journalPath, `${lines.join('\n')}\n`);
 }
 
 export function writeJournalHeader(rootDir: string, timestamp: number): void {
-  const dir = path.join(rootDir, '.codegraph');
-  const journalPath = path.join(dir, JOURNAL_FILENAME);
-  const tmpPath = `${journalPath}.tmp`;
-
-  if (!fs.existsSync(dir)) {
-    fs.mkdirSync(dir, { recursive: true });
-  }
+  withJournalLock(rootDir, () => {
+    const journalPath = path.join(rootDir, '.codegraph', JOURNAL_FILENAME);
+    const tmpPath = `${journalPath}.tmp`;
 
-  try {
-    fs.writeFileSync(tmpPath, `${HEADER_PREFIX}${timestamp}\n`);
-    fs.renameSync(tmpPath, journalPath);
-  } catch (err) {
-    warn(`Failed to write journal header: ${(err as Error).message}`);
     try {
-      fs.unlinkSync(tmpPath);
-    } catch {
-      /* ignore */
+      fs.writeFileSync(tmpPath, `${HEADER_PREFIX}${timestamp}\n`);
+      fs.renameSync(tmpPath, journalPath);
+    } catch (err) {
+      warn(`Failed to write journal header: ${(err as Error).message}`);
+      try {
+        fs.unlinkSync(tmpPath);
+      } catch {
+        /* ignore */
+      }
     }
-  }
+  });
 }

tests/unit/journal.test.ts

@@ -196,6 +196,53 @@ describe('appendJournalEntries', () => {
   });
 });
 
+describe('concurrent-append safety', () => {
+  it('cleans up the .lock file after a successful append', () => {
+    const root = makeRoot();
+    writeJournalHeader(root, 1700000000000);
+    appendJournalEntries(root, [{ file: 'src/a.js' }]);
+
+    const lockPath = path.join(root, '.codegraph', `${JOURNAL_FILENAME}.lock`);
+    expect(fs.existsSync(lockPath)).toBe(false);
+  });
+
+  it('steals a stale lock whose holder PID is dead', () => {
+    const root = makeRoot();
+    writeJournalHeader(root, 1700000000000);
+
+    // Pre-stage a lockfile with a PID that is guaranteed not to exist
+    // (max 32-bit value; well above any real process).
+    const lockPath = path.join(root, '.codegraph', `${JOURNAL_FILENAME}.lock`);
+    fs.writeFileSync(lockPath, '2147483646\n');
+
+    expect(() => appendJournalEntries(root, [{ file: 'src/a.js' }])).not.toThrow();
+    expect(fs.existsSync(lockPath)).toBe(false);
+
+    const result = readJournal(root);
+    expect(result.changed).toEqual(['src/a.js']);
+  });
+
+  it('produces no interleaved lines under repeated appends', () => {
+    const root = makeRoot();
+    writeJournalHeader(root, 1700000000000);
+
+    // Many small appends — every emitted line must be a complete,
+    // well-formed entry (no truncated "DELETED " prefixes, no split paths).
+    for (let i = 0; i < 200; i++) {
+      appendJournalEntries(root, [
+        { file: `src/changed-${i}.js` },
+        { file: `src/gone-${i}.js`, deleted: true },
+      ]);
+    }
+
+    const content = fs.readFileSync(path.join(root, '.codegraph', JOURNAL_FILENAME), 'utf-8');
+    for (const line of content.split('\n')) {
+      if (!line || line.startsWith('#')) continue;
+      expect(line).toMatch(/^(DELETED src\/gone-\d+\.js|src\/changed-\d+\.js)$/);
+    }
+  });
+});
+
 describe('read/write/append lifecycle', () => {
   it('full lifecycle: header → append → read → new header', () => {
     const root = makeRoot();