github-featured/deusdata--codebase-memory-mcp
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

fix(pipeline): preserve coverage and artifact ordering

Browse Source 
Keep caller-requested discovery scope while rebuilding changed weaker-mode requests at the stronger stored coverage. Export persistent artifacts only after the replacement database generation is published.

Signed-off-by: astandrik <astandrik@yandex-team.ru>
This commit is contained in:
astandrik
2026-08-17 00:03:37 +03:00
parent c1a5de3bda
commit d024f41e7f
4 changed files with 365 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/pipeline/pipeline.c
+78 -48
View File
@@ -151,6 +151,7 @@ struct cbm_pipeline {
char *project_name;
cbm_git_context_t git_ctx;
char *branch_qn;
cbm_index_mode_t requested_mode;
cbm_index_mode_t mode;
atomic_int cancelled_storage;
atomic_int *cancelled;
@@ -268,6 +269,7 @@ cbm_pipeline_t *cbm_pipeline_new(const char *repo_path, const char *db_path,
p->project_name = cbm_project_name_from_path(repo_path);
(void)cbm_git_context_resolve(repo_path, &p->git_ctx);
p->branch_qn = cbm_git_context_branch_qn(p->project_name, &p->git_ctx);
p->requested_mode = mode;
p->mode = mode;
p->persistence = false;
p->committed_nodes = -1;
@@ -1371,7 +1373,7 @@ static int capture_existing_adr(cbm_pipeline_t *p, const char *db_path) {
* metadata write has succeeded. */
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) {
int baseline_count, bool force_full_on_mismatch) {
char *db_path = resolve_db_path(p);
if (!db_path) {
return CBM_PIPELINE_FORCE_FULL_REINDEX;
@@ -1393,7 +1395,7 @@ static int try_incremental_or_delete_db(cbm_pipeline_t *p, cbm_file_info_t *file
}
cbm_log_info("pipeline.route", "path", "incremental_manifest");
int rc = cbm_pipeline_run_incremental(p, db_path, files, file_count, baseline_manifest,
baseline_count);
baseline_count, force_full_on_mismatch);
/* Delete the existing generation ONLY when we are about to rebuild it.
* On main this was guarded by an early `return rc` for the incremental
* path; this function has no such early return, so the delete must be
@@ -1442,19 +1444,20 @@ static int pipeline_mode_coverage_rank(cbm_index_mode_t mode) {
* must never erase files that the cheaper discovery intentionally skips. The
* exact-manifest pipeline therefore keeps the most comprehensive successfully
* published mode and performs any changed rebuild at that coverage level. */
static void promote_mode_to_existing_coverage(cbm_pipeline_t *p) {
static bool promote_mode_to_existing_coverage(cbm_pipeline_t *p) {
if (!p || !p->project_name) {
return;
return false;
}
char *db_path = resolve_db_path(p);
if (!db_path) {
return;
return false;
}
cbm_store_t *store = cbm_store_open_path_query(db_path);
free(db_path);
if (!store) {
return;
return false;
}
bool promoted = false;
cbm_coverage_meta_t meta = {0};
if (cbm_store_coverage_meta_get(store, p->project_name, &meta) == CBM_STORE_OK &&
meta.index_mode) {
@@ -1470,31 +1473,12 @@ static void promote_mode_to_existing_coverage(cbm_pipeline_t *p) {
cbm_log_info("pipeline.mode", "requested", pipeline_mode_name(p->mode), "effective",
pipeline_mode_name(stored_mode), "reason", "preserve_existing_coverage");
p->mode = stored_mode;
promoted = true;
}
}
cbm_store_coverage_meta_clear(&meta);
cbm_store_close(store);
}
int cbm_pipeline_refresh_artifact(cbm_pipeline_t *p, const char *db_path) {
if (!p || !db_path || !p->repo_path || !p->project_name) {
return 0;
}
bool existing = cbm_artifact_exists(p->repo_path);
if (!p->persistence && !existing) {
return 0;
}
int quality = p->persistence ? CBM_ARTIFACT_BEST : CBM_ARTIFACT_FAST;
int rc = cbm_artifact_export(db_path, p->repo_path, p->project_name, quality);
if (rc != 0) {
const char *err = cbm_artifact_export_last_error();
if (p->persistence) {
cbm_log_error("pipeline.err", "phase", "artifact_export", "err", err ? err : "unknown");
return rc;
}
cbm_log_warn("artifact.refresh_failed", "err", err ? err : "unknown");
}
return 0;
return promoted;
}
/* Defined below, next to the other publication helpers. */
@@ -2014,8 +1998,8 @@ static int dump_and_persist_hashes(cbm_pipeline_t *p, const cbm_file_hash_t *bas
cbm_pipeline_free_semantic_manifest(manifest, manifest_count);
if (rc != 0) {
/* db_path is this function's strdup (resolve_db_path); every return
* must release it -- refresh_artifact below only borrows it. LSan on
* the Linux leg caught exactly this pair of exits leaking. */
* must release it. LSan on the Linux leg caught exactly this pair of
* exits leaking. */
free(db_path);
return rc;
}
@@ -2028,12 +2012,8 @@ static int dump_and_persist_hashes(cbm_pipeline_t *p, const cbm_file_hash_t *bas
free(p->saved_adr);
p->saved_adr = NULL;
/* The SQLite generation is the commit point. Automatic refresh of an
* existing artifact is best-effort, but an explicitly requested artifact
* is caller-visible and must report an export failure. */
int artifact_rc = cbm_pipeline_refresh_artifact(p, db_path);
free(db_path);
return artifact_rc;
return 0;
}
/* Run githistory pass. */
@@ -2157,11 +2137,10 @@ static int run_extraction_phase(cbm_pipeline_t *p, cbm_pipeline_ctx_t *ctx,
return rc;
}
static int cbm_pipeline_run_staged(cbm_pipeline_t *p, bool *was_incremental) {
static int cbm_pipeline_run_staged(cbm_pipeline_t *p) {
if (!p) {
return CBM_NOT_FOUND;
}
*was_incremental = false;
CBM_PROF_START(t_pipeline_total);
struct timespec t0;
@@ -2169,8 +2148,15 @@ static int cbm_pipeline_run_staged(cbm_pipeline_t *p, bool *was_incremental) {
cbm_path_alias_collection_t *path_aliases = NULL;
cbm_file_hash_t *baseline_manifest = NULL;
int baseline_count = 0;
char **requested_excluded_dirs = NULL;
int requested_excluded_count = 0;
cbm_ignored_file_t *requested_ignored_files = NULL;
int requested_ignored_count = 0;
int requested_ignored_total = 0;
bool restore_requested_discovery = false;
promote_mode_to_existing_coverage(p);
p->mode = p->requested_mode;
bool mode_promoted = promote_mode_to_existing_coverage(p);
/* cbm_pipeline_new() may precede the actual run by an arbitrary interval.
* Refresh once here, then use this exact snapshot for both Branch graph
@@ -2193,7 +2179,7 @@ static int cbm_pipeline_run_staged(cbm_pipeline_t *p, bool *was_incremental) {
/* Phase 1: Discover files */
CBM_PROF_START(t_discover);
cbm_discover_opts_t opts = {
.mode = p->mode,
.mode = p->requested_mode,
.ignore_file = NULL,
.max_file_size = 0,
};
@@ -2227,26 +2213,62 @@ static int cbm_pipeline_run_staged(cbm_pipeline_t *p, bool *was_incremental) {
/* Snapshot every semantic input once before routing/extraction. The same
* bytes drive exact no-op comparison and are checked against a fresh
* rediscovery immediately before any replacement is published. */
rc = cbm_pipeline_build_semantic_manifest(p->project_name, p->repo_path, files, file_count,
p->excluded_dirs, p->excluded_count, &p->git_ctx,
p->userconfig, &baseline_manifest, &baseline_count);
rc = mode_promoted
? cbm_pipeline_build_fresh_semantic_manifest(p->project_name, p->repo_path, p->mode,
&baseline_manifest, &baseline_count)
: cbm_pipeline_build_semantic_manifest(p->project_name, p->repo_path, files,
file_count, p->excluded_dirs, p->excluded_count,
&p->git_ctx, p->userconfig, &baseline_manifest,
&baseline_count);
if (rc != 0) {
rc = CBM_PIPELINE_ABORT_PRESERVE_DB;
goto cleanup;
}
/* Check for existing DB → try incremental or delete for reindex */
rc = try_incremental_or_delete_db(p, files, file_count, baseline_manifest, baseline_count);
rc = try_incremental_or_delete_db(p, files, file_count, baseline_manifest, baseline_count,
mode_promoted);
if (rc == CBM_PIPELINE_ABORT_PRESERVE_DB || rc == CBM_PIPELINE_PERSIST_FAILED) {
goto cleanup;
}
if (rc >= 0) {
*was_incremental = true;
goto cleanup;
}
if (rc != CBM_PIPELINE_FORCE_FULL_REINDEX) {
goto cleanup;
}
/* A changed downgrade rebuilds the complete graph at the stored effective
* mode. Keep the requested discovery lists to report the caller's scope. */
if (mode_promoted) {
cbm_discover_free(files, file_count);
files = NULL;
file_count = 0;
requested_excluded_dirs = p->excluded_dirs;
requested_excluded_count = p->excluded_count;
requested_ignored_files = p->ignored_files;
requested_ignored_count = p->ignored_count;
requested_ignored_total = p->ignored_total;
restore_requested_discovery = true;
p->excluded_dirs = NULL;
p->excluded_count = 0;
p->ignored_files = NULL;
p->ignored_count = 0;
p->ignored_total = 0;
opts.mode = p->mode;
rc = cbm_discover_ex2(p->repo_path, &opts, &files, &file_count, &p->excluded_dirs,
&p->excluded_count, &p->ignored_files, &p->ignored_count,
&p->ignored_total);
cbm_log_info("pipeline.rediscover", "requested_mode", pipeline_mode_name(p->requested_mode),
"effective_mode", pipeline_mode_name(p->mode), "files", itoa_buf(file_count));
if (rc != 0 || check_cancel(p)) {
rc = CBM_NOT_FOUND;
goto cleanup;
}
}
cbm_log_info("pipeline.route", "path", "full");
/* Phase 2: Create graph buffer and registry */
@@ -2296,6 +2318,15 @@ cleanup:
cbm_registry_free(p->registry);
p->registry = NULL;
cbm_path_alias_collection_free(path_aliases);
if (restore_requested_discovery) {
cbm_discover_free_excluded(p->excluded_dirs, p->excluded_count);
cbm_discover_free_ignored(p->ignored_files, p->ignored_count);
p->excluded_dirs = requested_excluded_dirs;
p->excluded_count = requested_excluded_count;
p->ignored_files = requested_ignored_files;
p->ignored_count = requested_ignored_count;
p->ignored_total = requested_ignored_total;
}
/* Clear and free user extension config */
cbm_set_user_lang_config(NULL);
cbm_userconfig_free(p->userconfig);
@@ -2458,7 +2489,7 @@ static int seal_staging_db(const char *staging_path) {
return rc;
}
static int export_after_publish(cbm_pipeline_t *p, const char *final_path, bool was_incremental) {
static int export_after_publish(cbm_pipeline_t *p, const char *final_path) {
if (p->persistence) {
CBM_PROF_START(t_art);
int rc = cbm_artifact_export(final_path, p->repo_path, p->project_name, CBM_ARTIFACT_BEST);
@@ -2469,7 +2500,7 @@ static int export_after_publish(cbm_pipeline_t *p, const char *final_path, bool
}
return rc;
}
if (was_incremental && p->repo_path && cbm_artifact_exists(p->repo_path)) {
if (p->repo_path && cbm_artifact_exists(p->repo_path)) {
(void)cbm_artifact_export(final_path, p->repo_path, p->project_name, CBM_ARTIFACT_FAST);
}
return 0;
@@ -2511,8 +2542,7 @@ int cbm_pipeline_run(cbm_pipeline_t *p) {
free(final_path);
return CBM_NOT_FOUND;
}
bool was_incremental = false;
int rc = cbm_pipeline_run_staged(p, &was_incremental);
int rc = cbm_pipeline_run_staged(p);
free(p->db_path);
p->db_path = configured_db_path;
@@ -2582,7 +2612,7 @@ int cbm_pipeline_run(cbm_pipeline_t *p) {
return CBM_PIPELINE_PERSIST_FAILED;
}
rc = export_after_publish(p, final_path, was_incremental);
rc = export_after_publish(p, final_path);
free(staging_path);
free(final_path);
return rc;
src/pipeline/pipeline_incremental.c
+15 -12
View File
@@ -15,7 +15,6 @@ enum { INCR_RING_BUF = 4, INCR_RING_MASK = 3, INCR_TS_BUF = 24 };
#include "pipeline/pipeline.h"
#include <stdio.h>
#include <time.h>
#include "pipeline/artifact.h"
#include "pipeline/lsp_surface.h"
#include "pipeline/pass_lsp_cross.h"
#include "sqlite3.h"
@@ -1468,7 +1467,7 @@ static int run_postpasses(cbm_pipeline_ctx_t *ctx, cbm_file_info_t *changed_file
* generation boundary as full indexing. */
static int dump_and_persist(cbm_gbuf_t *gbuf, const char *db_path, const char *project,
atomic_int *cancelled, const cbm_file_hash_t *manifest,
int manifest_count, const char *adr_content, const char *repo_path,
int manifest_count, const char *adr_content,
const cbm_coverage_row_t *cov, int cov_count,
const cbm_coverage_meta_t *meta_template,
const cbm_lsp_surface_row_t *surface_rows, int surface_row_count) {
@@ -1494,11 +1493,6 @@ static int dump_and_persist(cbm_gbuf_t *gbuf, const char *db_path, const char *p
if (rc != 0) {
return rc;
}
/* Auto-update artifact if one already exists (persistence was enabled previously) */
if (repo_path && cbm_artifact_exists(repo_path)) {
cbm_artifact_export(db_path, repo_path, project, CBM_ARTIFACT_FAST);
}
return 0;
}
@@ -2364,7 +2358,7 @@ out:
int cbm_pipeline_run_incremental(cbm_pipeline_t *p, const char *db_path, cbm_file_info_t *files,
int file_count, const cbm_file_hash_t *baseline_manifest,
int baseline_count) {
int baseline_count, bool force_full_on_mismatch) {
struct timespec t0;
cbm_clock_gettime(CLOCK_MONOTONIC, &t0);
closure_plan_t closure_plan = {0};
@@ -2425,7 +2419,16 @@ int cbm_pipeline_run_incremental(cbm_pipeline_t *p, const char *db_path, cbm_fil
incr_test_set_last_route(CBM_INCREMENTAL_ROUTE_NOOP);
#endif
cbm_log_info("incremental.noop", "reason", "semantic_manifest_equal");
return cbm_pipeline_refresh_artifact(p, db_path);
return 0;
}
if (force_full_on_mismatch) {
cbm_store_free_file_hashes(stored, stored_count);
cbm_store_close(store);
#if defined(CBM_INCREMENTAL_TEST_API) && CBM_INCREMENTAL_TEST_API
incr_test_set_last_route(CBM_INCREMENTAL_ROUTE_FORCED_FULL);
#endif
cbm_log_info("incremental.force_full", "reason", "mode_downgrade_changed");
return CBM_PIPELINE_FORCE_FULL_REINDEX;
}
/* Manifest delta. Closure repair recomputes exactly the changed
* files plus the recorded consumers of any changed SURFACE; every
@@ -2850,9 +2853,9 @@ int cbm_pipeline_run_incremental(cbm_pipeline_t *p, const char *db_path, cbm_fil
* re-parsed files have no codec output, and publishing a stale row
* would satisfy a future closure plan with yesterday's surface; an
* empty table just routes the next incremental to a full rebuild. */
int persist_rc = dump_and_persist(
existing, db_path, project, cbm_pipeline_cancelled_ptr(p), manifest, manifest_count,
saved_adr, cbm_pipeline_repo_path(p), cov, cov_n, &coverage_meta, NULL, 0);
int persist_rc =
dump_and_persist(existing, db_path, project, cbm_pipeline_cancelled_ptr(p), manifest,
manifest_count, saved_adr, cov, cov_n, &coverage_meta, NULL, 0);
cbm_pipeline_free_semantic_manifest(manifest, manifest_count);
free(saved_adr);
free(cov);
src/pipeline/pipeline_internal.h
+1 -6
View File
@@ -657,7 +657,7 @@ int cbm_scan_project_env_urls_excluded(const char *root_path, cbm_env_binding_t
* files, merges into disk DB. Returns 0 on success. */
int cbm_pipeline_run_incremental(cbm_pipeline_t *p, const char *db_path, cbm_file_info_t *files,
int file_count, const cbm_file_hash_t *baseline_manifest,
int baseline_count);
int baseline_count, bool force_full_on_mismatch);
/* Exact semantic inputs for no-op/forced-full routing. The manifest contains
* every discovered source plus repository controls actually consumed by
@@ -752,11 +752,6 @@ int cbm_delta_patch(cbm_store_t *store, const char *project, cbm_gbuf_t *gbuf, i
const cbm_delta_saved_edge_t *snapshot, int snapshot_count);
/* discard helper shared with the delta executor (unlink stage + sidecars). */
void cbm_pipeline_discard_stage(const char *stage_path);
/* The SQLite generation is authoritative. An explicitly requested artifact is
* part of the caller-visible operation and its export error is returned;
* automatic refresh of an already-existing artifact remains best-effort. */
int cbm_pipeline_refresh_artifact(cbm_pipeline_t *p, const char *db_path);
/* Hand the pipeline the per-file LSP-surface rows serialized at the
* collect_all_defs seam (the only moment the result cache is alive).
* Takes ownership; dump_and_persist_hashes writes them into the staging
tests/test_pipeline.c
+271 -3
View File
@@ -17,6 +17,7 @@
#include "foundation/dump_verify.h"
#include "foundation/sha256.h"
#include "foundation/compat_fs.h"
#include "foundation/log.h"
#include "foundation/win_utf8.h" // cbm_utf8_to_wide (Windows pipeline_test_set_mtime); no-op elsewhere
#include "discover/userconfig.h"
@@ -3314,6 +3315,116 @@ TEST(pipeline_exact_inputs_migrate_coverage_metadata_and_index_mode) {
PASS();
}
typedef struct {
bool published;
int rename_calls;
int export_count;
int exports_before_publish;
} artifact_publish_observer_t;
static artifact_publish_observer_t *g_artifact_publish_observer;
static void observe_artifact_publish_log(const char *line) {
if (g_artifact_publish_observer && line && strstr(line, "msg=artifact.export")) {
g_artifact_publish_observer->export_count++;
if (!g_artifact_publish_observer->published) {
g_artifact_publish_observer->exports_before_publish++;
}
}
}
static int observe_successful_publish_rename(const char *staging_path, const char *final_path,
void *arg) {
artifact_publish_observer_t *observer = (artifact_publish_observer_t *)arg;
observer->rename_calls++;
int rc = cbm_rename_replace(staging_path, final_path);
observer->published = rc == 0;
return rc;
}
static int run_observing_artifact_publish(cbm_pipeline_t *pipeline,
artifact_publish_observer_t *observer) {
cbm_pipeline_set_rename_hook_for_tests(pipeline, observe_successful_publish_rename, observer);
CBMLogLevel previous_level = cbm_log_get_level();
CBMLogFormat previous_format = cbm_log_get_format();
g_artifact_publish_observer = observer;
cbm_log_set_level(CBM_LOG_INFO);
cbm_log_set_format(CBM_LOG_FORMAT_TEXT);
cbm_log_set_sink_ex(observe_artifact_publish_log, CBM_LOG_SINK_REPLACE);
int rc = cbm_pipeline_run(pipeline);
cbm_log_set_sink(NULL);
cbm_log_set_format(previous_format);
cbm_log_set_level(previous_level);
g_artifact_publish_observer = NULL;
return rc;
}
static bool pipeline_reports_excluded_dir(cbm_pipeline_t *pipeline, const char *rel_path) {
char **excluded = NULL;
int excluded_count = 0;
cbm_pipeline_get_excluded(pipeline, &excluded, &excluded_count);
for (int i = 0; i < excluded_count; i++) {
if (excluded[i] && strcmp(excluded[i], rel_path) == 0) {
return true;
}
}
return false;
}
typedef struct {
int rc;
cbm_incremental_route_t route;
bool tools_excluded;
artifact_publish_observer_t publish;
} observed_fast_run_t;
static observed_fast_run_t run_observed_fast_pipeline(const char *repo_path, const char *db_path) {
observed_fast_run_t result = {.rc = CBM_NOT_FOUND};
cbm_pipeline_t *pipeline = cbm_pipeline_new(repo_path, db_path, CBM_MODE_FAST);
if (!pipeline) {
return result;
}
result.rc = run_observing_artifact_publish(pipeline, &result.publish);
result.route = cbm_pipeline_incremental_test_last_route();
result.tools_excluded = pipeline_reports_excluded_dir(pipeline, "tools");
cbm_pipeline_free(pipeline);
return result;
}
typedef struct {
int rc;
int before_nodes;
int after_nodes;
} imported_generation_t;
static imported_generation_t import_artifact_generation(const char *repo_path,
const char *import_path,
const char *project) {
imported_generation_t result = {
.rc = cbm_artifact_import(repo_path, import_path),
.before_nodes = -1,
.after_nodes = -1,
};
if (result.rc == 0) {
observe_named_generation(import_path, project, "StoredBefore", "StoredAfter",
&result.before_nodes, &result.after_nodes);
}
return result;
}
static bool stored_mode_is_full(const char *db_path, const char *project) {
cbm_store_t *store = cbm_store_open_path(db_path);
if (!store) {
return false;
}
cbm_coverage_meta_t meta = {0};
bool full = cbm_store_coverage_meta_get(store, project, &meta) == CBM_STORE_OK &&
meta.index_mode && strcmp(meta.index_mode, "full") == 0;
cbm_store_coverage_meta_clear(&meta);
cbm_store_close(store);
return full;
}
/* Once a repository contains a shared artifact, every subsequently published
* full generation must refresh it, even when persistence was not explicitly
* requested on that invocation. Otherwise the live DB advances while a clean
@@ -3331,10 +3442,11 @@ TEST(pipeline_existing_artifact_refreshes_after_default_forced_full_reindex) {
ASSERT_EQ(th_write_file(source_path, "def ArtifactGenerationBefore():\n return 1\n"), 0);
cbm_pipeline_incremental_test_reset_faults();
artifact_publish_observer_t baseline_observer = {0};
cbm_pipeline_t *baseline = cbm_pipeline_new(tmp, db_path, CBM_MODE_FULL);
ASSERT_NOT_NULL(baseline);
cbm_pipeline_set_persistence(baseline, true);
ASSERT_EQ(cbm_pipeline_run(baseline), 0);
int baseline_rc = run_observing_artifact_publish(baseline, &baseline_observer);
char project[256];
snprintf(project, sizeof(project), "%s", cbm_pipeline_project_name(baseline));
cbm_pipeline_free(baseline);
@@ -3342,18 +3454,29 @@ TEST(pipeline_existing_artifact_refreshes_after_default_forced_full_reindex) {
ASSERT_EQ(th_write_file(source_path, "def ArtifactGenerationAfter():\n return 2\n"), 0);
cbm_pipeline_incremental_test_reset_faults();
artifact_publish_observer_t reindex_observer = {0};
cbm_pipeline_t *default_reindex = cbm_pipeline_new(tmp, db_path, CBM_MODE_FULL);
ASSERT_NOT_NULL(default_reindex);
int reindex_rc = cbm_pipeline_run(default_reindex);
int reindex_rc = run_observing_artifact_publish(default_reindex, &reindex_observer);
cbm_incremental_route_t reindex_route = cbm_pipeline_incremental_test_last_route();
cbm_pipeline_free(default_reindex);
/* A derived artifact must not become an input that forces another rebuild,
* and refreshing it must not switch the authoritative DB back to WAL. */
cbm_pipeline_incremental_test_reset_faults();
artifact_publish_observer_t explicit_observer = {0};
cbm_pipeline_t *explicit_noop = cbm_pipeline_new(tmp, db_path, CBM_MODE_FULL);
ASSERT_NOT_NULL(explicit_noop);
cbm_pipeline_set_persistence(explicit_noop, true);
int explicit_rc = run_observing_artifact_publish(explicit_noop, &explicit_observer);
cbm_incremental_route_t explicit_route = cbm_pipeline_incremental_test_last_route();
cbm_pipeline_free(explicit_noop);
cbm_pipeline_incremental_test_reset_faults();
artifact_publish_observer_t observer = {0};
cbm_pipeline_t *unchanged = cbm_pipeline_new(tmp, db_path, CBM_MODE_FULL);
ASSERT_NOT_NULL(unchanged);
int unchanged_rc = cbm_pipeline_run(unchanged);
int unchanged_rc = run_observing_artifact_publish(unchanged, &observer);
cbm_incremental_route_t unchanged_route = cbm_pipeline_incremental_test_last_route();
cbm_pipeline_free(unchanged);
@@ -3397,10 +3520,25 @@ TEST(pipeline_existing_artifact_refreshes_after_default_forced_full_reindex) {
cbm_pipeline_incremental_test_reset_faults();
th_rmtree(tmp);
ASSERT_EQ(baseline_rc, 0);
ASSERT_EQ(baseline_observer.rename_calls, 1);
ASSERT_EQ(baseline_observer.exports_before_publish, 0);
ASSERT_EQ(baseline_observer.export_count, 1);
ASSERT_EQ(reindex_rc, 0);
ASSERT_EQ(reindex_route, CBM_INCREMENTAL_ROUTE_FORCED_FULL);
ASSERT_EQ(reindex_observer.rename_calls, 1);
ASSERT_EQ(reindex_observer.exports_before_publish, 0);
ASSERT_EQ(reindex_observer.export_count, 1);
ASSERT_EQ(explicit_rc, 0);
ASSERT_EQ(explicit_route, CBM_INCREMENTAL_ROUTE_NOOP);
ASSERT_EQ(explicit_observer.rename_calls, 1);
ASSERT_EQ(explicit_observer.exports_before_publish, 0);
ASSERT_EQ(explicit_observer.export_count, 1);
ASSERT_EQ(unchanged_rc, 0);
ASSERT_EQ(unchanged_route, CBM_INCREMENTAL_ROUTE_NOOP);
ASSERT_EQ(observer.rename_calls, 1);
ASSERT_EQ(observer.exports_before_publish, 0);
ASSERT_EQ(observer.export_count, 1);
ASSERT_TRUE(journal_ok);
ASSERT_STR_EQ(journal_mode, "delete");
ASSERT_EQ(live_before, 0);
@@ -10411,6 +10549,135 @@ TEST(full_reindex_preserves_exact_long_db_path) {
}
#endif
TEST(incremental_downgrade_preserves_scope_and_artifact_across_change_noop_delete) {
char tmpdir[256];
char artifact_tmpdir[256];
snprintf(tmpdir, sizeof(tmpdir), "/tmp/cbm_mode_scope_XXXXXX");
snprintf(artifact_tmpdir, sizeof(artifact_tmpdir), "/tmp/cbm_mode_artifact_XXXXXX");
ASSERT_NOT_NULL(cbm_mkdtemp(tmpdir));
ASSERT_NOT_NULL(cbm_mkdtemp(artifact_tmpdir));
char dbpath[512];
char cancelled_import_path[512];
char retry_import_path[512];
char deleted_import_path[512];
snprintf(dbpath, sizeof(dbpath), "%s/test.db", tmpdir);
snprintf(cancelled_import_path, sizeof(cancelled_import_path), "%s/cancelled.db",
artifact_tmpdir);
snprintf(retry_import_path, sizeof(retry_import_path), "%s/retry.db", artifact_tmpdir);
snprintf(deleted_import_path, sizeof(deleted_import_path), "%s/deleted.db", artifact_tmpdir);
ASSERT_EQ(th_write_file(TH_PATH(tmpdir, "main.go"), "package main\n\nfunc main() {}\n"), 0);
ASSERT_TRUE(cbm_mkdir_p(TH_PATH(tmpdir, "tools"), 0755));
ASSERT_EQ(th_write_file(TH_PATH(tmpdir, "tools/util.go"),
"package tools\n\nfunc StoredBefore() string { return \"old\" }\n"),
0);
cbm_pipeline_t *pipeline = cbm_pipeline_new(tmpdir, dbpath, CBM_MODE_FULL);
ASSERT_NOT_NULL(pipeline);
cbm_pipeline_set_persistence(pipeline, true);
ASSERT_EQ(cbm_pipeline_run(pipeline), 0);
char *project = strdup(cbm_pipeline_project_name(pipeline));
ASSERT_NOT_NULL(project);
cbm_pipeline_free(pipeline);
ASSERT_TRUE(cbm_artifact_exists(tmpdir));
ASSERT_EQ(th_write_file(TH_PATH(tmpdir, "tools/util.go"),
"package tools\n\nfunc StoredAfter() string { return \"new\" }\n"),
0);
cbm_pipeline_incremental_test_reset_faults();
cbm_pipeline_incremental_test_cancel_after_predump_once();
observed_fast_run_t cancelled = run_observed_fast_pipeline(tmpdir, dbpath);
int cancelled_live_before;
int cancelled_live_after;
observe_named_generation(dbpath, project, "StoredBefore", "StoredAfter", &cancelled_live_before,
&cancelled_live_after);
imported_generation_t cancelled_artifact =
import_artifact_generation(tmpdir, cancelled_import_path, project);
cbm_pipeline_incremental_test_reset_faults();
observed_fast_run_t retry = run_observed_fast_pipeline(tmpdir, dbpath);
int retry_live_before;
int retry_live_after;
observe_named_generation(dbpath, project, "StoredBefore", "StoredAfter", &retry_live_before,
&retry_live_after);
bool retry_full_mode = stored_mode_is_full(dbpath, project);
imported_generation_t retry_artifact =
import_artifact_generation(tmpdir, retry_import_path, project);
cbm_pipeline_incremental_test_reset_faults();
observed_fast_run_t noop = run_observed_fast_pipeline(tmpdir, dbpath);
ASSERT_EQ(cbm_unlink(TH_PATH(tmpdir, "tools/util.go")), 0);
cbm_pipeline_incremental_test_reset_faults();
observed_fast_run_t deleted = run_observed_fast_pipeline(tmpdir, dbpath);
int deleted_live_before;
int deleted_live_after;
observe_named_generation(dbpath, project, "StoredBefore", "StoredAfter", &deleted_live_before,
&deleted_live_after);
bool delete_full_mode = stored_mode_is_full(dbpath, project);
cbm_store_t *store = cbm_store_open_path(dbpath);
ASSERT_NOT_NULL(store);
cbm_file_hash_t deleted_hash = {0};
int deleted_hash_rc = cbm_store_get_file_hash(store, project, "tools/util.go", &deleted_hash);
cbm_store_clear_file_hash(&deleted_hash);
cbm_store_close(store);
imported_generation_t deleted_artifact =
import_artifact_generation(tmpdir, deleted_import_path, project);
cbm_pipeline_incremental_test_reset_faults();
free(project);
th_rmtree(artifact_tmpdir);
th_rmtree(tmpdir);
ASSERT_EQ(cancelled.rc, CBM_PIPELINE_ABORT_PRESERVE_DB);
ASSERT_EQ(cancelled.route, CBM_INCREMENTAL_ROUTE_FORCED_FULL);
ASSERT_TRUE(cancelled.tools_excluded);
ASSERT_EQ(cancelled_live_before, 1);
ASSERT_EQ(cancelled_live_after, 0);
ASSERT_EQ(cancelled.publish.rename_calls, 0);
ASSERT_EQ(cancelled.publish.export_count, 0);
ASSERT_EQ(cancelled_artifact.rc, 0);
ASSERT_EQ(cancelled_artifact.before_nodes, 1);
ASSERT_EQ(cancelled_artifact.after_nodes, 0);
ASSERT_EQ(retry.rc, 0);
ASSERT_EQ(retry.route, CBM_INCREMENTAL_ROUTE_FORCED_FULL);
ASSERT_TRUE(retry.tools_excluded);
ASSERT_TRUE(retry_full_mode);
ASSERT_EQ(retry_live_before, 0);
ASSERT_EQ(retry_live_after, 1);
ASSERT_EQ(retry.publish.rename_calls, 1);
ASSERT_EQ(retry.publish.exports_before_publish, 0);
ASSERT_EQ(retry.publish.export_count, 1);
ASSERT_EQ(retry_artifact.rc, 0);
ASSERT_EQ(retry_artifact.before_nodes, 0);
ASSERT_EQ(retry_artifact.after_nodes, 1);
ASSERT_EQ(noop.rc, 0);
ASSERT_EQ(noop.route, CBM_INCREMENTAL_ROUTE_NOOP);
ASSERT_TRUE(noop.tools_excluded);
ASSERT_EQ(noop.publish.rename_calls, 1);
ASSERT_EQ(noop.publish.exports_before_publish, 0);
ASSERT_EQ(noop.publish.export_count, 1);
ASSERT_EQ(deleted.rc, 0);
ASSERT_EQ(deleted.route, CBM_INCREMENTAL_ROUTE_FORCED_FULL);
ASSERT_TRUE(deleted.tools_excluded);
ASSERT_EQ(deleted.publish.rename_calls, 1);
ASSERT_EQ(deleted.publish.exports_before_publish, 0);
ASSERT_EQ(deleted.publish.export_count, 1);
ASSERT_EQ(deleted_hash_rc, CBM_STORE_NOT_FOUND);
ASSERT_TRUE(delete_full_mode);
ASSERT_EQ(deleted_live_before, 0);
ASSERT_EQ(deleted_live_after, 0);
ASSERT_EQ(deleted_artifact.rc, 0);
ASSERT_EQ(deleted_artifact.before_nodes, 0);
ASSERT_EQ(deleted_artifact.after_nodes, 0);
PASS();
}
TEST(incremental_fast_preserves_mode_skipped_tools_dir) {
/* Regression: 2026-04-13. A fast-mode reindex after a full-mode index
* was silently destroying every file under FAST_SKIP_DIRS directories
@@ -11901,6 +12168,7 @@ SUITE(pipeline) {
* broad pipeline suite so RED/GREEN iterations exercise only this boundary;
* the default all-suite run still executes it. */
SUITE(pipeline_semantic_manifest_repro) {
RUN_TEST(incremental_downgrade_preserves_scope_and_artifact_across_change_noop_delete);
RUN_TEST(pipeline_incremental_repoints_call_reference_without_stale_edge);
RUN_TEST(pipeline_parallel_manifest_is_byte_stable_above_threshold);
RUN_TEST(pipeline_closure_repair_body_edit_converges_with_fresh_full);