Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 11 additions & 5 deletions src/main.c
Original file line number Diff line number Diff line change
Expand Up @@ -164,32 +164,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();
Expand Down
41 changes: 38 additions & 3 deletions src/mcp/mcp.c
Original file line number Diff line number Diff line change
Expand Up @@ -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);

Expand Down Expand Up @@ -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];
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand All @@ -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);
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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();

Expand Down
141 changes: 136 additions & 5 deletions src/pipeline/pipeline.c
Original file line number Diff line number Diff line change
Expand Up @@ -41,32 +41,163 @@ 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
(void)project;
(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);
}

Expand Down
13 changes: 8 additions & 5 deletions src/pipeline/pipeline.h
Original file line number Diff line number Diff line change
Expand Up @@ -111,16 +111,19 @@ 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).
/* Try to acquire the index lock for a project. Returns true if acquired,
* false if that project is already being rebuilt (non-blocking).
* 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.
/* Acquire the index lock for a project, blocking until available.
* Returns false only when the lock backend cannot be opened/acquired.
* Use this in MCP handler and autoindex — wait for busy watcher to finish. */
void cbm_pipeline_lock(void);
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) ──────────── */
Expand Down
Loading
Loading