Eliminate rename() — reindex deletes old DB, index writes directly

Architecture change to prevent concurrent write corruption:

  Indexing (new project): cbm_write_db writes project.db directly
  Reindexing (existing): delete project.db, then index as new
  Incremental: upsert changes into existing project.db (unchanged)

Root cause: rename(project.db.tmp, project.db) while watcher/MCP
had the old file open via SQLite caused WAL corruption — stale
connections wrote to deleted inodes.

Changes:
- cbm_gbuf_dump_to_sqlite: write to final path, no .tmp + rename
- pipeline.c: if DB exists but not eligible for incremental, delete
  it before proceeding with full index (reindex = delete + index)
- Pipeline lock (atomic spinlock) serializes concurrent runs:
  MCP/autoindex block, watcher skips if busy
- delete_project acquires pipeline lock before unlinking
- BEGIN IMMEDIATE for all write transactions (respects busy_timeout)
- cbm_store_check_integrity: detects corrupt DBs (>5 project rows
  or invalid root_path), auto-deletes with ERROR log
- cbm_store_get_db accessor for test corruption simulation

Tests: 5 integrity tests + 4 concurrency lock tests

Author: DeusData

src/graph_buffer/graph_buffer.c

@@ -971,23 +971,12 @@ int cbm_gbuf_dump_to_sqlite(cbm_gbuf_t *gb, const char *path) {
     cbm_ht_free(gb->edges_by_type);
     gb->edges_by_type = NULL;
 
-    /* Write to .tmp first, then atomic rename */
-    char tmp_path[1040];
-    snprintf(tmp_path, sizeof(tmp_path), "%s.tmp", path);
-
-    int rc = cbm_write_db(tmp_path, gb->project, gb->root_path, indexed_at, dump_nodes, node_idx,
+    /* Write directly to final path — no .tmp + rename.
+     * Callers must delete the old .db before calling this (reindex)
+     * or ensure no file exists (first index). */
+    int rc = cbm_write_db(path, gb->project, gb->root_path, indexed_at, dump_nodes, node_idx,
                           dump_edges, edge_idx);
 
-    if (rc == 0) {
-        rc = rename(tmp_path, path);
-        if (rc != 0) {
-            cbm_log_error("gbuf.dump", "op", "rename", "path", path);
-        }
-    } else {
-        // NOLINTNEXTLINE(cert-err33-c) — best-effort cleanup
-        remove(tmp_path);
-    }
-
     {
         char b1[16];
         char b2[16];

src/main.c

@@ -77,15 +77,25 @@ static void *http_thread(void *arg) {
 
 static int watcher_index_fn(const char *project_name, const char *root_path, void *user_data) {
     (void)user_data;
+
+    /* Non-blocking: skip if another pipeline is already running.
+     * Watcher will retry on next poll cycle (5-60s). */
+    if (!cbm_pipeline_try_lock()) {
+        cbm_log_info("watcher.skip", "project", project_name, "reason", "pipeline_busy");
+        return 0;
+    }
+
     cbm_log_info("watcher.reindex", "project", project_name, "path", root_path);
 
     cbm_pipeline_t *p = cbm_pipeline_new(root_path, NULL, CBM_MODE_FULL);
     if (!p) {
+        cbm_pipeline_unlock();
         return -1;
     }
 
     int rc = cbm_pipeline_run(p);
     cbm_pipeline_free(p);
+    cbm_pipeline_unlock();
     return rc;
 }
 

src/mcp/mcp.c

@@ -1113,6 +1113,9 @@ static char *handle_delete_project(cbm_mcp_server_t *srv, const char *args) {
         srv->current_project = NULL;
     }
 
+    /* Wait for any in-progress pipeline to finish before deleting */
+    cbm_pipeline_lock();
+
     /* Delete the .db file + WAL/SHM */
     char path[1024];
     project_db_path(name, path, sizeof(path));
@@ -1131,6 +1134,8 @@ static char *handle_delete_project(cbm_mcp_server_t *srv, const char *args) {
         status = "deleted";
     }
 
+    cbm_pipeline_unlock();
+
     yyjson_mut_doc *doc = yyjson_mut_doc_new(NULL);
     yyjson_mut_val *root = yyjson_mut_obj(doc);
     yyjson_mut_doc_set_root(doc, root);
@@ -1437,9 +1442,19 @@ static char *handle_index_repository(cbm_mcp_server_t *srv, const char *args) {
 
     char *project_name = heap_strdup(cbm_pipeline_project_name(p));
 
-    /* Pipeline builds everything in-memory, then dumps to file atomically.
-     * No need to close srv->store — pipeline doesn't touch the open store. */
+    /* Close cached store — pipeline will delete + recreate the .db file */
+    if (srv->owns_store && srv->store) {
+        cbm_store_close(srv->store);
+        srv->store = NULL;
+    }
+    free(srv->current_project);
+    srv->current_project = NULL;
+
+    /* Serialize pipeline runs to prevent concurrent writes */
+    cbm_pipeline_lock();
     int rc = cbm_pipeline_run(p);
+    cbm_pipeline_unlock();
+
     cbm_pipeline_free(p);
     cbm_mem_collect(); /* return mimalloc pages to OS after large indexing */
 
@@ -2749,7 +2764,11 @@ static void *autoindex_thread(void *arg) {
         return NULL;
     }
 
+    /* Block until any concurrent pipeline finishes */
+    cbm_pipeline_lock();
     int rc = cbm_pipeline_run(p);
+    cbm_pipeline_unlock();
+
     cbm_pipeline_free(p);
     cbm_mem_collect(); /* return mimalloc pages to OS after indexing */
 

src/pipeline/pipeline.c

@@ -33,6 +33,28 @@
 #include <sys/stat.h>
 #include <time.h>
 
+/* ── Global index lock ─────────────────────────────────────────── */
+/* Prevents concurrent pipeline runs on the same DB file.
+ * Atomic spinlock: 0 = free, 1 = locked. */
+static atomic_int g_pipeline_busy = 0;
+
+bool cbm_pipeline_try_lock(void) {
+    return atomic_exchange(&g_pipeline_busy, 1) == 0;
+}
+
+#define LOCK_SPIN_NS 100000000 /* 100ms between lock retries */
+
+void cbm_pipeline_lock(void) {
+    while (atomic_exchange(&g_pipeline_busy, 1) != 0) {
+        struct timespec ts = {0, LOCK_SPIN_NS};
+        cbm_nanosleep(&ts, NULL);
+    }
+}
+
+void cbm_pipeline_unlock(void) {
+    atomic_store(&g_pipeline_busy, 0);
+}
+
 /* ── Internal state ──────────────────────────────────────────────── */
 
 struct cbm_pipeline {
@@ -344,45 +366,44 @@ int cbm_pipeline_run(cbm_pipeline_t *p) {
         return -1;
     }
 
-    /* Check for existing DB with file hashes → incremental path */
+    /* Check for existing DB → route to incremental or delete for reindex */
     {
         char *db_path = resolve_db_path(p);
         if (db_path) {
             struct stat db_st;
             if (stat(db_path, &db_st) == 0) {
-                /* DB exists — check if it has file hashes */
+                /* DB exists — try incremental path first */
                 cbm_store_t *check_store = cbm_store_open_path(db_path);
-                if (check_store) {
-                    /* Integrity check — corrupt DB → delete and fall through to full reindex */
-                    if (!cbm_store_check_integrity(check_store)) {
-                        cbm_log_error("pipeline.corrupt_db", "path", db_path, "action",
-                                      "deleting — will do full reindex");
-                        cbm_store_close(check_store);
-                        cbm_unlink(db_path);
-                        char wal[1040];
-                        char shm[1040];
-                        snprintf(wal, sizeof(wal), "%s-wal", db_path);
-                        snprintf(shm, sizeof(shm), "%s-shm", db_path);
-                        cbm_unlink(wal);
-                        cbm_unlink(shm);
-                    } else {
-                        cbm_file_hash_t *hashes = NULL;
-                        int hash_count = 0;
-                        cbm_store_get_file_hashes(check_store, p->project_name, &hashes,
-                                                  &hash_count);
-                        cbm_store_free_file_hashes(hashes, hash_count);
-                        cbm_store_close(check_store);
-
-                        if (hash_count > 0) {
-                            cbm_log_info("pipeline.route", "path", "incremental", "stored_hashes",
-                                         itoa_buf(hash_count));
-                            rc = cbm_pipeline_run_incremental(p, db_path, files, file_count);
-                            cbm_discover_free(files, file_count);
-                            free(db_path);
-                            return rc;
-                        }
+                if (check_store && cbm_store_check_integrity(check_store)) {
+                    cbm_file_hash_t *hashes = NULL;
+                    int hash_count = 0;
+                    cbm_store_get_file_hashes(check_store, p->project_name, &hashes, &hash_count);
+                    cbm_store_free_file_hashes(hashes, hash_count);
+                    cbm_store_close(check_store);
+
+                    if (hash_count > 0) {
+                        cbm_log_info("pipeline.route", "path", "incremental", "stored_hashes",
+                                     itoa_buf(hash_count));
+                        rc = cbm_pipeline_run_incremental(p, db_path, files, file_count);
+                        cbm_discover_free(files, file_count);
+                        free(db_path);
+                        return rc;
                     }
+                } else if (check_store) {
+                    cbm_store_close(check_store);
                 }
+
+                /* Not eligible for incremental → reindex: delete old DB first.
+                 * cbm_write_db writes directly to the final path (no rename),
+                 * so the old file must be gone before the pipeline dumps. */
+                cbm_log_info("pipeline.route", "path", "reindex", "action", "deleting old db");
+                cbm_unlink(db_path);
+                char wal[1040];
+                char shm[1040];
+                snprintf(wal, sizeof(wal), "%s-wal", db_path);
+                snprintf(shm, sizeof(shm), "%s-shm", db_path);
+                cbm_unlink(wal);
+                cbm_unlink(shm);
             }
             free(db_path);
         }

src/pipeline/pipeline.h

@@ -55,6 +55,20 @@ void cbm_pipeline_cancel(cbm_pipeline_t *p);
  * owned by the pipeline. Valid until cbm_pipeline_free(). */
 const char *cbm_pipeline_project_name(const cbm_pipeline_t *p);
 
+/* ── Index lock (prevents concurrent pipeline runs on same DB) ──── */
+
+/* Try to acquire the global index lock. Returns true if acquired,
+ * false if another pipeline is already running (non-blocking).
+ * Use this in the watcher — skip reindex if busy. */
+bool cbm_pipeline_try_lock(void);
+
+/* Acquire the global index lock, blocking until available.
+ * Use this in MCP handler and autoindex — wait for busy watcher to finish. */
+void cbm_pipeline_lock(void);
+
+/* Release the global index lock. */
+void cbm_pipeline_unlock(void);
+
 /* ── FQN helpers (used by passes and external callers) ──────────── */
 
 /* Compute a qualified name: project.dir.parts.name

src/store/store.c

@@ -1055,7 +1055,7 @@ int cbm_store_upsert_node_batch(cbm_store_t *s, const cbm_node_t *nodes, int cou
         return CBM_STORE_OK;
     }
 
-    exec_sql(s, "BEGIN;");
+    exec_sql(s, "BEGIN IMMEDIATE;");
     for (int i = 0; i < count; i++) {
         int64_t id = cbm_store_upsert_node(s, &nodes[i]);
         if (id == CBM_STORE_ERR) {
@@ -1291,7 +1291,7 @@ int cbm_store_insert_edge_batch(cbm_store_t *s, const cbm_edge_t *edges, int cou
         return CBM_STORE_OK;
     }
 
-    exec_sql(s, "BEGIN;");
+    exec_sql(s, "BEGIN IMMEDIATE;");
     for (int i = 0; i < count; i++) {
         int64_t id = cbm_store_insert_edge(s, &edges[i]);
         if (id == CBM_STORE_ERR) {

tests/test_pipeline.c

@@ -14,6 +14,7 @@
 #include <stdlib.h>
 #include <string.h>
 #include <stdatomic.h>
+#include "foundation/compat_thread.h"
 #include <sys/stat.h>
 #include <unistd.h>
 #include "graph_buffer/graph_buffer.h"
@@ -5213,7 +5214,89 @@ TEST(incremental_kustomize_module_indexed) {
     PASS();
 }
 
+/* ── Index lock tests ───────────────────────────────────────────── */
+
+TEST(pipeline_lock_try_acquire) {
+    /* First try-lock should succeed */
+    ASSERT_TRUE(cbm_pipeline_try_lock());
+    /* Second try-lock should fail (already held) */
+    ASSERT_FALSE(cbm_pipeline_try_lock());
+    /* Release, then re-acquire should succeed */
+    cbm_pipeline_unlock();
+    ASSERT_TRUE(cbm_pipeline_try_lock());
+    cbm_pipeline_unlock();
+    PASS();
+}
+
+TEST(pipeline_lock_blocking) {
+    /* Lock, then unlock — basic sanity */
+    cbm_pipeline_lock();
+    cbm_pipeline_unlock();
+    /* Should be immediately re-acquirable */
+    cbm_pipeline_lock();
+    cbm_pipeline_unlock();
+    PASS();
+}
+
+/* Thread function that tries to acquire the lock and records result */
+static atomic_int g_thread_acquired = 0;
+static atomic_int g_thread_done = 0;
+
+static void *try_lock_thread(void *arg) {
+    (void)arg;
+    if (cbm_pipeline_try_lock()) {
+        atomic_store(&g_thread_acquired, 1);
+        cbm_pipeline_unlock();
+    } else {
+        atomic_store(&g_thread_acquired, 0);
+    }
+    atomic_store(&g_thread_done, 1);
+    return NULL;
+}
+
+TEST(pipeline_lock_contention) {
+    /* Main thread holds lock, spawned thread should fail try_lock */
+    cbm_pipeline_lock();
+    atomic_store(&g_thread_acquired, -1);
+    atomic_store(&g_thread_done, 0);
+
+    cbm_thread_t tid;
+    int rc = cbm_thread_create(&tid, 0, try_lock_thread, NULL);
+    ASSERT_EQ(rc, 0);
+
+    /* Wait for thread to finish */
+    cbm_thread_join(&tid);
+
+    /* Thread should NOT have acquired the lock */
+    ASSERT_EQ(atomic_load(&g_thread_acquired), 0);
+    cbm_pipeline_unlock();
+    PASS();
+}
+
+TEST(pipeline_lock_release_allows_contender) {
+    /* Main thread acquires and releases, then spawned thread should succeed */
+    cbm_pipeline_lock();
+    cbm_pipeline_unlock();
+
+    atomic_store(&g_thread_acquired, -1);
+    atomic_store(&g_thread_done, 0);
+
+    cbm_thread_t tid;
+    int rc = cbm_thread_create(&tid, 0, try_lock_thread, NULL);
+    ASSERT_EQ(rc, 0);
+    cbm_thread_join(&tid);
+
+    /* Thread SHOULD have acquired the lock */
+    ASSERT_EQ(atomic_load(&g_thread_acquired), 1);
+    PASS();
+}
+
 SUITE(pipeline) {
+    /* Index lock */
+    RUN_TEST(pipeline_lock_try_acquire);
+    RUN_TEST(pipeline_lock_blocking);
+    RUN_TEST(pipeline_lock_contention);
+    RUN_TEST(pipeline_lock_release_allows_contender);
     /* Lifecycle */
     RUN_TEST(pipeline_create_free);
     RUN_TEST(pipeline_null_repo);