fix: recover interrupted secondary indexes off-thread

CHANGELOG.md

@@ -145,6 +145,8 @@ and adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ### Fixes
 
+- Restarting after an interrupted index no longer gets trapped in repeated watchdog restarts while repairing the database. (#1887)
+
 - Rust calls on `self` now stay with the enclosing type instead of linking to an unrelated type’s same-named method. Thanks @L4XB. (#1861)
 
 - Turning telemetry off now resets its identity and stops running processes from recording, sending, or restoring unsent data. (#1869)

__tests__/concurrent-locking.test.ts

@@ -112,13 +112,13 @@ describe('issue #238 — ToolHandler reuses the default instance (#2)', () => {
     fs.rmSync(dir, { recursive: true, force: true });
   });
 
-  it('getCodeGraph(defaultRoot) returns the default instance, not a new connection', () => {
-    const openSpy = vi.spyOn(CodeGraph, 'openSync');
+  it('getCodeGraph(defaultRoot) returns the default instance, not a new connection', async () => {
+    const openSpy = vi.spyOn(CodeGraph, 'open');
     try {
       // eslint-disable-next-line @typescript-eslint/no-explicit-any
-      const resolved = (handler as any).getCodeGraph(root);
+      const resolved = await (handler as any).getCodeGraph(root);
       // eslint-disable-next-line @typescript-eslint/no-explicit-any
-      const nested = (handler as any).getCodeGraph(path.join(root, 'does', 'not', 'exist'));
+      const nested = await (handler as any).getCodeGraph(path.join(root, 'does', 'not', 'exist'));
       expect(resolved).toBe(cg);
       expect(nested).toBe(cg); // a sub-path resolves up to the same default project
       expect(openSpy).not.toHaveBeenCalled(); // no second connection opened
@@ -128,7 +128,7 @@ describe('issue #238 — ToolHandler reuses the default instance (#2)', () => {
   });
 
   it('concurrent read tool calls (mixed projectPath) all succeed without "database is locked"', async () => {
-    const openSpy = vi.spyOn(CodeGraph, 'openSync');
+    const openSpy = vi.spyOn(CodeGraph, 'open');
     try {
       const calls: Promise<{ content: Array<{ text: string }>; isError?: boolean }>[] = [
         handler.execute('codegraph_search', { query: 'helper' }),

__tests__/db-reopen-on-replace.test.ts

@@ -14,7 +14,7 @@
  * repros are gated to non-Windows; `isReplacedOnDisk` is verified to stay false
  * on Windows.
  */
-import { describe, it, expect, beforeEach, afterEach } from 'vitest';
+import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest';
 import * as fs from 'fs';
 import * as os from 'os';
 import * as path from 'path';
@@ -114,4 +114,29 @@ describe('CodeGraph.reopenIfReplaced (issue #925)', () => {
     expect(server.reopenIfReplaced()).toBe(false);
     server.destroy();
   });
+
+  it('does not publish a replacement connection after close wins the race', async () => {
+    const server = CodeGraph.initSync(root);
+    const stale = (server as unknown as { db: DatabaseConnection }).db;
+    const fresh = DatabaseConnection.initialize(path.join(root, 'replacement.db'));
+    let finishOpen!: (connection: DatabaseConnection) => void;
+    const openGate = new Promise<DatabaseConnection>((resolve) => { finishOpen = resolve; });
+    const replacedSpy = vi.spyOn(stale, 'isReplacedOnDisk').mockReturnValue(true);
+    const openSpy = vi.spyOn(DatabaseConnection, 'openAsync').mockReturnValue(openGate);
+
+    try {
+      const reopening = server.reopenIfReplacedAsync();
+      server.close();
+      finishOpen(fresh);
+
+      expect(await reopening).toBe(false);
+      expect(fresh.isOpen()).toBe(false);
+      expect(openSpy).toHaveBeenCalledOnce();
+    } finally {
+      replacedSpy.mockRestore();
+      openSpy.mockRestore();
+      try { server.close(); } catch { /* may already be closed */ }
+      try { fresh.close(); } catch { /* recovery closes it on the guarded path */ }
+    }
+  });
 });

__tests__/foundation.test.ts

@@ -138,6 +138,24 @@ describe('CodeGraph Foundation', () => {
       expect(after).toEqual(before);
     });
 
+    it('restores every secondary index through the asynchronous open path (#1887)', async () => {
+      const dbPath = getDatabasePath(tempDir);
+      const first = DatabaseConnection.initialize(dbPath);
+      const before = (first.getDb()
+        .prepare("SELECT name FROM sqlite_master WHERE type = 'index' ORDER BY name")
+        .all() as Array<{ name: string }>).map((r) => r.name);
+      first.beginBulkParseLoad();
+      first.close();
+
+      const reopened = await DatabaseConnection.openAsync(dbPath);
+      const after = (reopened.getDb()
+        .prepare("SELECT name FROM sqlite_master WHERE type = 'index' ORDER BY name")
+        .all() as Array<{ name: string }>).map((r) => r.name);
+      reopened.close();
+
+      expect(after).toEqual(before);
+    });
+
     it('skips secondary-index DDL when the schema is already healthy', () => {
       const dbPath = getDatabasePath(tempDir);
       const connection = DatabaseConnection.initialize(dbPath);

__tests__/liveness-watchdog.test.ts

@@ -2,6 +2,9 @@ import { describe, it, expect, beforeAll } from 'vitest';
 import { spawn } from 'child_process';
 import * as fs from 'fs';
 import * as path from 'path';
+import * as os from 'os';
+import { CodeGraph } from '../src';
+import { DatabaseConnection, getDatabasePath } from '../src/db';
 import {
   parseWatchdogTimeoutMs,
   deriveCheckIntervalMs,
@@ -58,25 +61,35 @@ describe('liveness watchdog (spawned, real watchdog process)', () => {
     env: Record<string, string>,
     body: string,
     hardTimeoutMs: number,
-    progressPaths?: string[]
-  ): Promise<{ code: number | null; signal: NodeJS.Signals | 'TIMEOUT' | null }> {
+    progressPaths?: string[],
+    prelude = ''
+  ): Promise<{ code: number | null; signal: NodeJS.Signals | 'TIMEOUT' | null; stderr: string }> {
     const src = `
-      const { installMainThreadWatchdog } = require(${JSON.stringify(MODULE)});
-      installMainThreadWatchdog(${progressPaths ? JSON.stringify({ progressPaths }) : ''});
-      ${body}
+      (async () => {
+        ${prelude}
+        const { installMainThreadWatchdog } = require(${JSON.stringify(MODULE)});
+        installMainThreadWatchdog(${progressPaths ? JSON.stringify({ progressPaths }) : ''});
+        ${body}
+      })().catch((error) => {
+        console.error(error);
+        process.exit(23);
+      });
     `;
     const child = spawn(process.execPath, ['-e', src], {
       env: { ...process.env, ...env },
-      stdio: ['ignore', 'ignore', 'ignore'],
+      stdio: ['ignore', 'ignore', 'pipe'],
     });
+    let stderr = '';
+    child.stderr.setEncoding('utf8');
+    child.stderr.on('data', (chunk: string) => { stderr += chunk; });
     return new Promise((resolve) => {
       const timer = setTimeout(() => {
         child.kill('SIGKILL');
-        resolve({ code: null, signal: 'TIMEOUT' });
+        resolve({ code: null, signal: 'TIMEOUT', stderr });
       }, hardTimeoutMs);
       child.on('exit', (code, signal) => {
         clearTimeout(timer);
-        resolve({ code, signal });
+        resolve({ code, signal, stderr });
       });
     });
   }
@@ -183,6 +196,69 @@ describe('liveness watchdog (spawned, real watchdog process)', () => {
     expectKilled(r);
   }, 20000);
 
+  it('keeps interrupted secondary-index recovery off the watched event loop (#1887)', async () => {
+    const projectRoot = fs.mkdtempSync(path.join(os.tmpdir(), 'cg-index-recovery-'));
+    try {
+      const project = CodeGraph.initSync(projectRoot);
+      project.close();
+
+      const dbPath = getDatabasePath(projectRoot);
+      const seed = DatabaseConnection.open(dbPath);
+      const expectedIndexes = (seed.getDb()
+        .prepare("SELECT name FROM sqlite_master WHERE type = 'index' ORDER BY name")
+        .all() as Array<{ name: string }>).map((row) => row.name);
+
+      seed.beginBulkNodeLoad();
+      seed.beginBulkParseLoad();
+      const insert = seed.getDb().prepare(
+        `INSERT INTO nodes (
+          id, kind, name, qualified_name, file_path, language,
+          start_line, end_line, start_column, end_column, updated_at
+        ) VALUES (?, 'function', ?, ?, ?, 'typescript', 1, 1, 0, 1, ?)`
+      );
+      seed.getDb().exec('BEGIN');
+      try {
+        for (let i = 0; i < 600_000; i++) {
+          const name = `symbol_${i}`;
+          insert.run(`node_${i}`, name, name, `src/file_${i % 1000}.ts`, Date.now());
+        }
+        seed.getDb().exec('COMMIT');
+      } catch (error) {
+        seed.getDb().exec('ROLLBACK');
+        throw error;
+      }
+      seed.endBulkNodeLoad();
+      seed.close();
+
+      const DIST_INDEX = path.resolve(__dirname, '../dist/index.js');
+      const prelude = `
+        const { CodeGraph, initGrammars } = require(${JSON.stringify(DIST_INDEX)});
+        await initGrammars();
+      `;
+      const body = `
+        CodeGraph.open(${JSON.stringify(projectRoot)}).then((cg) => {
+          const names = cg.db.getDb()
+            .prepare("SELECT name FROM sqlite_master WHERE type = 'index' ORDER BY name")
+            .all().map((row) => row.name);
+          cg.close();
+          process.exit(JSON.stringify(names) === ${JSON.stringify(JSON.stringify(expectedIndexes))} ? 0 : 21);
+        }).catch(() => process.exit(22));
+      `;
+      const r = await runChild(
+        { CODEGRAPH_WATCHDOG_TIMEOUT_MS: '100' },
+        body,
+        20_000,
+        [dbPath, `${dbPath}-wal`],
+        prelude
+      );
+
+      expect(r.signal).toBeNull();
+      expect(r.code, r.stderr).toBe(0);
+    } finally {
+      fs.rmSync(projectRoot, { recursive: true, force: true, maxRetries: 10, retryDelay: 100 });
+    }
+  }, 30000);
+
   it('does NOT kill a wedged process when CODEGRAPH_NO_WATCHDOG=1', async () => {
     const { code, signal } = await runChild(
       { CODEGRAPH_WATCHDOG_TIMEOUT_MS: '500', CODEGRAPH_NO_WATCHDOG: '1' },

src/db/index.ts

@@ -163,6 +163,42 @@ export class DatabaseConnection {
    * Open an existing database
    */
   static open(dbPath: string): DatabaseConnection {
+    const conn = DatabaseConnection.openBase(dbPath);
+    try {
+      conn.healBulkSecondaryIndexes();
+
+      // Self-heal a killed session's leftover oversized WAL (#1431) — one
+      // statSync when healthy, off-thread checkpoint+truncate when not.
+      void conn.healOversizedWal();
+      return conn;
+    } catch (error) {
+      try { conn.close(); } catch { /* preserve the recovery error */ }
+      throw error;
+    }
+  }
+
+  /**
+   * Open an existing database without rebuilding dropped secondary indexes on
+   * the caller's event loop. A killed bulk-load window can leave those indexes
+   * absent, and rebuilding them on a large graph is long-running SQLite work.
+   */
+  static async openAsync(dbPath: string): Promise<DatabaseConnection> {
+    const conn = DatabaseConnection.openBase(dbPath);
+    try {
+      await conn.healBulkSecondaryIndexesOffThread();
+
+      // Self-heal a killed session's leftover oversized WAL (#1431) — one
+      // statSync when healthy, off-thread checkpoint+truncate when not.
+      void conn.healOversizedWal();
+      return conn;
+    } catch (error) {
+      try { conn.close(); } catch { /* preserve the recovery error */ }
+      throw error;
+    }
+  }
+
+  /** Open and perform the short, synchronous recovery shared by both paths. */
+  private static openBase(dbPath: string): DatabaseConnection {
     if (!fs.existsSync(dbPath)) {
       throw new Error(`Database not found: ${dbPath}`);
     }
@@ -191,12 +227,6 @@ export class DatabaseConnection {
     // beginBulkNodeLoad and endBulkNodeLoad): the FTS triggers are missing and
     // nodes_fts is stale. Rebuild + recreate so search stays in sync.
     conn.healBulkNodeLoad();
-    conn.healBulkSecondaryIndexes();
-
-    // Self-heal a killed session's leftover oversized WAL (#1431) — one
-    // statSync when healthy, off-thread checkpoint+truncate when not.
-    void conn.healOversizedWal();
-
     return conn;
   }
 
@@ -415,6 +445,20 @@ export class DatabaseConnection {
 
   /** Recreate every secondary index a killed bulk parse/ref/edge window may leave dropped. */
   private healBulkSecondaryIndexes(): void {
+    for (const ddl of this.bulkSecondaryIndexDdls()) {
+      this.db.exec(ddl);
+    }
+  }
+
+  /** Recreate dropped bulk-load indexes on a dedicated SQLite connection. */
+  private async healBulkSecondaryIndexesOffThread(): Promise<void> {
+    const ddls = this.bulkSecondaryIndexDdls();
+    if (ddls.length === 0) return;
+    await this.runSqlOffThread(ddls);
+  }
+
+  /** Return the canonical DDL needed to heal a killed bulk-load window. */
+  private bulkSecondaryIndexDdls(): string[] {
     const names = [...new Set<string>([
       ...DatabaseConnection.BULK_PARSE_INDEX_NAMES,
       ...DatabaseConnection.BULK_REF_INDEX_NAMES,
@@ -424,15 +468,15 @@ export class DatabaseConnection {
     const row = this.db
       .prepare(`SELECT count(*) AS c FROM sqlite_master WHERE type = 'index' AND name IN (${placeholders})`)
       .get(...names) as { c: number } | undefined;
-    if ((row?.c ?? 0) >= names.length) return;
+    if ((row?.c ?? 0) >= names.length) return [];
 
     const schemaPath = path.join(__dirname, 'schema.sql');
     const schema = fs.readFileSync(schemaPath, 'utf-8');
-    for (const idx of names) {
+    return names.map((idx) => {
       const m = schema.match(new RegExp(`CREATE INDEX IF NOT EXISTS ${idx}\\b[^;]*;`));
       if (!m) throw new Error(`schema.sql: index ${idx} not found for crash recovery`);
-      this.db.exec(m[0]);
-    }
+      return m[0];
+    });
   }
 
   /**
@@ -804,6 +848,58 @@ export class DatabaseConnection {
     }
   }
 
+  /**
+   * Run required SQL on a worker-owned connection. Unlike maintenance pragmas,
+   * crash recovery is not best-effort: the database must not be handed to a
+   * caller until every statement succeeds.
+   */
+  private async runSqlOffThread(statements: string[]): Promise<void> {
+    const { Worker } = await import('node:worker_threads');
+    const workerSource = `
+      const { workerData, parentPort } = require('node:worker_threads');
+      let db;
+      try {
+        const { DatabaseSync } = require('node:sqlite');
+        db = new DatabaseSync(workerData.dbPath);
+        db.exec('PRAGMA busy_timeout=30000');
+        for (const sql of workerData.statements) db.exec(sql);
+        db.close();
+        db = undefined;
+        parentPort.postMessage({ ok: true });
+      } catch (error) {
+        try { if (db) db.close(); } catch {}
+        parentPort.postMessage({
+          ok: false,
+          error: error && error.message ? error.message : String(error),
+        });
+      }
+    `;
+
+    await new Promise<void>((resolve, reject) => {
+      let settled = false;
+      const finish = (error?: Error): void => {
+        if (settled) return;
+        settled = true;
+        if (error) reject(error);
+        else resolve();
+      };
+
+      const worker = new Worker(workerSource, {
+        eval: true,
+        workerData: { dbPath: this.dbPath, statements },
+      });
+      worker.once('message', (message: { ok: boolean; error?: string }) => {
+        if (message.ok) finish();
+        else finish(new Error(`Secondary-index recovery failed: ${message.error ?? 'unknown worker error'}`));
+      });
+      worker.once('error', (error) => finish(error));
+      worker.once('exit', (code) => {
+        if (code !== 0) finish(new Error(`Secondary-index recovery worker exited with code ${code}`));
+        else finish(new Error('Secondary-index recovery worker exited before reporting completion'));
+      });
+    });
+  }
+
   /**
    * Close the database connection
    */