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 0ffb46c0b..dba7fd629 100644 --- a/src/pipeline/pipeline.c +++ b/src/pipeline/pipeline.c @@ -37,7 +37,9 @@ 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 #include #include @@ -72,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); @@ -105,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); } @@ -124,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 @@ -195,6 +224,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; @@ -1402,6 +1440,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; @@ -1530,12 +1578,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); + } +} + +/* 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); + return false; + } + owner->stage_path = path_copy; + owner->lock_fd = lock_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 { @@ -1929,6 +2113,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; @@ -2386,8 +2571,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) { @@ -2415,12 +2600,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; } @@ -2440,17 +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 - 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 @@ -2554,6 +2845,200 @@ 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 (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) { + 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; @@ -2565,6 +3050,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); @@ -2577,10 +3063,15 @@ 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); } } + 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) { @@ -2660,6 +3151,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..36bd0cbf6 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. */ @@ -897,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); @@ -906,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 0f6332cb2..54e2aa6f9 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"); @@ -3416,6 +3416,481 @@ 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(); +} + +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(); +} + +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(); +} + +/* #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; + +/* 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 @@ -7137,19 +7612,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"); @@ -13564,7 +14038,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 @@ -14120,6 +14593,12 @@ 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_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); RUN_TEST(pipeline_tsconfig_mutation_before_publication_preserves_previous_generation);