fix(db): processes opening an index that needs a migration all open it (#3058)

Issue #2980: opening an index whose schema is behind read its version,
then ran each pending migration in a deferred transaction. When several
processes opened it at the same moment, those that read the version
just before another one migrated applied the same migration again, and
their open failed: on this Windows machine with "UNIQUE constraint
failed: schema_versions.version", since the version was already
recorded, and in the report with SQLITE_BUSY, which a deferred
transaction that has already read gets at once when it comes to write.

Each migration now runs in a transaction that takes the write lock up
front (BEGIN IMMEDIATE, a new option of the adapter's transaction()),
so a process that finds another one migrating waits it out on
busy_timeout, and reads the version again inside it, skipping a
migration already applied. The version the caller read stays a floor,
as before: a database built from schema.sql records only v1 and the
current version, and tests replay migrations above a version they
rewound to. An open of a current index takes no lock and waits for
nothing, as before.

Validation: __tests__/migration-concurrent-opens.test.ts. A second
process holds the write lock with version 16 recorded but uncommitted,
so the open reads 15: main fails with the UNIQUE error after waiting
1.1 s; the fix waits, opens and leaves the migration to the other
process. Three processes released at once from a stdin barrier on an
index with 150k unresolved references: main fails in round 0, the fix
opens all three in both rounds. On excalidraw's real index, 8 rounds of
three openers: main 7 rounds with failures, the fix 0. The single
process and the current-index cases pass on both.

Fixes #2980

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

CHANGELOG.md

@@ -90,6 +90,7 @@ and adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 - `codegraph status` no longer says the index is up to date while indexed files are missing their symbols: it now names files the parser couldn't read and files stored without their symbols (which `codegraph sync` repairs), `status --json` counts both, and `codegraph files --json` lists each file's recorded errors. Thanks @lipchey for the report. (#2336)
 - Running `codegraph status` while `codegraph index` rebuilds a project no longer makes the rebuild fail with "database is locked": status now only reads the index, and it says when a rebuild is running in another process, in its output and as `rebuildInProgress` in `status --json`. Thanks @sealad886 for the report, and @danusha2345 and @KissFeng for the fixes. (#2913)
 - Running `codegraph query`, `explore`, `node`, `callers`, `callees`, `impact`, `context`, `files` or `affected` while `codegraph index` rebuilds a project no longer makes the rebuild fail with "database is locked", and neither does the context hook Claude Code runs on its own when you ask about your code: like `codegraph status`, they now only read the index and say when a rebuild is running, and the hook stays quiet until the rebuild is done. They still answer from an index an older CodeGraph version built. Thanks @sealad886 for the report. (#2913)
+- Opening a project whose index needs a schema update no longer fails when several CodeGraph processes open it at the same moment, like agent sessions starting right after an upgrade, or the MCP server and a git hook: one of them updates the index while the others wait for it and then carry on. Before, the ones that lost the race applied the same update again, and failed with "UNIQUE constraint failed" or "database is locked". Thanks @inth3shadows for the report and @johnhldiver-beep for the first fix. (#2980)
 - A qualified name given to `codegraph callers`, `callees` or `impact`, or to `codegraph_node`, like `Format.Price`, now names only the symbol whose scope is spelled that way, so its answer no longer merges in `LegacyFormat.Price` or any other class whose name only ends with `Format`, nor a class that just sits in a file named `Format.cs`. In C#, `BasketService.AddItemToBasket` no longer also selects the `IBasketService` interface's method; ask for `IBasketService.AddItemToBasket` for that one. A file or folder still names the module in Rust, Python and Go, and a PHP namespace can be written with its own backslashes, like `Billing\Format::price`. No re-index needed. Thanks @Gobakabana for the report and @yang0228. (#2908)
 - `codegraph affected --filter` now matches its glob against whole names: `tests/**/*.py` also finds the tests directly inside `tests/`, including a changed test file itself, `tests/*.py` no longer counts a changed `tests/x.py.bak` as a test, and `e2e/*.spec.ts` no longer matches a folder that only ends in `e2e`, like `xe2e/`. A pattern with a folder in it still matches that folder at any depth, so `e2e/*.spec.ts` keeps finding `apps/web/e2e/login.spec.ts`; start it with `/` to match from the project root only. Thanks @Jseca666.
 - `codegraph init` and `codegraph index` no longer report a file you deleted as one that "could not be read" while git still lists it, which it does until the deletion is staged: such a file is now skipped, and so is a symlink that points at nothing. A file that is there but can't be read is still reported. Thanks @xianzhi-xzw for the report and the first fix. (#2904)

__tests__/migration-concurrent-opens.test.ts

@@ -0,0 +1,202 @@
+/**
+ * Opening an index that still needs a schema migration, while another process
+ * opens it too, must not fail (#2980).
+ *
+ * The open read the schema version, then ran each pending migration in its own
+ * deferred transaction. A process that read the version just before another
+ * one migrated went on to apply the same migration again: it either failed
+ * recording a version that was already recorded ("UNIQUE constraint failed:
+ * schema_versions.version"), or, for a migration that reads before it writes,
+ * hit SQLITE_BUSY at once, since SQLite does not wait on a lock for a
+ * transaction that has already read. Either way the open failed.
+ */
+import { describe, it, expect, beforeEach, afterEach } from 'vitest';
+import { spawn, type ChildProcess } from 'child_process';
+import * as fs from 'fs';
+import * as os from 'os';
+import * as path from 'path';
+import CodeGraph from '../src';
+import { DatabaseConnection } from '../src/db';
+import { createDatabase } from '../src/db/sqlite-adapter';
+import { CURRENT_SCHEMA_VERSION, getCurrentVersion } from '../src/db/migrations';
+
+const DIST_INDEX = path.resolve(__dirname, '../dist/index.js');
+/** The index migration 16 adds: the one migration the fixture leaves pending. */
+const V16_INDEX = 'idx_unresolved_failed_module_member';
+
+/** Opens the project once the test writes a byte to its stdin, and says how it went. */
+const OPENER = `
+const fs = require('fs');
+const CodeGraph = require(process.argv[1]).default;
+process.stdout.write('READY\\n');
+try { fs.readSync(0, Buffer.alloc(1), 0, 1, null); } catch (_) {}
+try {
+  CodeGraph.openSync(process.argv[2]).close();
+  process.stdout.write('OPENED\\n');
+} catch (err) {
+  process.stdout.write('FAILED ' + String((err && err.message) || err).replace(/\\s+/g, ' ') + '\\n');
+}
+`;
+
+/**
+ * Another process in the middle of migrating: it holds the write lock with
+ * version 16 recorded but not committed, and commits on its own after a while.
+ * Until then any other connection still reads version 15.
+ */
+const MIGRATING = `
+const { DatabaseSync } = require('node:sqlite');
+const db = new DatabaseSync(process.argv[1]);
+db.exec('PRAGMA busy_timeout = 5000');
+db.exec('BEGIN IMMEDIATE');
+db.prepare("INSERT INTO schema_versions (version, applied_at, description) VALUES (16, ?, 'another process')").run(Date.now());
+process.stdout.write('HOLDING\\n');
+setTimeout(() => { db.exec('COMMIT'); db.close(); process.stdout.write('COMMITTED\\n'); }, Number(process.argv[2]));
+`;
+
+/** A writer holding the lock on a current index, until the test ends it. */
+const WRITING = `
+const fs = require('fs');
+const { DatabaseSync } = require('node:sqlite');
+const db = new DatabaseSync(process.argv[1]);
+db.exec('BEGIN IMMEDIATE');
+db.exec("INSERT INTO project_metadata (key, value, updated_at) VALUES ('cg2980', '1', 0) ON CONFLICT(key) DO UPDATE SET value = excluded.value");
+process.stdout.write('HOLDING\\n');
+try { fs.readSync(0, Buffer.alloc(1), 0, 1, null); } catch (_) {}
+db.exec('ROLLBACK');
+db.close();
+`;
+
+interface Child {
+  child: ChildProcess;
+  output: () => string;
+  exited: Promise<number | null>;
+}
+
+describe('opening an index with a pending migration from several processes (#2980)', () => {
+  let dir: string;
+  let dbPath: string;
+  let children: Child[] = [];
+
+  beforeEach(() => {
+    dir = fs.mkdtempSync(path.join(os.tmpdir(), 'cg-migrate-'));
+    CodeGraph.initSync(dir).close();
+    dbPath = path.join(dir, '.codegraph', 'codegraph.db');
+  });
+
+  afterEach(async () => {
+    for (const { child, exited } of children) {
+      if (child.exitCode === null && child.signalCode === null) child.kill();
+      await exited;
+    }
+    children = [];
+    fs.rmSync(dir, { recursive: true, force: true });
+  });
+
+  /** Take the index back to version 15, the way an index built before migration 16 is. */
+  function pendingV16(refs = 0): void {
+    const { db } = createDatabase(dbPath);
+    try {
+      db.exec('PRAGMA foreign_keys = OFF');
+      db.exec(`DROP INDEX IF EXISTS ${V16_INDEX};
+        DELETE FROM schema_versions WHERE version >= 16;
+        INSERT OR IGNORE INTO schema_versions (version, applied_at, description) VALUES (15, 0, 'fixture');`);
+      if (refs > 0) {
+        // Enough rows that building the index takes a while, so openers overlap.
+        db.exec(`WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < ${refs})
+          INSERT INTO unresolved_refs (from_node_id, reference_name, reference_kind, line, col, file_path, language, status, name_tail)
+          SELECT 'n' || (i % 500), 'store.get' || i, 'calls', i % 900, 1, 'src/f' || (i % 300) || '.ts', 'typescript',
+            CASE WHEN i % 3 = 0 THEN 'failed' ELSE 'pending' END,
+            CASE WHEN i % 3 = 0 THEN 'module:./m' || (i % 50) ELSE 'get' || i END
+          FROM n`);
+      }
+    } finally {
+      db.close();
+    }
+  }
+
+  function state(): { version: number; v16Records: number; index: boolean } {
+    const { db } = createDatabase(dbPath, { readOnly: true });
+    try {
+      return {
+        version: getCurrentVersion(db),
+        v16Records: (db.prepare('SELECT COUNT(*) AS n FROM schema_versions WHERE version = 16').get() as { n: number }).n,
+        index: db.prepare("SELECT 1 FROM sqlite_master WHERE type = 'index' AND name = ?").get(V16_INDEX) !== undefined,
+      };
+    } finally {
+      db.close();
+    }
+  }
+
+  function run(args: string[]): Child {
+    const child = spawn(process.execPath, args, {
+      stdio: ['pipe', 'pipe', 'ignore'],
+      // Off the fixture, so a child still exiting cannot hold it (Windows EBUSY).
+      cwd: os.tmpdir(),
+      windowsHide: true,
+    });
+    let out = '';
+    child.stdout!.on('data', (chunk) => { out += String(chunk); });
+    const exited = new Promise<number | null>((resolve) => child.once('exit', (code) => resolve(code)));
+    const handle = { child, output: () => out, exited };
+    children.push(handle);
+    return handle;
+  }
+
+  async function says(who: Child, word: string): Promise<void> {
+    const deadline = Date.now() + 30_000;
+    while (!who.output().includes(word)) {
+      if (who.child.exitCode !== null || Date.now() > deadline) {
+        throw new Error(`expected "${word}", got: ${who.output()}`);
+      }
+      await new Promise((resolve) => setTimeout(resolve, 10));
+    }
+  }
+
+  it('waits for a process already migrating, then skips the migration it applied', async () => {
+    pendingV16();
+    const migrating = run(['-e', MIGRATING, dbPath, '1500']);
+    await says(migrating, 'HOLDING');
+
+    // Version 16 is not committed yet: this open reads 15, so it has a migration to run.
+    const started = Date.now();
+    const conn = DatabaseConnection.open(dbPath);
+    const waited = Date.now() - started;
+    try {
+      expect(getCurrentVersion(conn.getDb())).toBe(CURRENT_SCHEMA_VERSION);
+    } finally {
+      conn.close();
+    }
+    await says(migrating, 'COMMITTED');
+    expect(waited).toBeGreaterThanOrEqual(500);
+    // Recorded once, by the other process, and not run again here: its index was never built.
+    expect(state()).toEqual({ version: CURRENT_SCHEMA_VERSION, v16Records: 1, index: false });
+  }, 60_000);
+
+  it('opens in every one of three processes that start at the same moment', async () => {
+    for (let round = 0; round < 2; round++) {
+      pendingV16(round === 0 ? 150_000 : 0);
+      const openers = [0, 1, 2].map(() => run(['-e', OPENER, DIST_INDEX, dir]));
+      await Promise.all(openers.map((opener) => says(opener, 'READY')));
+      for (const opener of openers) opener.child.stdin!.write('g');
+      await Promise.all(openers.map((opener) => opener.exited));
+      for (const opener of openers) expect(opener.output(), `round ${round}`).toContain('OPENED');
+      expect(state()).toEqual({ version: CURRENT_SCHEMA_VERSION, v16Records: 1, index: true });
+    }
+  }, 120_000);
+
+  it('still applies a pending migration in a single process', () => {
+    pendingV16();
+    DatabaseConnection.open(dbPath).close();
+    expect(state()).toEqual({ version: CURRENT_SCHEMA_VERSION, v16Records: 1, index: true });
+  });
+
+  it('opens a current index without waiting for another process that is writing', async () => {
+    const writing = run(['-e', WRITING, dbPath]);
+    await says(writing, 'HOLDING');
+    const started = Date.now();
+    DatabaseConnection.open(dbPath).close();
+    expect(Date.now() - started).toBeLessThan(2_000);
+    writing.child.stdin!.write('g');
+    expect(await writing.exited).toBe(0);
+  }, 60_000);
+});

src/db/migrations.ts

@@ -343,6 +343,16 @@ function recordMigration(db: SqliteDatabase, version: number, description: strin
 
 /**
  * Run all pending migrations
+ *
+ * Another process may be opening the same index right now and running them
+ * too (#2980), so `fromVersion`, read before, can be out of date by the time a
+ * migration runs. Each one runs in a transaction that takes the write lock up
+ * front, waiting out a process already migrating, and reads the version again
+ * there, skipping a migration that process has applied. Before, the second
+ * process applied it again and its open failed: recording the version broke
+ * the version's uniqueness, or a migration that reads first got SQLITE_BUSY.
+ * `fromVersion` stays a floor: a database built from schema.sql records only
+ * v1 and the current version, and a rewound one replays what is above it.
  */
 export function runMigrations(db: SqliteDatabase, fromVersion: number): void {
   const pending = migrations.filter((m) => m.version > fromVersion);
@@ -357,9 +367,10 @@ export function runMigrations(db: SqliteDatabase, fromVersion: number): void {
   // Run each migration in a transaction
   for (const migration of pending) {
     db.transaction(() => {
+      if (Math.max(fromVersion, getCurrentVersion(db)) >= migration.version) return;
       migration.up(db);
       recordMigration(db, migration.version, migration.description);
-    })();
+    }, { immediate: true })();
   }
 }
 

src/db/sqlite-adapter.ts

@@ -27,11 +27,22 @@ export interface SqliteStatement {
   iterate(...params: any[]): IterableIterator<any>;
 }
 
+export interface TransactionOptions {
+  /**
+   * Begin with `BEGIN IMMEDIATE`, taking the write lock up front, instead of a
+   * deferred `BEGIN`. A connection that finds the lock taken then waits it out
+   * (busy_timeout); a deferred transaction that has already read gets
+   * SQLITE_BUSY at once when it comes to write, since SQLite will not wait for
+   * a lock there. No effect when nested in an open transaction.
+   */
+  immediate?: boolean;
+}
+
 export interface SqliteDatabase {
   prepare(sql: string): SqliteStatement;
   exec(sql: string): void;
   pragma(str: string, options?: { simple?: boolean }): any;
-  transaction<T>(fn: (...args: any[]) => T): (...args: any[]) => T;
+  transaction<T>(fn: (...args: any[]) => T, options?: TransactionOptions): (...args: any[]) => T;
   close(): void;
   readonly open: boolean;
   /** Undefined on a runtime without `isTransaction`; callers must then not memoize. */
@@ -153,7 +164,7 @@ class NodeSqliteAdapter implements SqliteDatabase {
     return row;
   }
 
-  transaction<T>(fn: (...args: any[]) => T): (...args: any[]) => T {
+  transaction<T>(fn: (...args: any[]) => T, options?: TransactionOptions): (...args: any[]) => T {
     return (...args: any[]) => {
       // Nested call (a transaction()-wrapped helper invoked from inside another
       // transaction): run the body directly inside the enclosing transaction.
@@ -168,7 +179,7 @@ class NodeSqliteAdapter implements SqliteDatabase {
           this._txDepth--;
         }
       }
-      this._db.exec('BEGIN');
+      this._db.exec(options?.immediate ? 'BEGIN IMMEDIATE' : 'BEGIN');
       this._txDepth = 1;
       try {
         const result = fn(...args);