fix(mcp): serialize ADR access across processes

SS-42 · SS-42 · commit 0a8b63ea8bcf · 2026-07-09T21:20:59.000+03:00
Use project-scoped cache-local advisory locks so independent MCP and CLI processes sharing CBM_CACHE_DIR cannot rebuild the same project DB concurrently without blocking unrelated project indexes.

Serialize manage_adr reads and writes with same-project rebuilds, reopen stale file-backed stores before ADR access, and avoid watcher self-locking when a supervised worker performs the reindex.

Cover the cross-process lock behavior with fork-based regression tests.

Signed-off-by: SS-42 &lt;noreply@incogni.to&gt;
diff --git a/src/main.c b/src/main.c
@@ -86,8 +86,9 @@ static void request_shutdown(void) {
             cbm_pipeline_cancel(p);
         }
     }
-    /* Release pipeline lock to prevent stale lock on restart */
-    cbm_pipeline_unlock();
+    /* Do not unlock here: cancellation is cooperative, so cbm_pipeline_run()
+     * may still be writing. The lock owner releases after the run returns; if
+     * the process dies first, the OS releases the file descriptor lock. */
 
     if (g_watcher) {
         cbm_watcher_stop(g_watcher);
@@ -164,32 +165,38 @@ static int watcher_index_fn(const char *project_name, const char *root_path, voi
         return 0;
     }
 
-    /* Non-blocking: skip if another pipeline is already running.
-     * Watcher will retry on next poll cycle (5-60s). */
-    if (!cbm_pipeline_try_lock()) {
+    /* Non-blocking: skip if this project is already being rebuilt. Watcher will
+     * retry on next poll cycle (5-60s). */
+    if (!cbm_pipeline_try_lock_project(project_name)) {
         cbm_log_info("watcher.skip", "project", project_name, "reason", "pipeline_busy");
         return 0;
     }
+    cbm_pipeline_unlock();
 
     cbm_log_info("watcher.reindex", "project", project_name, "path", root_path);
 
     /* #832: route the re-index through the supervised worker subprocess so this
      * long-lived server process hands its RSS back to the OS on every cycle
      * instead of ratcheting (mimalloc v3 does not reclaim pages that worker
      * threads abandon at exit). The child writes the DB; the parent only needs the
-     * return code. The pipeline lock (already held) still serialises re-indexes.
+     * return code. The worker takes the project lock; the parent must not keep
+     * it held while waiting, otherwise the child blocks on the same file lock.
      * Degrade to the in-process pipeline when the supervisor is off (kill switch)
      * or the spawn fails. */
     if (cbm_index_supervisor_should_wrap()) {
         char *resp = cbm_mcp_index_run_supervised_path(root_path);
         if (resp) {
             free(resp);
-            cbm_pipeline_unlock();
             return 0;
         }
         /* resp == NULL → spawn-failure degrade → fall through to in-process. */
     }
 
+    if (!cbm_pipeline_try_lock_project(project_name)) {
+        cbm_log_info("watcher.skip", "project", project_name, "reason", "pipeline_busy");
+        return 0;
+    }
+
     cbm_pipeline_t *p = cbm_pipeline_new(root_path, NULL, CBM_MODE_FULL);
     if (!p) {
         cbm_pipeline_unlock();
diff --git a/src/mcp/mcp.c b/src/mcp/mcp.c
@@ -1084,6 +1084,18 @@ static cbm_store_t *resolve_store(cbm_mcp_server_t *srv, const char *project) {
     return srv->store;
 }
 
+static void invalidate_cached_file_store(cbm_mcp_server_t *srv) {
+    if (!srv || !srv->store || !cbm_store_db_path(srv->store)) {
+        return;
+    }
+    if (srv->owns_store) {
+        cbm_store_close(srv->store);
+        srv->store = NULL;
+        free(srv->current_project);
+        srv->current_project = NULL;
+    }
+}
+
 /* Forward decl — definition lives below alongside list_projects. */
 static bool is_project_db_file(const char *name, size_t len);
 
@@ -2192,7 +2204,10 @@ static char *handle_delete_project(cbm_mcp_server_t *srv, const char *args) {
     }
 
     /* Wait for any in-progress pipeline to finish before deleting */
-    cbm_pipeline_lock();
+    if (!cbm_pipeline_lock_project(name)) {
+        free(name);
+        return cbm_mcp_text_result("failed to acquire pipeline lock", true);
+    }
 
     /* Delete the .db file + WAL/SHM */
     char path[CBM_SZ_1K];
@@ -3811,7 +3826,12 @@ static char *handle_index_repository(cbm_mcp_server_t *srv, const char *args) {
     /* Serialize pipeline runs to prevent concurrent writes.
      * Track active pipeline so signal handler and notifications/cancelled
      * can cancel it mid-run. */
-    cbm_pipeline_lock();
+    if (!cbm_pipeline_lock_project(project_name)) {
+        cbm_pipeline_free(p);
+        free(project_name);
+        free(repo_path);
+        return cbm_mcp_text_result("failed to acquire pipeline lock", true);
+    }
     srv->active_pipeline = p;
     int rc = cbm_pipeline_run(p);
     srv->active_pipeline = NULL;
@@ -5468,11 +5488,20 @@ static char *handle_manage_adr(cbm_mcp_server_t *srv, const char *args) {
     /* ADRs are stored in the SQLite store (project_summaries), the SAME
      * backend the UI /api/adr endpoints use — so writes via the MCP tool and
      * the UI are visible to each other (#256). */
+    if (!cbm_pipeline_lock_project(project)) {
+        free(project);
+        free(mode_str);
+        free(content);
+        return cbm_mcp_text_result("failed to acquire pipeline lock", true);
+    }
+    invalidate_cached_file_store(srv);
+
     cbm_store_t *resolved = resolve_store(srv, project);
     if (!resolved) {
         char *err = build_no_store_error(project);
         char *res = cbm_mcp_text_result(err, true);
         free(err);
+        cbm_pipeline_unlock();
         free(project);
         free(mode_str);
         free(content);
@@ -5495,6 +5524,7 @@ static char *handle_manage_adr(cbm_mcp_server_t *srv, const char *args) {
             char *err = build_no_store_error(project);
             char *res = cbm_mcp_text_result(err, true);
             free(err);
+            cbm_pipeline_unlock();
             free(project);
             free(mode_str);
             free(content);
@@ -5553,6 +5583,7 @@ static char *handle_manage_adr(cbm_mcp_server_t *srv, const char *args) {
     if (owned_rw) {
         cbm_store_close(owned_rw);
     }
+    cbm_pipeline_unlock();
     free(project);
     free(mode_str);
     free(content);
@@ -5741,7 +5772,11 @@ static void *autoindex_thread(void *arg) {
     }
 
     /* Block until any concurrent pipeline finishes */
-    cbm_pipeline_lock();
+    if (!cbm_pipeline_lock_project(srv->session_project)) {
+        cbm_pipeline_free(p);
+        cbm_log_warn("autoindex.err", "msg", "pipeline_lock_failed");
+        return NULL;
+    }
     int rc = cbm_pipeline_run(p);
     cbm_pipeline_unlock();
 
diff --git a/src/pipeline/pipeline.c b/src/pipeline/pipeline.c
@@ -41,32 +41,165 @@ enum { CBM_DIR_PERMS = 0755, PL_RING = 4, PL_RING_MASK = 3, PL_SEQ_PASSES = 6, P
 #include <stdatomic.h>
 #include <sys/stat.h>
 #include <time.h>
+#include <errno.h>
+#ifndef _WIN32
+#include <fcntl.h>
+#include <unistd.h>
+#endif
 
 static inline void *intptr_to_ptr(intptr_t v) {
     void *p;
     memcpy(&p, &v, sizeof(p));
     return p;
 }
 
-/* ── Global index lock ─────────────────────────────────────────── */
-/* Prevents concurrent pipeline runs on the same DB file.
- * Atomic spinlock: 0 = free, 1 = locked. */
+/* ── Project index lock ────────────────────────────────────────── */
+/* Prevents concurrent rebuilds of the same project DB. The atomic lock handles
+ * threads inside one process; the project-scoped cache-local file lock serializes
+ * independent MCP/CLI processes that share CBM_CACHE_DIR. */
 static atomic_int g_pipeline_busy = 0;
+#ifndef _WIN32
+static int g_pipeline_lock_fd = -1;
+#endif
+
+static bool pipeline_file_lock(const char *project, bool wait) {
+#ifdef _WIN32
+    /* Windows still has only the in-process guard; add a LockFileEx backend
+     * here before claiming cross-process serialization on that platform. */
+    cbm_log_debug("pipeline.lock.win32_noop", "project", project ? project : "(global)");
+    (void)wait;
+    return true;
+#else
+    const char *cache_dir = cbm_resolve_cache_dir();
+    if (!cache_dir || !cache_dir[0]) {
+        cache_dir = cbm_tmpdir();
+    }
+    if (!cache_dir || !cache_dir[0]) {
+        cbm_log_warn("pipeline.lock.err", "phase", "cache_dir");
+        return false;
+    }
+    if (!cbm_mkdir_p(cache_dir, CBM_DIR_PERMS)) {
+        cbm_log_warn("pipeline.lock.err", "phase", "mkdir", "path", cache_dir);
+        return false;
+    }
+
+    char lock_path[CBM_SZ_1K];
+    int n;
+    if (project && cbm_validate_project_name(project)) {
+        n = snprintf(lock_path, sizeof(lock_path), "%s/.pipeline.%s.lock", cache_dir, project);
+    } else {
+        n = snprintf(lock_path, sizeof(lock_path), "%s/.pipeline.lock", cache_dir);
+    }
+    if (n <= 0 || (size_t)n >= sizeof(lock_path)) {
+        cbm_log_warn("pipeline.lock.err", "phase", "path", "project",
+                     project ? project : "(global)");
+        return false;
+    }
+
+    int flags = O_RDWR | O_CREAT;
+#ifdef O_CLOEXEC
+    flags |= O_CLOEXEC;
+#endif
+    int fd = open(lock_path, flags, 0600);
+#ifdef O_CLOEXEC
+    if (fd < 0 && errno == EINVAL) {
+        fd = open(lock_path, O_RDWR | O_CREAT, 0600);
+    }
+#endif
+    if (fd < 0) {
+        char errbuf[CBM_SZ_64];
+        snprintf(errbuf, sizeof(errbuf), "%d", errno);
+        cbm_log_warn("pipeline.lock.err", "phase", "open", "errno", errbuf);
+        return false;
+    }
+#ifdef FD_CLOEXEC
+    int fd_flags = fcntl(fd, F_GETFD);
+    if (fd_flags >= 0) {
+        (void)fcntl(fd, F_SETFD, fd_flags | FD_CLOEXEC);
+    }
+#endif
+
+    struct flock fl;
+    memset(&fl, 0, sizeof(fl));
+    fl.l_type = F_WRLCK;
+    fl.l_whence = SEEK_SET;
+
+    for (;;) {
+        if (fcntl(fd, wait ? F_SETLKW : F_SETLK, &fl) == 0) {
+            g_pipeline_lock_fd = fd;
+            return true;
+        }
+        if (wait && errno == EINTR) {
+            continue;
+        }
+        int err = errno;
+        bool lock_busy = err == EACCES || err == EAGAIN;
+#ifdef EWOULDBLOCK
+        lock_busy = lock_busy || err == EWOULDBLOCK;
+#endif
+        if (!wait && lock_busy) {
+            close(fd);
+            return false;
+        }
+        char errbuf[CBM_SZ_64];
+        snprintf(errbuf, sizeof(errbuf), "%d", err);
+        cbm_log_warn("pipeline.lock.err", "phase", "acquire", "errno", errbuf);
+        close(fd);
+        return false;
+    }
+#endif
+}
+
+static void pipeline_file_unlock(void) {
+#ifndef _WIN32
+    if (g_pipeline_lock_fd < 0) {
+        return;
+    }
+    struct flock fl;
+    memset(&fl, 0, sizeof(fl));
+    fl.l_type = F_UNLCK;
+    fl.l_whence = SEEK_SET;
+    (void)fcntl(g_pipeline_lock_fd, F_SETLK, &fl);
+    close(g_pipeline_lock_fd);
+    g_pipeline_lock_fd = -1;
+#endif
+}
+
+bool cbm_pipeline_try_lock_project(const char *project) {
+    if (atomic_exchange(&g_pipeline_busy, 1) != 0) {
+        return false;
+    }
+    if (!pipeline_file_lock(project, false)) {
+        atomic_store(&g_pipeline_busy, 0);
+        return false;
+    }
+    return true;
+}
 
 bool cbm_pipeline_try_lock(void) {
-    return atomic_exchange(&g_pipeline_busy, 1) == 0;
+    return cbm_pipeline_try_lock_project(NULL);
 }
 
 #define LOCK_SPIN_NS 100000000 /* 100ms between lock retries */
 
-void cbm_pipeline_lock(void) {
+bool cbm_pipeline_lock_project(const char *project) {
     while (atomic_exchange(&g_pipeline_busy, 1) != 0) {
         struct timespec ts = {0, LOCK_SPIN_NS};
         cbm_nanosleep(&ts, NULL);
     }
+    if (!pipeline_file_lock(project, true)) {
+        atomic_store(&g_pipeline_busy, 0);
+        return false;
+    }
+    return true;
+}
+
+bool cbm_pipeline_lock(void) {
+    return cbm_pipeline_lock_project(NULL);
 }
 
 void cbm_pipeline_unlock(void) {
+    pipeline_file_unlock();
     atomic_store(&g_pipeline_busy, 0);
 }
 
diff --git a/src/pipeline/pipeline.h b/src/pipeline/pipeline.h
@@ -111,16 +111,24 @@ void cbm_pipeline_get_file_errors(const cbm_pipeline_t *p, cbm_file_error_t **ou
 
 /* ── 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. */
+/* Try to acquire the index lock for a project. Returns true if acquired,
+ * false if this process is already running any pipeline or another process is
+ * rebuilding the same project (non-blocking). The cross-process file lock is
+ * project-scoped; the in-process guard remains process-wide because the
+ * pipeline uses process-global state. Use this in the watcher — skip reindex if
+ * busy. */
+bool cbm_pipeline_try_lock_project(const char *project);
 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);
+/* Acquire the index lock for a project, blocking until available.
+ * Returns false only when the lock backend cannot be opened/acquired. Inside
+ * one process this still serializes all pipeline runs; across processes it only
+ * waits for the same project. Use this in MCP handler and autoindex — wait for
+ * busy watcher to finish. */
+bool cbm_pipeline_lock_project(const char *project);
+bool cbm_pipeline_lock(void);
 
-/* Release the global index lock. */
+/* Release the held index lock. */
 void cbm_pipeline_unlock(void);
 
 /* ── FQN helpers (used by passes and external callers) ──────────── */
diff --git a/tests/test_mcp.c b/tests/test_mcp.c
diff --git a/tests/test_pipeline.c b/tests/test_pipeline.c
