From 4ce6b115956d51e7a0e115187c29424188170d7c Mon Sep 17 00:00:00 2001 From: Martin Vogel Date: Tue, 8 Sep 2026 01:19:34 +0200 Subject: [PATCH 1/5] fix(pipeline): never mint a staging path from a staging path The outer run rewrites the pipeline's db_path to its own stage, so the inner publication (gbuf dump and the delta clone) minted ITS stage from that stage: .stage.A.stage.B, with -wal/-shm beside it under WAL. Every such copy is a full-size generation, so one project could occupy three times its database on disk (#1839). Mint from the stage root instead. The root is recognised only by the exact minted shape ".stage.<6 alphanumerics>" in the basename, so a database whose own name merely contains ".stage." keeps its full name, and the marker is never matched across a path separator. The inner finalize still renames B over A; the unpredictable-name contract of the mkstemp helper is unchanged. Co-Authored-By: Claude Fable 5.1 Signed-off-by: Martin Vogel --- src/pipeline/pipeline.c | 47 +++++++++++++++++++++++- tests/test_pipeline.c | 81 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 127 insertions(+), 1 deletion(-) diff --git a/src/pipeline/pipeline.c b/src/pipeline/pipeline.c index 0ffb46c0b..247f72378 100644 --- a/src/pipeline/pipeline.c +++ b/src/pipeline/pipeline.c @@ -38,6 +38,7 @@ enum { CBM_DIR_PERMS = 0755, PL_RING = 4, PL_RING_MASK = 3, PL_SEQ_PASSES = 6 }; #include "foundation/profile.h" #include "foundation/mem.h" +#include #include #include #include @@ -2415,12 +2416,56 @@ static bool ensure_db_parent(const char *path) { return ok; } +/* Length of the path a stage was minted for: the input itself unless its + * basename has exactly the minted shape ".stage.<6 alphanumerics>", in + * which case the root is . The outer run rewrites the pipeline's db_path + * to its stage, so the inner publication (dump and delta clone) used to mint + * ITS stage from that stage: .stage.A.stage.B, with -wal/-shm beside it + * (#1839). Minting from the root keeps every generation's stage a sibling of + * the live database. Only the exact minted shape is recognised: a database + * named "x.stage.y.db" is not a stage and keeps its full name. */ +enum { CBM_STAGE_SUFFIX_RANDOM_CHARS = 6 }; +static const char cbm_stage_marker[] = ".stage."; + +static bool stage_suffix_at(const char *tail) { + if (strncmp(tail, cbm_stage_marker, sizeof(cbm_stage_marker) - 1) != 0) { + return false; + } + const char *random = tail + sizeof(cbm_stage_marker) - 1; + for (int i = 0; i < CBM_STAGE_SUFFIX_RANDOM_CHARS; i++) { + if (!isalnum((unsigned char)random[i])) { + return false; + } + } + return random[CBM_STAGE_SUFFIX_RANDOM_CHARS] == '\0'; +} + +static size_t stage_root_length(const char *path) { + size_t len = strlen(path); + const size_t suffix_len = sizeof(cbm_stage_marker) - 1 + CBM_STAGE_SUFFIX_RANDOM_CHARS; + if (len <= suffix_len) { + return len; + } + size_t root_len = len - suffix_len; + /* The marker must sit inside the basename, never span a separator. */ + for (size_t i = root_len; i < len; i++) { + if (path[i] == '/' +#ifdef _WIN32 + || path[i] == '\\' +#endif + ) { + return len; + } + } + return stage_suffix_at(path + root_len) ? root_len : len; +} + static char *create_staging_path(const char *final_path) { if (!final_path) { return NULL; } static const char suffix[] = ".stage.XXXXXX"; - size_t final_len = strlen(final_path); + size_t final_len = stage_root_length(final_path); if (final_len > SIZE_MAX - sizeof(suffix)) { return NULL; } diff --git a/tests/test_pipeline.c b/tests/test_pipeline.c index 0f6332cb2..e724736e2 100644 --- a/tests/test_pipeline.c +++ b/tests/test_pipeline.c @@ -3416,6 +3416,86 @@ TEST(pipeline_publication_never_uses_a_predictable_staging_path) { PASS(); } +static int count_substring(const char *haystack, const char *needle) { + int count = 0; + size_t needle_len = strlen(needle); + for (const char *at = strstr(haystack, needle); at; at = strstr(at + needle_len, needle)) { + count++; + } + return count; +} + +static int count_nested_stage_entries(const char *dir_path, const char *db_basename) { + cbm_dir_t *dir = cbm_opendir(dir_path); + if (!dir) { + return -1; + } + size_t base_len = strlen(db_basename); + int count = 0; + cbm_dirent_t *entry; + while ((entry = cbm_readdir(dir)) != NULL) { + if (strncmp(entry->name, db_basename, base_len) == 0 && + count_substring(entry->name + base_len, ".stage.") >= 2) { + count++; + } + } + cbm_closedir(dir); + return count; +} + +/* #1839: the outer run rewrites the pipeline's db_path to its stage, so the + * inner publication (dump or delta clone) minted ITS stage from a stage: + * .stage.A.stage.B, plus -wal/-shm under WAL. A generation's stage is a + * sibling of the live database, whichever path it is minted from; a database + * whose own basename merely contains ".stage." is not a stage and keeps its + * full name. */ +TEST(pipeline_stage_names_never_nest) { + char tmp[256]; + snprintf(tmp, sizeof(tmp), "/tmp/cbm_stage_nesting_XXXXXX"); + ASSERT_NOT_NULL(cbm_mkdtemp(tmp)); + char db_path[512]; + snprintf(db_path, sizeof(db_path), "%s/generation.db", tmp); + char odd_db[512]; + snprintf(odd_db, sizeof(odd_db), "%s/x.stage.y.db", tmp); + + char *outer = cbm_pipeline_create_staging_path(db_path); + ASSERT_NOT_NULL(outer); + char *inner = cbm_pipeline_create_staging_path(outer); + ASSERT_NOT_NULL(inner); + char *odd_stage = cbm_pipeline_create_staging_path(odd_db); + ASSERT_NOT_NULL(odd_stage); + + size_t db_len = strlen(db_path); + size_t odd_len = strlen(odd_db); + int outer_tokens = count_substring(outer, ".stage."); + int inner_tokens = count_substring(inner, ".stage."); + bool inner_is_sibling = + strncmp(inner, db_path, db_len) == 0 && strncmp(inner + db_len, ".stage.", 7) == 0; + bool inner_distinct = strcmp(inner, outer) != 0; + bool odd_keeps_basename = + strncmp(odd_stage, odd_db, odd_len) == 0 && strncmp(odd_stage + odd_len, ".stage.", 7) == 0; + int nested_entries = count_nested_stage_entries(tmp, "generation.db"); + + cbm_pipeline_discard_stage(inner); + cbm_pipeline_discard_stage(outer); + cbm_pipeline_discard_stage(odd_stage); + int leftover_entries = count_generation_stage_artifacts(tmp, "generation.db") + + count_generation_stage_artifacts(tmp, "x.stage.y.db"); + free(inner); + free(outer); + free(odd_stage); + th_rmtree(tmp); + + ASSERT_EQ(outer_tokens, 1); + ASSERT_EQ(inner_tokens, 1); + ASSERT_TRUE(inner_is_sibling); + ASSERT_TRUE(inner_distinct); + ASSERT_TRUE(odd_keeps_basename); + ASSERT_EQ(nested_entries, 0); + ASSERT_EQ(leftover_entries, 0); + PASS(); +} + /* Discovery and extraction must describe the same immutable generation. A * source file created after extraction is not present in the original file * list, so merely re-hashing that list cannot detect the race. Publication @@ -14120,6 +14200,7 @@ SUITE(pipeline_semantic_manifest_repro) { RUN_TEST(pipeline_git_context_change_forces_full_and_refreshes_branch); RUN_TEST(pipeline_global_extension_config_change_forces_full); RUN_TEST(pipeline_publication_never_uses_a_predictable_staging_path); + RUN_TEST(pipeline_stage_names_never_nest); RUN_TEST(pipeline_source_mutation_before_publication_preserves_previous_generation); RUN_TEST(pipeline_source_addition_before_publication_preserves_previous_generation); RUN_TEST(pipeline_tsconfig_mutation_before_publication_preserves_previous_generation); From 0c4831a3643ba1545d1cf709f056b70739aac04f Mon Sep 17 00:00:00 2001 From: Martin Vogel Date: Tue, 8 Sep 2026 01:37:44 +0200 Subject: [PATCH 2/5] fix(pipeline): own every staging file through an exclusive sidecar lock A stage was recognisable only by its name: the mkstemp descriptor was closed at once and nothing marked who was writing it. A worker killed mid-run -- the daemon cancels with SIGTERM and SIGKILLs after one second of grace, which a gigabyte backup or delta clone never finishes inside -- left its full-size stage behind for good, and no later run could tell a dead stage from a live one, so none tried (#1839). Ownership is now an exclusive kernel lock on the sidecar ".lock", taken when the stage is minted and dropped by the same helpers that discard the stage or rename it into place. The kernel releases it on any death, so "can I take this lock?" is exactly "is this stage dead?" -- no pid, no mtime, no grace period. The lock lives on a sidecar rather than the stage file itself because on macOS an flock conflicts with SQLite's fcntl byte locks on the same file (verified: flock(LOCK_EX) then fcntl(F_WRLCK) on another descriptor fails with EAGAIN, both directions). The primitive is cbm_lockfile_open: POSIX open(O_CLOEXEC|O_NOFOLLOW) plus flock(LOCK_EX|LOCK_NB); Windows _wsopen with _SH_DENYRW and _O_NOINHERIT. Neither is inherited by spawned children. A backup that fails keeps the outer stage NAME owned: the rebuilt generation is renamed over it by the inner finalize and published from it, so its lock is held to the end. The sweep that uses this ownership follows separately. Co-Authored-By: Claude Fable 5.1 Signed-off-by: Martin Vogel --- src/foundation/compat_fs.c | 61 +++++++++++- src/foundation/compat_fs.h | 12 +++ src/pipeline/pipeline.c | 156 ++++++++++++++++++++++++++++++- src/pipeline/pipeline_internal.h | 11 +++ tests/test_pipeline.c | 78 +++++++++++++--- 5 files changed, 295 insertions(+), 23 deletions(-) diff --git a/src/foundation/compat_fs.c b/src/foundation/compat_fs.c index bc1b06ac3..b478002eb 100644 --- a/src/foundation/compat_fs.c +++ b/src/foundation/compat_fs.c @@ -23,11 +23,13 @@ #include #include -#include /* _wmkdir */ -#include /* errno for spawn-failure logging */ -#include /* _O_RDONLY */ -#include /* _wunlink, _open_osfhandle, _close */ -#include /* intptr_t */ +#include /* _wmkdir */ +#include /* errno for spawn-failure logging */ +#include /* _O_RDONLY */ +#include /* _wunlink, _open_osfhandle, _close */ +#include /* _SH_DENYRW */ +#include /* _S_IREAD */ +#include /* intptr_t */ #include "foundation/log.h" #include "foundation/win_utf8.h" @@ -581,6 +583,26 @@ int cbm_rmdir(const char *path) { return ret; } +int cbm_lockfile_open(const char *path, bool create) { + wchar_t *wpath = cbm_path_to_wide(path); + if (!wpath) { + errno = EINVAL; + return -1; + } + int flags = _O_RDWR | _O_BINARY | _O_NOINHERIT | (create ? _O_CREAT : 0); + /* _SH_DENYRW: every other open of this file, from any process, fails + * with EACCES until this descriptor closes -- including at death. */ + int fd = _wsopen(wpath, flags, _SH_DENYRW, _S_IREAD | _S_IWRITE); + free(wpath); + return fd; +} + +void cbm_lockfile_close(int fd) { + if (fd >= 0) { + (void)_close(fd); + } +} + /* Build a properly-quoted Windows command line from an argv array. * Returns a heap-allocated wide string, or NULL on allocation failure. * Quoting follows the MSVC CRT convention: arguments containing spaces, @@ -730,6 +752,7 @@ int cbm_exec_no_shell(const char *const *argv) { #include #include #include +#include #include #include #include @@ -929,6 +952,34 @@ int cbm_rmdir(const char *path) { return rmdir(path); } +int cbm_lockfile_open(const char *path, bool create) { + int flags = O_RDWR | O_CLOEXEC | O_NOFOLLOW | (create ? O_CREAT : 0); + int fd; + do { + fd = open(path, flags, S_IRUSR | S_IWUSR); + } while (fd < 0 && errno == EINTR); + if (fd < 0) { + return -1; + } + int rc; + do { + rc = flock(fd, LOCK_EX | LOCK_NB); + } while (rc != 0 && errno == EINTR); + if (rc != 0) { + int saved = errno; + (void)close(fd); + errno = saved; + return -1; + } + return fd; +} + +void cbm_lockfile_close(int fd) { + if (fd >= 0) { + (void)close(fd); + } +} + int cbm_exec_no_shell(const char *const *argv) { if (!argv || !argv[0]) { return CBM_NOT_FOUND; diff --git a/src/foundation/compat_fs.h b/src/foundation/compat_fs.h index 8f0d16388..e7dd259a0 100644 --- a/src/foundation/compat_fs.h +++ b/src/foundation/compat_fs.h @@ -97,6 +97,18 @@ int cbm_canonical_path(const char *path, char *out, size_t out_sz); /* Delete an empty directory. Returns 0 on success. */ int cbm_rmdir(const char *path); +/* Exclusive lock file. Opens `path` and takes an exclusive lock the KERNEL + * releases on any process death -- POSIX open(O_CLOEXEC|O_NOFOLLOW) + + * flock(LOCK_EX|LOCK_NB); Windows _wsopen with _SH_DENYRW (deny every other + * open) and _O_NOINHERIT -- so ownership never outlives its holder and is + * never inherited by a spawned child. `create` false never creates the file. + * Returns the descriptor, or -1 with errno: EWOULDBLOCK/EAGAIN (POSIX) or + * EACCES (Windows sharing violation) when another holder is live, ENOENT + * when absent, otherwise the open error. Never use this on a SQLite file: + * on macOS an flock conflicts with SQLite's own fcntl byte locks. */ +int cbm_lockfile_open(const char *path, bool create); +void cbm_lockfile_close(int fd); + /* Open a file by UTF-8 path. * On Windows, converts to wide-char and calls _wfopen so paths with * non-ASCII characters (accents, CJK, etc.) are handled correctly. diff --git a/src/pipeline/pipeline.c b/src/pipeline/pipeline.c index 247f72378..59ff88837 100644 --- a/src/pipeline/pipeline.c +++ b/src/pipeline/pipeline.c @@ -1531,12 +1531,148 @@ static bool promote_mode_to_existing_coverage(cbm_pipeline_t *p) { /* Defined below, next to the other publication helpers. */ static char *create_staging_path(const char *final_path); +/* ── Stage ownership (#1839) ───────────────────────────────────── + * + * A stage used to be recognisable only by its name: the mkstemp descriptor + * was closed at once and nothing marked who was writing it. A worker killed + * mid-run (the daemon cancels with SIGTERM then SIGKILL after one second of + * grace, which a gigabyte backup or clone never finishes inside) left its + * full-size stage behind forever, and no later run could tell a dead stage + * from a live one -- so none tried. + * + * Ownership is now an exclusive kernel lock on the sidecar ".lock", + * held from minting until the stage is discarded or renamed into place. The + * kernel releases it on any death, so "can I take this lock?" is exactly + * "is this stage dead?" -- no pid, no mtime, no grace period. The lock lives + * on a sidecar rather than the stage itself because on macOS an flock on a + * file conflicts with SQLite's fcntl byte locks on that same file. + * + * The stage path is passed around as a plain string through publish and + * finalize, so the descriptor is kept in this per-process registry keyed by + * path, and released by the same helpers that remove the file. */ +typedef struct stage_owner { + char *stage_path; + int lock_fd; + struct stage_owner *next; +} stage_owner_t; + +static stage_owner_t *g_stage_owners = NULL; +static atomic_flag g_stage_owners_spin = ATOMIC_FLAG_INIT; + +static void stage_owners_lock(void) { + while (atomic_flag_test_and_set_explicit(&g_stage_owners_spin, memory_order_acquire)) {} +} + +static void stage_owners_unlock(void) { + atomic_flag_clear_explicit(&g_stage_owners_spin, memory_order_release); +} + +static char *stage_lock_sidecar_path(const char *stage_path) { + static const char suffix[] = ".lock"; + size_t len = strlen(stage_path); + if (len > SIZE_MAX - sizeof(suffix)) { + return NULL; + } + char *sidecar = (char *)malloc(len + sizeof(suffix)); + if (!sidecar) { + return NULL; + } + memcpy(sidecar, stage_path, len); + memcpy(sidecar + len, suffix, sizeof(suffix)); + return sidecar; +} + +int cbm_pipeline_stage_lock_hold(const char *stage_path) { + if (!stage_path) { + return -1; + } + char *sidecar = stage_lock_sidecar_path(stage_path); + if (!sidecar) { + return -1; + } + int fd = cbm_lockfile_open(sidecar, true); + free(sidecar); + return fd; +} + +void cbm_pipeline_stage_lock_drop(const char *stage_path, int lock_fd) { + if (!stage_path || lock_fd < 0) { + return; + } + char *sidecar = stage_lock_sidecar_path(stage_path); + /* Close before unlinking: Windows refuses to delete an open file. The + * sidecar exists unlocked for that instant, but by every drop the stage + * itself is already gone (discarded or renamed), so there is nothing a + * sweeper could take from us. */ + cbm_lockfile_close(lock_fd); + if (sidecar) { + (void)cbm_unlink(sidecar); + free(sidecar); + } +} + +static bool stage_owner_register(const char *stage_path) { + int fd = cbm_pipeline_stage_lock_hold(stage_path); + if (fd < 0) { + char errno_text[16]; + (void)snprintf(errno_text, sizeof(errno_text), "%d", errno); + cbm_log_warn("pipeline.stage", "action", "lock_failed", "errno", errno_text, "path", + stage_path); + return false; + } + stage_owner_t *owner = (stage_owner_t *)malloc(sizeof(*owner)); + char *path_copy = strdup(stage_path); + if (!owner || !path_copy) { + free(owner); + free(path_copy); + cbm_pipeline_stage_lock_drop(stage_path, fd); + return false; + } + owner->stage_path = path_copy; + owner->lock_fd = fd; + stage_owners_lock(); + owner->next = g_stage_owners; + g_stage_owners = owner; + stage_owners_unlock(); + return true; +} + +/* Release ownership of a stage that no longer exists under this name. A path + * this process never registered is a no-op. */ +static void stage_owner_release(const char *stage_path) { + if (!stage_path) { + return; + } + stage_owner_t *found = NULL; + stage_owners_lock(); + for (stage_owner_t **link = &g_stage_owners; *link; link = &(*link)->next) { + if (strcmp((*link)->stage_path, stage_path) == 0) { + found = *link; + *link = found->next; + break; + } + } + stage_owners_unlock(); + if (!found) { + return; + } + cbm_pipeline_stage_lock_drop(found->stage_path, found->lock_fd); + free(found->stage_path); + free(found); +} + +/* Remove a stage's main file and SQLite sidecars, keeping ownership. */ +static void remove_stage_files(const char *stage_path) { + (void)cbm_unlink(stage_path); + (void)cbm_remove_db_sidecars(stage_path); +} + static void discard_generation_stage(const char *stage_path) { if (!stage_path) { return; } - cbm_unlink(stage_path); - cbm_remove_db_sidecars(stage_path); + remove_stage_files(stage_path); + stage_owner_release(stage_path); } typedef struct { @@ -1930,6 +2066,7 @@ int cbm_pipeline_finalize_staged_generation(char *stage_path, const char *final_ discard_generation_stage(stage_path); return CBM_PIPELINE_PERSIST_FAILED; } + stage_owner_release(stage_path); cbm_log_info("finalize.timing", "block", "rename", "elapsed_ms", itoa_buf((int)elapsed_ms(t_fin))); return 0; @@ -2387,8 +2524,8 @@ static void cleanup_staging_db(const char *path) { if (!path) { return; } - (void)cbm_unlink(path); - (void)cbm_remove_db_sidecars(path); + remove_stage_files(path); + stage_owner_release(path); } static bool ensure_db_parent(const char *path) { @@ -2495,6 +2632,11 @@ static char *create_staging_path(const char *final_path) { #else close(fd); #endif + if (!stage_owner_register(path)) { + (void)cbm_unlink(path); + free(path); + return NULL; + } return path; } @@ -2622,7 +2764,10 @@ int cbm_pipeline_run(cbm_pipeline_t *p) { if (!backup_succeeded) { cbm_log_warn("pipeline.stage", "action", "backup_failed_full_rebuild", "path", final_path); - cleanup_staging_db(staging_path); + /* The copy is gone but the NAME stays ours: the rebuilt + * generation is renamed over it by the inner finalize and then + * published from it below, so its lock is held to the end. */ + remove_stage_files(staging_path); } } @@ -2705,6 +2850,7 @@ int cbm_pipeline_run(cbm_pipeline_t *p) { return CBM_PIPELINE_PERSIST_FAILED; } + stage_owner_release(staging_path); rc = export_after_publish(p, final_path); free(staging_path); free(final_path); diff --git a/src/pipeline/pipeline_internal.h b/src/pipeline/pipeline_internal.h index d749c695a..186b8ee66 100644 --- a/src/pipeline/pipeline_internal.h +++ b/src/pipeline/pipeline_internal.h @@ -792,6 +792,17 @@ int cbm_pipeline_publish_staged(char *stage_path, const cbm_pipeline_generation_ * executor; the dump path uses it internally). malloc'd, caller frees. */ char *cbm_pipeline_create_staging_path(const char *final_path); +/* Stage ownership (#1839). Every stage minted by cbm_pipeline_create_staging_path + * is owned through an exclusive kernel lock on the sidecar ".lock" for + * as long as the stage exists; the lock -- and the sidecar -- go away when the + * stage is discarded or renamed into place, and the kernel drops the lock + * when the writer dies. Hold/drop are the same primitive, exposed so a test + * can stand in for a live writer. hold returns a descriptor >= 0, or -1 when + * another holder is live or the sidecar cannot be created. drop releases the + * lock and unlinks the sidecar. */ +int cbm_pipeline_stage_lock_hold(const char *stage_path); +void cbm_pipeline_stage_lock_drop(const char *stage_path, int lock_fd); + /* ── Delta-repair staging primitives (pipeline_delta.c) ────────── * Closure-route-only subsystem: clone the live generation, patch exactly * the repaired node/edge set, publish through the shared finalize leg. */ diff --git a/tests/test_pipeline.c b/tests/test_pipeline.c index e724736e2..96857efb8 100644 --- a/tests/test_pipeline.c +++ b/tests/test_pipeline.c @@ -2278,7 +2278,7 @@ TEST(pipeline_complexity_props_independent_of_worker_order) { ASSERT_EQ(sequential_rc, 0); ASSERT_NOT_NULL(sequential_sig); ASSERT_GTE(sequential_funcs, copied); /* at least the one function per fixture file */ - ASSERT_TRUE(cycles_detected); /* the cycles must reach the pass at all */ + ASSERT_TRUE(cycles_detected); /* the cycles must reach the pass at all */ if (mismatch_run >= 0) { printf("\n parallel run %d diverges from sequential: %s\n", mismatch_run, diff); FAIL("complexity props depend on worker id order"); @@ -3496,6 +3496,59 @@ TEST(pipeline_stage_names_never_nest) { PASS(); } +static bool path_exists(const char *path) { + cbm_path_info_t info; + return cbm_path_info_utf8(path, &info) == CBM_PATH_INFO_OK; +} + +/* #1839: a minted stage is OWNED through an exclusive kernel lock on its + * ".lock" sidecar for exactly as long as the stage exists. A second + * holder cannot take it while the writer is live; discarding the stage frees + * the name and removes the sidecar, so nothing stays behind. */ +TEST(pipeline_minted_stage_is_owned_until_released) { + char tmp[256]; + snprintf(tmp, sizeof(tmp), "/tmp/cbm_stage_owner_XXXXXX"); + ASSERT_NOT_NULL(cbm_mkdtemp(tmp)); + char db_path[512]; + snprintf(db_path, sizeof(db_path), "%s/generation.db", tmp); + + char *stage = cbm_pipeline_create_staging_path(db_path); + ASSERT_NOT_NULL(stage); + char lock_path[600]; + snprintf(lock_path, sizeof(lock_path), "%s.lock", stage); + bool lock_present_while_live = path_exists(lock_path); + int hold_while_live = cbm_pipeline_stage_lock_hold(stage); + if (hold_while_live >= 0) { + cbm_pipeline_stage_lock_drop(stage, hold_while_live); + } + + cbm_pipeline_discard_stage(stage); + bool stage_gone = !path_exists(stage); + bool lock_gone = !path_exists(lock_path); + + int hold_after_discard = cbm_pipeline_stage_lock_hold(stage); + int second_holder = cbm_pipeline_stage_lock_hold(stage); + if (second_holder >= 0) { + cbm_pipeline_stage_lock_drop(stage, second_holder); + } + cbm_pipeline_stage_lock_drop(stage, hold_after_discard); + int hold_after_drop = cbm_pipeline_stage_lock_hold(stage); + cbm_pipeline_stage_lock_drop(stage, hold_after_drop); + int leftovers = count_generation_stage_artifacts(tmp, "generation.db"); + free(stage); + th_rmtree(tmp); + + ASSERT_TRUE(lock_present_while_live); + ASSERT_EQ(hold_while_live, -1); + ASSERT_TRUE(stage_gone); + ASSERT_TRUE(lock_gone); + ASSERT_TRUE(hold_after_discard >= 0); + ASSERT_EQ(second_holder, -1); + ASSERT_TRUE(hold_after_drop >= 0); + ASSERT_EQ(leftovers, 0); + PASS(); +} + /* Discovery and extraction must describe the same immutable generation. A * source file created after extraction is not present in the original file * list, so merely re-hashing that list cannot detect the race. Publication @@ -7217,19 +7270,18 @@ TEST(pipeline_python_cross_module_call) { * unique_name (candidates==1) is #1572 and is not this claim. */ TEST(pipeline_cross_language_same_name_does_not_share_calls_issue725) { const char *files[] = {"store.py", "app.py", "web/src/pages/Editor.js"}; - const char *contents[] = { - "class Store:\n" - " def commit(self):\n" - " return True\n", + const char *contents[] = {"class Store:\n" + " def commit(self):\n" + " return True\n", - "from store import Store\n" - "\n" - "def save():\n" - " return Store().commit()\n", + "from store import Store\n" + "\n" + "def save():\n" + " return Store().commit()\n", - "export function commit() {\n" - " return 1;\n" - "}\n"}; + "export function commit() {\n" + " return 1;\n" + "}\n"}; if (setup_lang_repo(files, contents, 3) != 0) FAIL("tmpdir"); @@ -13644,7 +13696,6 @@ TEST(pipeline_delta_patch_indexes_docstring_into_fts_body) { PASS(); } - /* End-to-end for #518/#519: source → docstring → properties JSON → nodes_fts * `body` → findable. Each layer has its own test; this one proves they connect. * It is also the guard on the size budget: build_def_props drops an oversized @@ -14201,6 +14252,7 @@ SUITE(pipeline_semantic_manifest_repro) { RUN_TEST(pipeline_global_extension_config_change_forces_full); RUN_TEST(pipeline_publication_never_uses_a_predictable_staging_path); RUN_TEST(pipeline_stage_names_never_nest); + RUN_TEST(pipeline_minted_stage_is_owned_until_released); RUN_TEST(pipeline_source_mutation_before_publication_preserves_previous_generation); RUN_TEST(pipeline_source_addition_before_publication_preserves_previous_generation); RUN_TEST(pipeline_tsconfig_mutation_before_publication_preserves_previous_generation); From b6458f924da7ac1dc1fc2d7969bc7a3f2ca4d248 Mon Sep 17 00:00:00 2001 From: Martin Vogel Date: Tue, 8 Sep 2026 01:50:32 +0200 Subject: [PATCH 3/5] fix(pipeline): sweep the orphaned stages of a dead writer before each run Nothing on the worker-death path ever cleaned a stage up: the supervisor reaps only an exit status, and a cancel is SIGTERM then SIGKILL after one second, which a gigabyte backup or clone never finishes inside. Every daemon-side cancel or OOM kill therefore left a full-size .stage.X (plus -wal/-shm) behind for good, and with auto_index a large project filled the disk with copies of itself (#1839). Each run now sweeps before minting its own stage. Only names of the exact minted shape for THIS database are considered -- ".stage.<6 alphanumerics>" and that stage's -wal/-shm/-journal/.lock sidecars -- never the live database, a quarantined .corrupt, or another project's files. Liveness is the ownership lock and nothing else: a stage whose sidecar lock can be taken has no writer and is removed (logged as pipeline.stage action=orphan_removed bytes=N); one whose lock a live writer holds is kept (orphan_kept reason=live_writer). No pid, no mtime, no grace period, so the sweep is deterministic and never races a writer. A stage minted by an older binary carries no lock and is swept too; that writer's final rename then fails and it discards its stage. The live database is never named on either path. Co-Authored-By: Claude Fable 5.1 Signed-off-by: Martin Vogel --- src/pipeline/pipeline.c | 185 ++++++++++++++++++++++++++++++++++++++++ tests/test_pipeline.c | 160 ++++++++++++++++++++++++++++++++++ 2 files changed, 345 insertions(+) diff --git a/src/pipeline/pipeline.c b/src/pipeline/pipeline.c index 59ff88837..3d600034e 100644 --- a/src/pipeline/pipeline.c +++ b/src/pipeline/pipeline.c @@ -2741,6 +2741,190 @@ static int export_after_publish(cbm_pipeline_t *p, const char *final_path) { return 0; } +/* ── Orphan sweep (#1839) ──────────────────────────────────────── + * + * Nothing on the worker-death path ever cleaned a stage up, so the sweep + * runs at the start of every run, before this run mints its own stage. It + * considers ONLY names of the exact minted shape for THIS database -- + * ".stage.<6 alphanumerics>" plus that stage's -wal/-shm/-journal + * and .lock sidecars -- never the live database, a quarantined .corrupt, or + * another project's files. A stage is removed when its ownership lock can be + * taken (its writer is dead, or the stage predates ownership) and kept when + * a live writer holds the lock. The pre-ownership case is the one honest + * gap: a stage an OLDER binary is still writing against this database has + * no lock and is swept; that writer's final rename then fails and it + * discards. The live database is never named here on either path. */ + +static const char *const cbm_stage_sidecar_tails[] = {"", "-wal", "-shm", "-journal", ".lock"}; + +/* If `name` is ".stage.<6 alphanumerics>", return the + * length of the stage name proper (without the tail); 0 otherwise. */ +static size_t stage_entry_stage_length(const char *name, const char *base, size_t base_len) { + if (strncmp(name, base, base_len) != 0) { + return 0; + } + const char *at = name + base_len; + if (strncmp(at, cbm_stage_marker, sizeof(cbm_stage_marker) - 1) != 0) { + return 0; + } + at += sizeof(cbm_stage_marker) - 1; + for (int i = 0; i < CBM_STAGE_SUFFIX_RANDOM_CHARS; i++) { + /* NUL is not alphanumeric, so a short name fails here too. */ + if (!isalnum((unsigned char)at[i])) { + return 0; + } + } + at += CBM_STAGE_SUFFIX_RANDOM_CHARS; + for (size_t i = 0; i < sizeof(cbm_stage_sidecar_tails) / sizeof(cbm_stage_sidecar_tails[0]); + i++) { + if (strcmp(at, cbm_stage_sidecar_tails[i]) == 0) { + return (size_t)(at - name); + } + } + return 0; +} + +typedef struct { + char **names; + int count; + int cap; +} stage_name_list_t; + +/* Add a stage name once, however many of its files were listed. */ +static void stage_name_list_add(stage_name_list_t *list, const char *name, size_t len) { + for (int i = 0; i < list->count; i++) { + if (strlen(list->names[i]) == len && strncmp(list->names[i], name, len) == 0) { + return; + } + } + if (list->count == list->cap) { + int cap = list->cap ? list->cap * 2 : 8; + char **grown = (char **)realloc(list->names, (size_t)cap * sizeof(*grown)); + if (!grown) { + return; + } + list->names = grown; + list->cap = cap; + } + char *copy = (char *)malloc(len + 1); + if (!copy) { + return; + } + memcpy(copy, name, len); + copy[len] = '\0'; + list->names[list->count++] = copy; +} + +static int64_t stage_bytes_on_disk(const char *stage_path) { + static const char *const files[] = {"", "-wal", "-shm", "-journal"}; + int64_t total = 0; + char side[CBM_SZ_4K]; + for (size_t i = 0; i < sizeof(files) / sizeof(files[0]); i++) { + int n = snprintf(side, sizeof(side), "%s%s", stage_path, files[i]); + if (n <= 0 || (size_t)n >= sizeof(side)) { + continue; + } + cbm_path_info_t info; + if (cbm_path_info_utf8(side, &info) == CBM_PATH_INFO_OK && info.is_regular) { + total += info.size; + } + } + return total; +} + +static void sweep_one_stage(const char *stage_path) { + char *sidecar = stage_lock_sidecar_path(stage_path); + if (!sidecar) { + return; + } + errno = 0; + int lock_fd = cbm_lockfile_open(sidecar, false); + int probe_errno = errno; + free(sidecar); + if (lock_fd < 0 && probe_errno != ENOENT) { + bool live = probe_errno == EAGAIN || probe_errno == EACCES; +#if defined(EWOULDBLOCK) && EWOULDBLOCK != EAGAIN + live = live || probe_errno == EWOULDBLOCK; +#endif + if (live) { + cbm_log_info("pipeline.stage", "action", "orphan_kept", "reason", "live_writer", "path", + stage_path); + return; + } + char errno_text[16]; + (void)snprintf(errno_text, sizeof(errno_text), "%d", probe_errno); + cbm_log_warn("pipeline.stage", "action", "orphan_kept", "reason", "lock_probe_failed", + "errno", errno_text, "path", stage_path); + return; + } + /* No owner: absent sidecar (pre-ownership stage, or dropped) or a lock + * the kernel released with its writer. Ours now, from the lock down. */ + int64_t bytes = stage_bytes_on_disk(stage_path); + remove_stage_files(stage_path); + if (lock_fd >= 0) { + cbm_pipeline_stage_lock_drop(stage_path, lock_fd); + } + char bytes_text[32]; + (void)snprintf(bytes_text, sizeof(bytes_text), "%lld", (long long)bytes); + cbm_log_info("pipeline.stage", "action", "orphan_removed", "bytes", bytes_text, "path", + stage_path); +} + +static void sweep_orphan_stages(const char *final_path) { + /* Directory part INCLUDING its trailing separator, so the stage paths + * are joined exactly as the final path was spelled. */ + size_t prefix_len = 0; + for (const char *c = final_path; *c; c++) { + if (*c == '/' +#ifdef _WIN32 + || *c == '\\' +#endif + ) { + prefix_len = (size_t)(c - final_path) + 1; + } + } + const char *base = final_path + prefix_len; + size_t base_len = strlen(base); + if (base_len == 0) { + return; + } + char *dir_path = prefix_len ? (char *)malloc(prefix_len + 1) : strdup("."); + if (!dir_path) { + return; + } + if (prefix_len) { + memcpy(dir_path, final_path, prefix_len); + dir_path[prefix_len] = '\0'; + } + cbm_dir_t *dir = cbm_opendir(dir_path); + if (!dir) { + free(dir_path); + return; + } + stage_name_list_t list = {0}; + cbm_dirent_t *entry; + while ((entry = cbm_readdir(dir)) != NULL) { + size_t stage_len = stage_entry_stage_length(entry->name, base, base_len); + if (stage_len) { + stage_name_list_add(&list, entry->name, stage_len); + } + } + cbm_closedir(dir); + for (int i = 0; i < list.count; i++) { + size_t name_len = strlen(list.names[i]); + char *stage_path = (char *)malloc(prefix_len + name_len + 1); + if (stage_path) { + memcpy(stage_path, final_path, prefix_len); + memcpy(stage_path + prefix_len, list.names[i], name_len + 1); + sweep_one_stage(stage_path); + free(stage_path); + } + free(list.names[i]); + } + free(list.names); + free(dir_path); +} + int cbm_pipeline_run(cbm_pipeline_t *p) { if (!p) { return CBM_NOT_FOUND; @@ -2752,6 +2936,7 @@ int cbm_pipeline_run(cbm_pipeline_t *p) { } struct stat final_st; bool final_existed = stat(final_path, &final_st) == 0; + sweep_orphan_stages(final_path); char *staging_path = create_staging_path(final_path); if (!staging_path) { free(final_path); diff --git a/tests/test_pipeline.c b/tests/test_pipeline.c index 96857efb8..9ee54b41b 100644 --- a/tests/test_pipeline.c +++ b/tests/test_pipeline.c @@ -3549,6 +3549,164 @@ TEST(pipeline_minted_stage_is_owned_until_released) { PASS(); } +static bool file_has_content(const char *path, const char *expected) { + FILE *f = cbm_fopen(path, "rb"); + if (!f) { + return false; + } + char buf[256] = {0}; + size_t n = fread(buf, 1, sizeof(buf) - 1, f); + (void)fclose(f); + return n == strlen(expected) && memcmp(buf, expected, n) == 0; +} + +/* #1839 / #1864: a stage a dead writer left beside a VALID database is swept + * by the next run and never influences its route. The 0-byte shape is what a + * worker killed before its backup wrote a page leaves behind; a stale -shm + * beside it is swept with it. */ +TEST(pipeline_stale_zero_byte_stage_beside_valid_db_routes_incremental_and_is_swept) { + char tmp[256]; + snprintf(tmp, sizeof(tmp), "/tmp/cbm_stale_stage_XXXXXX"); + ASSERT_NOT_NULL(cbm_mkdtemp(tmp)); + write_temp_file(tmp, "generation.py", "def StableGeneration():\n return 1\n"); + char db_path[512]; + snprintf(db_path, sizeof(db_path), "%s/generation.db", tmp); + + cbm_pipeline_incremental_test_reset_faults(); + cbm_pipeline_t *baseline = cbm_pipeline_new(tmp, db_path, CBM_MODE_FULL); + ASSERT_NOT_NULL(baseline); + ASSERT_EQ(cbm_pipeline_run(baseline), 0); + char project[256]; + snprintf(project, sizeof(project), "%s", cbm_pipeline_project_name(baseline)); + cbm_pipeline_free(baseline); + + char stale_stage[600]; + char stale_shm[600]; + snprintf(stale_stage, sizeof(stale_stage), "%s.stage.deadbe", db_path); + snprintf(stale_shm, sizeof(stale_shm), "%s.stage.deadbe-shm", db_path); + ASSERT_EQ(th_write_file(stale_stage, ""), 0); + ASSERT_EQ(th_write_file(stale_shm, "stale-shm"), 0); + + /* A body-only change: no added names, so the planner may repair. */ + write_temp_file(tmp, "generation.py", "def StableGeneration():\n return 2\n"); + cbm_pipeline_incremental_test_reset_faults(); + cbm_pipeline_t *incr = cbm_pipeline_new(tmp, db_path, CBM_MODE_FULL); + ASSERT_NOT_NULL(incr); + int incr_rc = cbm_pipeline_run(incr); + cbm_incremental_route_t route = cbm_pipeline_incremental_test_last_route(); + cbm_pipeline_free(incr); + + bool stale_stage_gone = !path_exists(stale_stage); + bool stale_shm_gone = !path_exists(stale_shm); + int stage_count = count_generation_stage_artifacts(tmp, "generation.db"); + int stable_count = -1; + int absent_count = -1; + observe_named_generation(db_path, project, "StableGeneration", "NeverDefined", &stable_count, + &absent_count); + cbm_pipeline_incremental_test_reset_faults(); + th_rmtree(tmp); + + ASSERT_EQ(incr_rc, 0); + ASSERT_TRUE(route != CBM_INCREMENTAL_ROUTE_FORCED_FULL); + ASSERT_TRUE(route != CBM_INCREMENTAL_ROUTE_NONE); + ASSERT_TRUE(stale_stage_gone); + ASSERT_TRUE(stale_shm_gone); + ASSERT_EQ(stage_count, 0); + ASSERT_EQ(stable_count, 1); + ASSERT_EQ(absent_count, 0); + PASS(); +} + +/* #1839: the sweep removes exactly the stages nobody owns. A dead writer's + * stage (main, -wal, and the unlocked .lock sidecar its death left) goes; a + * stage whose writer is LIVE -- here the test, holding its lock -- is kept + * byte for byte, and goes only once that lock is dropped. No timing: liveness + * is the kernel lock, nothing else. */ +TEST(pipeline_sweep_removes_dead_writer_stage_keeps_locked_stage) { + char tmp[256]; + snprintf(tmp, sizeof(tmp), "/tmp/cbm_stage_sweep_XXXXXX"); + ASSERT_NOT_NULL(cbm_mkdtemp(tmp)); + write_temp_file(tmp, "generation.py", "def StableGeneration():\n return 1\n"); + char db_path[512]; + snprintf(db_path, sizeof(db_path), "%s/generation.db", tmp); + + cbm_pipeline_incremental_test_reset_faults(); + cbm_pipeline_t *baseline = cbm_pipeline_new(tmp, db_path, CBM_MODE_FULL); + ASSERT_NOT_NULL(baseline); + ASSERT_EQ(cbm_pipeline_run(baseline), 0); + char project[256]; + snprintf(project, sizeof(project), "%s", cbm_pipeline_project_name(baseline)); + cbm_pipeline_free(baseline); + + char dead_stage[600]; + char dead_wal[600]; + char dead_lock[600]; + char live_stage[600]; + char live_lock[600]; + snprintf(dead_stage, sizeof(dead_stage), "%s.stage.aaaaaa", db_path); + snprintf(dead_wal, sizeof(dead_wal), "%s.stage.aaaaaa-wal", db_path); + snprintf(dead_lock, sizeof(dead_lock), "%s.stage.aaaaaa.lock", db_path); + snprintf(live_stage, sizeof(live_stage), "%s.stage.bbbbbb", db_path); + snprintf(live_lock, sizeof(live_lock), "%s.stage.bbbbbb.lock", db_path); + static const char live_bytes[] = "live-stage-bytes"; + ASSERT_EQ(th_write_file(dead_stage, "dead-stage-bytes"), 0); + ASSERT_EQ(th_write_file(dead_wal, "dead-wal"), 0); + ASSERT_EQ(th_write_file(dead_lock, ""), 0); + ASSERT_EQ(th_write_file(live_stage, live_bytes), 0); + int live_fd = cbm_pipeline_stage_lock_hold(live_stage); + ASSERT_TRUE(live_fd >= 0); + + write_temp_file(tmp, "generation.py", "def StableGeneration():\n return 2\n"); + cbm_pipeline_incremental_test_reset_faults(); + cbm_pipeline_t *first = cbm_pipeline_new(tmp, db_path, CBM_MODE_FULL); + ASSERT_NOT_NULL(first); + int first_rc = cbm_pipeline_run(first); + cbm_pipeline_free(first); + + bool dead_stage_gone = !path_exists(dead_stage); + bool dead_wal_gone = !path_exists(dead_wal); + bool dead_lock_gone = !path_exists(dead_lock); + bool live_kept = file_has_content(live_stage, live_bytes); + bool live_lock_kept = path_exists(live_lock); + int stable_after_first = -1; + int absent_after_first = -1; + observe_named_generation(db_path, project, "StableGeneration", "NeverDefined", + &stable_after_first, &absent_after_first); + + cbm_pipeline_stage_lock_drop(live_stage, live_fd); + cbm_pipeline_incremental_test_reset_faults(); + cbm_pipeline_t *second = cbm_pipeline_new(tmp, db_path, CBM_MODE_FULL); + ASSERT_NOT_NULL(second); + int second_rc = cbm_pipeline_run(second); + cbm_pipeline_free(second); + + bool live_gone = !path_exists(live_stage); + bool live_lock_gone = !path_exists(live_lock); + int stage_count = count_generation_stage_artifacts(tmp, "generation.db"); + int stable_after_second = -1; + int absent_after_second = -1; + observe_named_generation(db_path, project, "StableGeneration", "NeverDefined", + &stable_after_second, &absent_after_second); + cbm_pipeline_incremental_test_reset_faults(); + th_rmtree(tmp); + + ASSERT_EQ(first_rc, 0); + ASSERT_TRUE(dead_stage_gone); + ASSERT_TRUE(dead_wal_gone); + ASSERT_TRUE(dead_lock_gone); + ASSERT_TRUE(live_kept); + ASSERT_TRUE(live_lock_kept); + ASSERT_EQ(stable_after_first, 1); + ASSERT_EQ(absent_after_first, 0); + ASSERT_EQ(second_rc, 0); + ASSERT_TRUE(live_gone); + ASSERT_TRUE(live_lock_gone); + ASSERT_EQ(stage_count, 0); + ASSERT_EQ(stable_after_second, 1); + ASSERT_EQ(absent_after_second, 0); + PASS(); +} + /* Discovery and extraction must describe the same immutable generation. A * source file created after extraction is not present in the original file * list, so merely re-hashing that list cannot detect the race. Publication @@ -14253,6 +14411,8 @@ SUITE(pipeline_semantic_manifest_repro) { RUN_TEST(pipeline_publication_never_uses_a_predictable_staging_path); RUN_TEST(pipeline_stage_names_never_nest); RUN_TEST(pipeline_minted_stage_is_owned_until_released); + RUN_TEST(pipeline_stale_zero_byte_stage_beside_valid_db_routes_incremental_and_is_swept); + RUN_TEST(pipeline_sweep_removes_dead_writer_stage_keeps_locked_stage); RUN_TEST(pipeline_source_mutation_before_publication_preserves_previous_generation); RUN_TEST(pipeline_source_addition_before_publication_preserves_previous_generation); RUN_TEST(pipeline_tsconfig_mutation_before_publication_preserves_previous_generation); From 7b4eb4a00ac600060be4836ee4e1bbfa7bbedf24 Mon Sep 17 00:00:00 2001 From: Martin Vogel Date: Tue, 8 Sep 2026 02:02:08 +0200 Subject: [PATCH 4/5] fix(pipeline): route a first index as no_existing_db, not invalid_existing_db The route probe opened whatever resolve_db_path returned, which inside the staged run is the run's own stage. On a first index nothing was copied into it, so the probe opened a 0-byte mkstemp placeholder, failed its integrity check, and warned "pipeline.route path=full reason=invalid_existing_db" -- on every first index of every project. Read back after an OOM kill it looked like a corrupted database and a daily "corrupt -> full rebuild -> OOM" loop (#1864); the loop is "no database -> full -> OOM", and the stale 0-byte stage never influenced the next run at all (each run mints a fresh name). cbm_pipeline_run now records whether the destination existed and was copied into the stage, and the route decision returns full early with reason=no_existing_db (or existing_db_backup_failed, after the backup_failed_full_rebuild warning) without probing a placeholder. "invalid_existing_db" stays reserved for a real copy that fails its integrity check. Co-Authored-By: Claude Fable 5.1 Signed-off-by: Martin Vogel --- src/pipeline/pipeline.c | 21 ++++++++++++++ tests/test_pipeline.c | 61 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 82 insertions(+) diff --git a/src/pipeline/pipeline.c b/src/pipeline/pipeline.c index 3d600034e..5d21a1bde 100644 --- a/src/pipeline/pipeline.c +++ b/src/pipeline/pipeline.c @@ -196,6 +196,15 @@ struct cbm_pipeline { * full rebuild, so the MCP response can surface the migration. */ bool format_migration; + /* Recorded by cbm_pipeline_run for the staged run beneath it: whether + * the destination existed, and whether it was copied into the stage so + * that an incremental route has a real previous generation to work + * from. Without a copy the stage is the run's empty placeholder, and + * probing THAT for integrity is what reported every first index as + * "invalid_existing_db" (#1864). */ + bool final_existed; + bool existing_generation; + /* ADR (project_summaries) captured before a full-reindex DB delete, so it * can be restored after the rebuild. NULL when no ADR existed. Issue #516. */ char *saved_adr; @@ -1403,6 +1412,16 @@ static int capture_existing_adr(cbm_pipeline_t *p, const char *db_path) { static int try_incremental_or_delete_db(cbm_pipeline_t *p, cbm_file_info_t *files, int file_count, const cbm_file_hash_t *baseline_manifest, int baseline_count, bool force_full_on_mismatch) { + if (!p->existing_generation) { + /* Nothing to be incremental against: a first index, or a + * destination that could not be copied (already reported as + * backup_failed_full_rebuild). The stage is an empty placeholder, + * not a database, so it is not probed -- "invalid_existing_db" + * stays reserved for a real copy that fails its integrity check. */ + cbm_log_info("pipeline.route", "path", "full", "reason", + p->final_existed ? "existing_db_backup_failed" : "no_existing_db"); + return CBM_PIPELINE_FORCE_FULL_REINDEX; + } char *db_path = resolve_db_path(p); if (!db_path) { return CBM_PIPELINE_FORCE_FULL_REINDEX; @@ -2956,6 +2975,8 @@ int cbm_pipeline_run(cbm_pipeline_t *p) { } } + p->final_existed = final_existed; + p->existing_generation = final_existed && backup_succeeded; char *configured_db_path = p->db_path; p->db_path = strdup(staging_path); if (!p->db_path) { diff --git a/tests/test_pipeline.c b/tests/test_pipeline.c index 9ee54b41b..0892fd696 100644 --- a/tests/test_pipeline.c +++ b/tests/test_pipeline.c @@ -3707,6 +3707,66 @@ TEST(pipeline_sweep_removes_dead_writer_stage_keeps_locked_stage) { PASS(); } +static char g_route_log_capture[8192]; +static atomic_flag g_route_log_spin = ATOMIC_FLAG_INIT; + +/* Keeps only the route decisions; worker threads log too, and only the + * appends need serialising. */ +static void capture_route_log_sink(const char *line) { + if (!line || !strstr(line, "pipeline.route")) { + return; + } + while (atomic_flag_test_and_set_explicit(&g_route_log_spin, memory_order_acquire)) {} + size_t used = strlen(g_route_log_capture); + size_t avail = sizeof(g_route_log_capture) - used; + if (avail > 1) { + int n = snprintf(g_route_log_capture + used, avail, "%s\n", line); + if (n < 0 || (size_t)n >= avail) { + g_route_log_capture[sizeof(g_route_log_capture) - 1] = '\0'; + } + } + atomic_flag_clear_explicit(&g_route_log_spin, memory_order_release); +} + +/* #1864: a first index has no previous generation. The route probe used to + * open the run's own EMPTY stage placeholder, fail its integrity check, and + * warn "reason=invalid_existing_db" on every first index of every project, + * which the report read as a corrupted database and a crash loop. A fresh + * index says what it is. */ +TEST(pipeline_fresh_index_never_reports_invalid_existing_db) { + char tmp[256]; + snprintf(tmp, sizeof(tmp), "/tmp/cbm_fresh_route_XXXXXX"); + ASSERT_NOT_NULL(cbm_mkdtemp(tmp)); + write_temp_file(tmp, "generation.py", "def StableGeneration():\n return 1\n"); + char db_path[512]; + snprintf(db_path, sizeof(db_path), "%s/generation.db", tmp); + + g_route_log_capture[0] = '\0'; + CBMLogLevel previous_level = cbm_log_get_level(); + cbm_log_set_level(CBM_LOG_DEBUG); + cbm_log_set_sink(capture_route_log_sink); + cbm_pipeline_incremental_test_reset_faults(); + cbm_pipeline_t *fresh = cbm_pipeline_new(tmp, db_path, CBM_MODE_FULL); + int fresh_rc = fresh ? cbm_pipeline_run(fresh) : -1; + cbm_pipeline_free(fresh); + cbm_log_set_sink(NULL); + cbm_log_set_level(previous_level); + + bool invalid_reported = strstr(g_route_log_capture, "invalid_existing_db") != NULL; + bool fresh_reported = strstr(g_route_log_capture, "reason=no_existing_db") != NULL; + bool db_present = path_exists(db_path); + int stage_count = count_generation_stage_artifacts(tmp, "generation.db"); + cbm_pipeline_incremental_test_reset_faults(); + th_rmtree(tmp); + + ASSERT_EQ(fresh_rc, 0); + ASSERT_TRUE(!invalid_reported); + ASSERT_TRUE(fresh_reported); + ASSERT_TRUE(db_present); + ASSERT_EQ(stage_count, 0); + PASS(); +} + /* Discovery and extraction must describe the same immutable generation. A * source file created after extraction is not present in the original file * list, so merely re-hashing that list cannot detect the race. Publication @@ -14413,6 +14473,7 @@ SUITE(pipeline_semantic_manifest_repro) { RUN_TEST(pipeline_minted_stage_is_owned_until_released); RUN_TEST(pipeline_stale_zero_byte_stage_beside_valid_db_routes_incremental_and_is_swept); RUN_TEST(pipeline_sweep_removes_dead_writer_stage_keeps_locked_stage); + RUN_TEST(pipeline_fresh_index_never_reports_invalid_existing_db); RUN_TEST(pipeline_source_mutation_before_publication_preserves_previous_generation); RUN_TEST(pipeline_source_addition_before_publication_preserves_previous_generation); RUN_TEST(pipeline_tsconfig_mutation_before_publication_preserves_previous_generation); From 91e31211df9dd61459617d857d46b71699e34df8 Mon Sep 17 00:00:00 2001 From: Martin Vogel Date: Tue, 8 Sep 2026 22:30:34 +0200 Subject: [PATCH 5/5] =?UTF-8?q?fix(pipeline):=20take=20the=20stage=20lock?= =?UTF-8?q?=20before=20the=20stage=20file=20is=20visible=20(close=20#2111?= =?UTF-8?q?=20create=E2=86=92register=20race)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit create_staging_path() minted the stage's main file with mkstemp and only then took its sidecar .lock via stage_owner_register(). A second, concurrent cbm_pipeline_run() against the same final_path (auto_index, or the stale-rendezvous-recovery retry path) could land its sweep_orphan_stages() in that window: the main file was visible but not yet locked, so the sweep treated it like a genuinely orphaned stage and deleted it out from under the live writer (POSIX), or collided on the not-yet-created sidecar and surfaced EACCES/errno=13 (Windows). That is the #2111 red on test-windows-guards and both test-windows CLANG64 x86_64 shards (index_format / simhash / index_resilience, with pipeline.stage action=lock_failed errno=13). Fix: mint the six-char alphanumeric suffix in create_staging_path() itself, create and hold the sidecar lock FIRST, then create the main file with O_EXCL. The stage's main file is therefore never present on disk without its lock already held, so a racing sweep can only ever find the stage lock-held: it reaches the stage through the .lock entry too (stage_entry_stage_length() matches it), probes the lock, sees a live holder, and keeps it. The create->register TOCTOU is closed by construction. #1839 is preserved: sweep_one_stage() still removes a stage whose sidecar is absent (ENOENT). With live stages now always carrying their sidecar, ENOENT is only ever a genuine pre-lock-era orphan minted by an older binary, which #1839 requires be swept. stage_owner_register() becomes stage_owner_adopt(), recording the already-held descriptor; re-taking the lock would self-conflict, since flock() and Windows _SH_DENYRW deny a second acquire of the same sidecar even from this process. A new after_stage_created test seam and pipeline_concurrent_sweep_must_not_- remove_inflight_stage drive a real concurrent sweep in that window; the test is RED without the lock-first ordering. Co-Authored-By: Claude Fable 5.1 Signed-off-by: Martin Vogel --- src/pipeline/pipeline.c | 151 +++++++++++++++++++++++++------ src/pipeline/pipeline_internal.h | 10 ++ tests/test_pipeline.c | 125 +++++++++++++++++++++++++ 3 files changed, 258 insertions(+), 28 deletions(-) diff --git a/src/pipeline/pipeline.c b/src/pipeline/pipeline.c index 5d21a1bde..dba7fd629 100644 --- a/src/pipeline/pipeline.c +++ b/src/pipeline/pipeline.c @@ -37,6 +37,7 @@ enum { CBM_DIR_PERMS = 0755, PL_RING = 4, PL_RING_MASK = 3, PL_SEQ_PASSES = 6 }; #include "foundation/compat_thread.h" #include "foundation/profile.h" #include "foundation/mem.h" +#include "foundation/secure_random.h" #include #include @@ -73,6 +74,8 @@ static atomic_bool g_persist_test_cancel_after_destination_prepare = false; static atomic_bool g_persist_test_fail_adr_capture = false; static cbm_pipeline_test_hook_fn g_persist_test_before_final_manifest = NULL; static void *g_persist_test_before_final_manifest_userdata = NULL; +static cbm_pipeline_test_hook_fn g_persist_test_after_stage_created = NULL; +static void *g_persist_test_after_stage_created_userdata = NULL; void cbm_pipeline_incremental_test_fail_after_stage_dump_once(void) { atomic_store(&g_persist_test_fail_after_stage_dump, true); @@ -106,6 +109,29 @@ void cbm_pipeline_persist_test_run_before_final_manifest(void) { } } +void cbm_pipeline_incremental_test_after_stage_created_once(cbm_pipeline_test_hook_fn hook, + void *userdata) { + g_persist_test_after_stage_created = hook; + g_persist_test_after_stage_created_userdata = userdata; +} + +/* Fired by create_staging_path() right after the stage's main file is created + * with O_EXCL -- and, in the current lock-before-visible ordering, after its + * sidecar lock is already held. A test hook installed here can run a + * concurrent sweep (another cbm_pipeline_run() against the same final_path) at + * this instant to prove the just-created stage survives it. Under the OLD + * create-then-lock ordering this was the unlocked TOCTOU window, so the same + * hook binds RED if that ordering ever regresses. */ +void cbm_pipeline_persist_test_run_after_stage_created(void) { + cbm_pipeline_test_hook_fn hook = g_persist_test_after_stage_created; + void *userdata = g_persist_test_after_stage_created_userdata; + g_persist_test_after_stage_created = NULL; + g_persist_test_after_stage_created_userdata = NULL; + if (hook) { + hook(userdata); + } +} + bool cbm_pipeline_persist_test_take_failure_after_stage_dump(void) { return atomic_exchange(&g_persist_test_fail_after_stage_dump, false); } @@ -125,6 +151,8 @@ void cbm_pipeline_persist_test_reset_faults(void) { atomic_store(&g_persist_test_fail_adr_capture, false); g_persist_test_before_final_manifest = NULL; g_persist_test_before_final_manifest_userdata = NULL; + g_persist_test_after_stage_created = NULL; + g_persist_test_after_stage_created_userdata = NULL; } #endif @@ -1630,25 +1658,25 @@ void cbm_pipeline_stage_lock_drop(const char *stage_path, int lock_fd) { } } -static bool stage_owner_register(const char *stage_path) { - int fd = cbm_pipeline_stage_lock_hold(stage_path); - if (fd < 0) { - char errno_text[16]; - (void)snprintf(errno_text, sizeof(errno_text), "%d", errno); - cbm_log_warn("pipeline.stage", "action", "lock_failed", "errno", errno_text, "path", - stage_path); - return false; - } +/* Record an already-held stage lock in the per-process owner table, keyed by + * path so publish/finalize/discard can release it later. On success the table + * owns lock_fd; on failure the caller still does and must drop it. + * + * The lock must ALREADY be held: create_staging_path() takes it before the + * stage's main file is created (so the file is never visible on disk without + * its lock), then hands the descriptor here. Re-taking the lock in this helper + * would self-conflict -- both flock() and Windows _SH_DENYRW deny a second + * acquire of the same sidecar even from this same process. */ +static bool stage_owner_adopt(const char *stage_path, int lock_fd) { stage_owner_t *owner = (stage_owner_t *)malloc(sizeof(*owner)); char *path_copy = strdup(stage_path); if (!owner || !path_copy) { free(owner); free(path_copy); - cbm_pipeline_stage_lock_drop(stage_path, fd); return false; } owner->stage_path = path_copy; - owner->lock_fd = fd; + owner->lock_fd = lock_fd; stage_owners_lock(); owner->next = g_stage_owners; g_stage_owners = owner; @@ -2641,22 +2669,79 @@ static char *create_staging_path(const char *final_path) { } memcpy(path, final_path, final_len); memcpy(path + final_len, suffix, sizeof(suffix)); - int fd = cbm_mkstemp(path); - if (fd < 0) { - free(path); - return NULL; - } -#ifdef _WIN32 - _close(fd); -#else - close(fd); + /* The six random chars sit directly after the ".stage." marker; each + * attempt overwrites the "XXXXXX" template in place. Alphanumerics only, + * matching stage_suffix_at()/stage_entry_stage_length() so the sweep + * recognises the minted name and its sidecars. */ + char *random_at = path + final_len + (sizeof(cbm_stage_marker) - 1); + static const char alphabet[] = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"; + + /* Lock BEFORE the stage becomes visible: take the sidecar lock first, then + * create the stage's main file with O_EXCL. The main file is therefore + * never present on disk without its lock already held, so a concurrent + * run's sweep_orphan_stages() against the same final_path can only ever + * find this stage lock-held -- it reaches the stage through the .lock + * sidecar too (stage_entry_stage_length() matches it), probes the lock, + * sees a live holder, and keeps it. That closes the old create->register + * window that let a racing sweep delete a live-but-unlocked stage (POSIX) + * or collide on the sidecar with EACCES (Windows) -- #2111's windows-guards + * red. A pre-lock-era orphan minted by an OLDER binary still carries no + * lock, so the sweep's ENOENT path still removes it (#1839 preserved). + * + * A minted suffix collides with an existing stage only about 1 in 62^6; + * retry a bounded number of times, the way mkstemp/mkdtemp do, then fail. */ + for (int attempt = 0; attempt < 128; attempt++) { + unsigned char rnd[CBM_STAGE_SUFFIX_RANDOM_CHARS]; + if (!cbm_secure_random(rnd, sizeof(rnd))) { + free(path); + errno = EIO; + return NULL; + } + for (size_t i = 0; i < sizeof(rnd); i++) { + random_at[i] = alphabet[rnd[i] % (sizeof(alphabet) - 1)]; + } + errno = 0; + int lock_fd = cbm_pipeline_stage_lock_hold(path); + if (lock_fd < 0) { + /* A live twin already owns this exact suffix's sidecar (EAGAIN / + * EACCES), or the sidecar could not be created. Mint a fresh suffix + * and try again rather than contend for this one. */ + continue; + } + FILE *main_file = cbm_fopen(path, "wbx"); + if (!main_file) { + /* The suffix collided with a lock-less orphan's main file -- its + * sidecar was takeable, so it is not a live writer. Never inherit a + * stranger's bytes: drop the lock, remove the sidecar we just took, + * and mint a fresh suffix. The orphan's main file is left for a + * later sweep, which removes it as a pre-lock-era orphan. */ + cbm_pipeline_stage_lock_drop(path, lock_fd); + continue; + } + (void)fclose(main_file); +#if defined(CBM_INCREMENTAL_TEST_API) && CBM_INCREMENTAL_TEST_API + /* Main file now exists and its lock is already held. Under the OLD + * create-then-lock ordering this was the unlocked window; the + * concurrent-sweep test fires here to prove the stage now survives a + * racing sweep, and to bind RED if that ordering ever regresses. */ + cbm_pipeline_persist_test_run_after_stage_created(); #endif - if (!stage_owner_register(path)) { - (void)cbm_unlink(path); - free(path); - return NULL; - } - return path; + if (!stage_owner_adopt(path, lock_fd)) { + cbm_pipeline_stage_lock_drop(path, lock_fd); + (void)cbm_unlink(path); + free(path); + return NULL; + } + return path; + } + /* Every attempt failed to take a lock -- keep the observability the old + * stage_owner_register() emitted for a lock failure. */ + char errno_text[16]; + (void)snprintf(errno_text, sizeof(errno_text), "%d", errno); + cbm_log_warn("pipeline.stage", "action", "lock_failed", "errno", errno_text, "path", path); + free(path); + errno = EEXIST; + return NULL; } /* A backup-failed destination may still have the only recoverable WAL or @@ -2876,8 +2961,18 @@ static void sweep_one_stage(const char *stage_path) { "errno", errno_text, "path", stage_path); return; } - /* No owner: absent sidecar (pre-ownership stage, or dropped) or a lock - * the kernel released with its writer. Ours now, from the lock down. */ + /* No owner: absent sidecar (an ENOENT probe) or a lock the kernel released + * with its writer. Ours now, from the lock down. + * + * The absent-sidecar (ENOENT) case is exactly a pre-lock-era orphan: a + * stage an OLDER binary minted with no sidecar at all (#1839 pins that + * these ARE swept). It is NOT an in-flight stage of a current run: since + * create_staging_path() now takes the sidecar lock BEFORE the stage's main + * file becomes visible on disk, a live stage always has its sidecar, so a + * concurrent sweep landing here for one would instead find the lock held + * above and keep it. Removing on ENOENT therefore reclaims genuine orphans + * without ever deleting a live stage (the create->register race behind + * #2111's windows-guards red is closed at the source). */ int64_t bytes = stage_bytes_on_disk(stage_path); remove_stage_files(stage_path); if (lock_fd >= 0) { diff --git a/src/pipeline/pipeline_internal.h b/src/pipeline/pipeline_internal.h index 186b8ee66..36bd0cbf6 100644 --- a/src/pipeline/pipeline_internal.h +++ b/src/pipeline/pipeline_internal.h @@ -908,6 +908,15 @@ void cbm_pipeline_incremental_test_fail_adr_capture_once(void); typedef void (*cbm_pipeline_test_hook_fn)(void *userdata); void cbm_pipeline_incremental_test_before_final_manifest_once(cbm_pipeline_test_hook_fn hook, void *userdata); +/* Fires from create_staging_path(), right after the stage's main file is + * created with O_EXCL. In the current lock-before-visible ordering its sidecar + * lock is already held at this point, so a test hook installed here can run a + * concurrent sweep (via another cbm_pipeline_run() against the same + * final_path) and confirm the just-created stage survives it. Under the OLD + * create-then-lock ordering this was the unlocked window, so the hook also + * binds RED if that ordering regresses. */ +void cbm_pipeline_incremental_test_after_stage_created_once(cbm_pipeline_test_hook_fn hook, + void *userdata); cbm_incremental_route_t cbm_pipeline_incremental_test_last_route(void); void cbm_pipeline_incremental_test_reset_faults(void); @@ -917,6 +926,7 @@ bool cbm_pipeline_persist_test_take_failure_after_stage_dump(void); bool cbm_pipeline_persist_test_take_cancel_after_predump(void); bool cbm_pipeline_persist_test_take_cancel_after_destination_prepare(void); void cbm_pipeline_persist_test_run_before_final_manifest(void); +void cbm_pipeline_persist_test_run_after_stage_created(void); void cbm_pipeline_persist_test_reset_faults(void); #endif diff --git a/tests/test_pipeline.c b/tests/test_pipeline.c index 0892fd696..54e2aa6f9 100644 --- a/tests/test_pipeline.c +++ b/tests/test_pipeline.c @@ -3707,6 +3707,130 @@ TEST(pipeline_sweep_removes_dead_writer_stage_keeps_locked_stage) { PASS(); } +/* #2111 create->register TOCTOU guard. The bug this pins: create_staging_path() + * used to make the stage's main file visible (via mkstemp) BEFORE taking its + * sidecar lock, and sweep_orphan_stages() treats an absent lock sidecar + * (ENOENT) as a confirmed-dead writer (kernel released the lock on death). So a + * second, concurrent cbm_pipeline_run() against the SAME final_path, landing + * its own sweep in that narrow unlocked window, removed the first run's + * in-flight stage out from under it: every extraction pass still completed + * (none touch the stage file on disk), but the publish that followed found its + * own stage gone. Two writers racing the same project is a real scenario this + * PR's own sweep exists to clean up after (auto_index; the recently-fixed + * stale-rendezvous-recovery retry path) -- not hypothetical, and exactly the + * shape of #2111's windows-guards red (every pass logged success, nothing was + * ever committed, "Pipeline failed" surfaced generic; on Windows the sidecar + * collision surfaced as EACCES/errno=13). + * + * The fix takes the sidecar lock BEFORE the main file becomes visible, so this + * hook -- fired the instant the main file exists -- finds the stage already + * lock-protected and the racing sweep keeps it. The hook fires at the same + * point under the old ordering, where the lock was NOT yet held, so this test + * goes RED if that ordering ever regresses. + * + * The racing run is cancelled immediately so it never reaches ITS OWN + * publish -- isolating the sweep's effect on the first run's stage from the + * separate question of two full runs both completing for the same project. */ +typedef struct { + const char *tmp_dir; + const char *db_path; + bool stage_survived; +} racing_sweep_arg_t; + +static bool find_sole_stage_path(const char *dir, const char *db_basename, char *out, + size_t out_sz) { + cbm_dir_t *d = cbm_opendir(dir); + if (!d) { + return false; + } + char prefix[256]; + snprintf(prefix, sizeof(prefix), "%s.stage.", db_basename); + size_t prefix_len = strlen(prefix); + bool found = false; + cbm_dirent_t *entry; + while ((entry = cbm_readdir(d)) != NULL) { + size_t name_len = strlen(entry->name); + enum { STAGE_SUFFIX_RANDOM_CHARS = 6 }; /* mirrors CBM_STAGE_SUFFIX_RANDOM_CHARS */ + if (strncmp(entry->name, prefix, prefix_len) == 0 && + name_len == prefix_len + STAGE_SUFFIX_RANDOM_CHARS) { + snprintf(out, out_sz, "%s/%s", dir, entry->name); + found = true; + break; + } + } + cbm_closedir(d); + return found; +} + +static void *racing_sweep_thread(void *arg) { + racing_sweep_arg_t *a = (racing_sweep_arg_t *)arg; + cbm_pipeline_t *p = cbm_pipeline_new(a->tmp_dir, a->db_path, CBM_MODE_FULL); + if (p) { + /* Its own sweep_orphan_stages() runs at the very start, before this + * run mints its own stage -- exactly like the first run's. Cancel + * immediately: this run must reach the sweep and nothing past it. */ + cbm_pipeline_cancel(p); + (void)cbm_pipeline_run(p); + cbm_pipeline_free(p); + } + return NULL; +} + +/* Fired from inside the FIRST run's create_staging_path(), the instant its + * stage main file exists (under the fix, with its sidecar lock already held; + * under the old create-then-lock ordering, before the lock was taken): run a + * second, cancelled cbm_pipeline_run() for the same project synchronously on + * another thread, so its sweep has every chance to reach the stage before + * control returns to the first run. */ +static void racing_sweep_hook(void *userdata) { + racing_sweep_arg_t *a = (racing_sweep_arg_t *)userdata; + char stage_path[600] = {0}; + bool had_stage = + find_sole_stage_path(a->tmp_dir, "generation.db", stage_path, sizeof(stage_path)); + cbm_thread_t tid; + if (cbm_thread_create(&tid, 0, racing_sweep_thread, a) == 0) { + cbm_thread_join(&tid); + } + a->stage_survived = had_stage && path_exists(stage_path); +} + +TEST(pipeline_concurrent_sweep_must_not_remove_inflight_stage) { + char tmp[256]; + snprintf(tmp, sizeof(tmp), "/tmp/cbm_stage_race_XXXXXX"); + ASSERT_NOT_NULL(cbm_mkdtemp(tmp)); + write_temp_file(tmp, "generation.py", "def StableGeneration():\n return 1\n"); + char db_path[512]; + snprintf(db_path, sizeof(db_path), "%s/generation.db", tmp); + + cbm_pipeline_incremental_test_reset_faults(); + racing_sweep_arg_t race_arg = {.tmp_dir = tmp, .db_path = db_path, .stage_survived = false}; + cbm_pipeline_incremental_test_after_stage_created_once(racing_sweep_hook, &race_arg); + + cbm_pipeline_t *p = cbm_pipeline_new(tmp, db_path, CBM_MODE_FULL); + ASSERT_NOT_NULL(p); + char project[256]; + snprintf(project, sizeof(project), "%s", cbm_pipeline_project_name(p)); + int rc = cbm_pipeline_run(p); + cbm_pipeline_free(p); + + bool db_exists = path_exists(db_path); + int defined_count = -1; + int absent_count = -1; + observe_named_generation(db_path, project, "StableGeneration", "NeverDefined", &defined_count, + &absent_count); + int stage_count = count_generation_stage_artifacts(tmp, "generation.db"); + cbm_pipeline_incremental_test_reset_faults(); + th_rmtree(tmp); + + ASSERT_TRUE(race_arg.stage_survived); + ASSERT_EQ(rc, 0); + ASSERT_TRUE(db_exists); + ASSERT_EQ(defined_count, 1); + ASSERT_EQ(absent_count, 0); + ASSERT_EQ(stage_count, 0); + PASS(); +} + static char g_route_log_capture[8192]; static atomic_flag g_route_log_spin = ATOMIC_FLAG_INIT; @@ -14473,6 +14597,7 @@ SUITE(pipeline_semantic_manifest_repro) { RUN_TEST(pipeline_minted_stage_is_owned_until_released); RUN_TEST(pipeline_stale_zero_byte_stage_beside_valid_db_routes_incremental_and_is_swept); RUN_TEST(pipeline_sweep_removes_dead_writer_stage_keeps_locked_stage); + RUN_TEST(pipeline_concurrent_sweep_must_not_remove_inflight_stage); RUN_TEST(pipeline_fresh_index_never_reports_invalid_existing_db); RUN_TEST(pipeline_source_mutation_before_publication_preserves_previous_generation); RUN_TEST(pipeline_source_addition_before_publication_preserves_previous_generation);