diff --git a/src/cli/cli.c b/src/cli/cli.c index ffca9d2c6..f05d9b531 100644 --- a/src/cli/cli.c +++ b/src/cli/cli.c @@ -2927,12 +2927,14 @@ const cbm_config_entry_t CBM_CONFIG_REGISTRY[] = { CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_EAGER, NULL, "Indexing", - "When exact incremental publishes may defer global semantic/similarity edge refresh", + "When incremental publishes may defer global semantic/similarity edge refresh", CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_EAGER "|" - CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_STALE_ON_EXACT, + CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_STALE_ON_EXACT "|" + CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_STALE_ON_INCREMENTAL, "'eager' preserves full/moderate publish freshness. 'stale_on_exact' lets small exact " "incremental graph deltas publish after marking semantic_edges stale; semantic/similarity " - "queries warn until an eager index run or full reindex rebuilds those global edges."}, + "queries warn until an eager index run or full reindex rebuilds those global edges. " + "'stale_on_incremental' also allows containment incremental publishes to defer."}, /* ── Search ── */ {"search_limit", "50", NULL, "Search", "Default max results for search_graph/search_code", diff --git a/src/pipeline/pipeline.c b/src/pipeline/pipeline.c index fc18fc84c..5c5fb704a 100644 --- a/src/pipeline/pipeline.c +++ b/src/pipeline/pipeline.c @@ -83,6 +83,7 @@ typedef enum { typedef enum { CBM_INCREMENTAL_DERIVED_REFRESH_EAGER = 0, CBM_INCREMENTAL_DERIVED_REFRESH_STALE_ON_EXACT, + CBM_INCREMENTAL_DERIVED_REFRESH_STALE_ON_INCREMENTAL, } cbm_incremental_derived_refresh_policy_t; void cbm_pipeline_lock(void) { @@ -363,12 +364,17 @@ void cbm_pipeline_apply_config(cbm_pipeline_t *p, cbm_config_t *cfg) { const char *derived_refresh = cbm_config_get(cfg, CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH, CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_EAGER); - p->incremental_derived_refresh = - derived_refresh && - strcmp(derived_refresh, - CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_STALE_ON_EXACT) == 0 - ? CBM_INCREMENTAL_DERIVED_REFRESH_STALE_ON_EXACT - : CBM_INCREMENTAL_DERIVED_REFRESH_EAGER; + if (derived_refresh && + strcmp(derived_refresh, CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_STALE_ON_INCREMENTAL) == + 0) { + p->incremental_derived_refresh = CBM_INCREMENTAL_DERIVED_REFRESH_STALE_ON_INCREMENTAL; + } else if (derived_refresh && + strcmp(derived_refresh, CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_STALE_ON_EXACT) == + 0) { + p->incremental_derived_refresh = CBM_INCREMENTAL_DERIVED_REFRESH_STALE_ON_EXACT; + } else { + p->incremental_derived_refresh = CBM_INCREMENTAL_DERIVED_REFRESH_EAGER; + } int max_changed = cbm_config_get_int(cfg, CBM_CONFIG_INCREMENTAL_EXACT_MAX_CHANGED_PATHS, p->exact_delta_max_changed_paths); @@ -637,7 +643,16 @@ bool cbm_pipeline_overlay_publish_small_deltas(const cbm_pipeline_t *p) { } bool cbm_pipeline_incremental_derived_refresh_stale_on_exact(const cbm_pipeline_t *p) { - return p && p->incremental_derived_refresh == CBM_INCREMENTAL_DERIVED_REFRESH_STALE_ON_EXACT; + return p && + (p->incremental_derived_refresh == CBM_INCREMENTAL_DERIVED_REFRESH_STALE_ON_EXACT || + p->incremental_derived_refresh == + CBM_INCREMENTAL_DERIVED_REFRESH_STALE_ON_INCREMENTAL); +} + +bool cbm_pipeline_incremental_derived_refresh_stale_on_incremental(const cbm_pipeline_t *p) { + return p && + p->incremental_derived_refresh == + CBM_INCREMENTAL_DERIVED_REFRESH_STALE_ON_INCREMENTAL; } cbm_pipeline_exact_delta_stats_t cbm_pipeline_exact_delta_stats(const cbm_pipeline_t *p) { @@ -1532,7 +1547,8 @@ static int pipeline_persist_replacement_metadata(cbm_pipeline_t *p, cbm_store_t return fts_rc; } } - int derived_rc = cbm_pipeline_mark_replacement_derived_views(store, p->project_name, p->mode); + int derived_rc = + cbm_pipeline_mark_replacement_derived_views(store, p->project_name, p->mode, true); if (derived_rc != CBM_STORE_OK) { cbm_log_error("pipeline.err", "phase", "mark_derived_views", "rc", itoa_buf(derived_rc)); return derived_rc; diff --git a/src/pipeline/pipeline.h b/src/pipeline/pipeline.h index 35bd1b5a7..8732939bf 100644 --- a/src/pipeline/pipeline.h +++ b/src/pipeline/pipeline.h @@ -125,6 +125,7 @@ void cbm_pipeline_set_flush_store(cbm_pipeline_t *p, cbm_store_t *store); #define CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH "incremental_derived_refresh" #define CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_EAGER "eager" #define CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_STALE_ON_EXACT "stale_on_exact" +#define CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_STALE_ON_INCREMENTAL "stale_on_incremental" /* Set the Jaccard similarity threshold for SIMILAR-edge creation (pass_similarity). * <=0 (or unset) uses the CBM_MINHASH_JACCARD_THRESHOLD default. Before run(). */ diff --git a/src/pipeline/pipeline_incremental.c b/src/pipeline/pipeline_incremental.c index 11130e073..04530e55c 100644 --- a/src/pipeline/pipeline_incremental.c +++ b/src/pipeline/pipeline_incremental.c @@ -2371,7 +2371,8 @@ cleanup: static int publish_and_persist(cbm_gbuf_t *gbuf, const char *db_path, const char *project, cbm_file_info_t *files, int file_count, const cbm_file_hash_t *mode_skipped, int mode_skipped_count, - const char *repo_path, const char *pass_fingerprint, int mode) { + const char *repo_path, const char *pass_fingerprint, int mode, + bool semantic_edges_refreshed) { struct timespec t; cbm_clock_gettime(CLOCK_MONOTONIC, &t); @@ -2423,7 +2424,9 @@ static int publish_and_persist(cbm_gbuf_t *gbuf, const char *db_path, const char cbm_store_close(hash_store); return fts_rc; } - int derived_rc = cbm_pipeline_mark_replacement_derived_views(hash_store, project, mode); + int derived_rc = + cbm_pipeline_mark_replacement_derived_views(hash_store, project, mode, + semantic_edges_refreshed); if (derived_rc != CBM_STORE_OK) { cbm_log_error("incremental.err", "phase", "mark_derived_views", "rc", itoa_buf_incr(derived_rc)); @@ -2752,7 +2755,10 @@ int cbm_pipeline_run_incremental(cbm_pipeline_t *p, const char *db_path, cbm_fil itoa_buf_incr(CBM_NOT_FOUND)); pipeline_rc = CBM_NOT_FOUND; } else { - pipeline_rc = run_postpasses(&ctx, changed_files, ci, project, true); + bool refresh_global_semantic_edges = + !cbm_pipeline_incremental_derived_refresh_stale_on_incremental(p); + pipeline_rc = run_postpasses(&ctx, changed_files, ci, project, + refresh_global_semantic_edges); } } @@ -2807,10 +2813,12 @@ int cbm_pipeline_run_incremental(cbm_pipeline_t *p, const char *db_path, cbm_fil * covers incremental reindexes, not just full publishes. */ cbm_pipeline_set_committed_counts(p, cbm_gbuf_node_count(existing), cbm_gbuf_edge_count(existing)); + bool semantic_edges_refreshed = + !cbm_pipeline_incremental_derived_refresh_stale_on_incremental(p); int persist_rc = publish_and_persist(existing, db_path, project, files, file_count, cls.mode_skipped, cls.mode_skipped_count, cbm_pipeline_repo_path(p), pass_fingerprint, - cbm_pipeline_get_mode(p)); + cbm_pipeline_get_mode(p), semantic_edges_refreshed); if (persist_rc == 0) { cbm_pipeline_set_graph_changed(p, true); cbm_pipeline_set_publish_kind(p, CBM_PIPELINE_PUBLISH_INCREMENTAL_CONTAINMENT); diff --git a/src/pipeline/pipeline_internal.h b/src/pipeline/pipeline_internal.h index 99e6d4a2e..7cbc9625f 100644 --- a/src/pipeline/pipeline_internal.h +++ b/src/pipeline/pipeline_internal.h @@ -438,7 +438,8 @@ static inline bool cbm_pipeline_mode_extracts_macro_nodes(int mode) { } static inline int cbm_pipeline_mark_replacement_derived_views(cbm_store_t *store, - const char *project, int mode) { + const char *project, int mode, + bool semantic_edges_refreshed) { static const char *const complete_graph_views[] = { CBM_STORE_DERIVED_VIEW_ROUTES, CBM_STORE_DERIVED_VIEW_ARCHITECTURE, @@ -459,7 +460,7 @@ static inline int cbm_pipeline_mark_replacement_derived_views(cbm_store_t *store if (rc != CBM_STORE_OK) { return rc; } - if (cbm_pipeline_mode_builds_global_semantic_edges(mode)) { + if (cbm_pipeline_mode_builds_global_semantic_edges(mode) && semantic_edges_refreshed) { return cbm_store_mark_derived_views_complete( store, project, CBM_PIPELINE_COMPAT_GENERATION, semantic_views, (int)(sizeof(semantic_views) / sizeof(semantic_views[0]))); @@ -1151,6 +1152,7 @@ void cbm_pipeline_set_publish_kind(cbm_pipeline_t *p, cbm_pipeline_publish_kind_ void cbm_pipeline_set_publish_reason(cbm_pipeline_t *p, const char *reason); bool cbm_pipeline_overlay_publish_small_deltas(const cbm_pipeline_t *p); bool cbm_pipeline_incremental_derived_refresh_stale_on_exact(const cbm_pipeline_t *p); +bool cbm_pipeline_incremental_derived_refresh_stale_on_incremental(const cbm_pipeline_t *p); void cbm_pipeline_set_exact_delta_stats(cbm_pipeline_t *p, int changed_paths, int affected_paths, int published_paths); void cbm_pipeline_set_exact_delta_stats_with_limit(cbm_pipeline_t *p, int changed_paths, diff --git a/tests/test_pipeline.c b/tests/test_pipeline.c index fd0284412..144200b94 100644 --- a/tests/test_pipeline.c +++ b/tests/test_pipeline.c @@ -11343,6 +11343,81 @@ TEST(incremental_full_stale_on_exact_mixed_delete_upsert_marks_semantic_stale) { PASS(); } +TEST(incremental_full_stale_on_incremental_defers_containment_semantic_refresh) { + if (setup_incremental_repo() != 0) { + FAIL("setup failed"); + } + + char path[CBM_PATH_MAX]; + int n = snprintf(path, sizeof(path), "%s/leaf.go", g_incr_tmpdir); + ASSERT(n >= 0 && (size_t)n < sizeof(path)); + ASSERT_EQ(th_write_file(path, + "package main\n\n" + "func Leaf() int {\n\treturn 1\n}\n"), + 0); + + cbm_config_t *cfg = incremental_test_config(g_incr_tmpdir); + ASSERT_NOT_NULL(cfg); + ASSERT_EQ(cbm_config_set(cfg, CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH, + CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_STALE_ON_INCREMENTAL), + 0); + + cbm_pipeline_t *p = cbm_pipeline_new(g_incr_tmpdir, g_incr_dbpath, CBM_MODE_FULL); + ASSERT_NOT_NULL(p); + cbm_pipeline_apply_config(p, cfg); + ASSERT_EQ(cbm_pipeline_run(p), 0); + char *project = cbm_strdup(cbm_pipeline_project_name(p)); + cbm_pipeline_free(p); + ASSERT_NOT_NULL(project); + + n = snprintf(path, sizeof(path), "%s/main.go", g_incr_tmpdir); + ASSERT(n >= 0 && (size_t)n < sizeof(path)); + ASSERT_EQ(th_write_file(path, + "package main\n\n" + "func main() {\n\tHelper()\n\tNewHelper()\n\tLeaf()\n}\n\n" + "func NewMain() int {\n\treturn 11\n}\n"), + 0); + n = snprintf(path, sizeof(path), "%s/helper.go", g_incr_tmpdir); + ASSERT(n >= 0 && (size_t)n < sizeof(path)); + ASSERT_EQ(th_write_file(path, + "package main\n\n" + "func Helper() string {\n\treturn \"updated\"\n}\n\n" + "func NewHelper() int {\n\treturn 13\n}\n"), + 0); + n = snprintf(path, sizeof(path), "%s/leaf.go", g_incr_tmpdir); + ASSERT(n >= 0 && (size_t)n < sizeof(path)); + ASSERT_EQ(th_write_file(path, + "package main\n\n" + "func Leaf() int {\n\treturn 2\n}\n\n" + "func NewLeaf() int {\n\treturn 17\n}\n"), + 0); + + p = cbm_pipeline_new(g_incr_tmpdir, g_incr_dbpath, CBM_MODE_FULL); + ASSERT_NOT_NULL(p); + cbm_pipeline_apply_config(p, cfg); + pipeline_capture_logs_start(); + int run_rc = cbm_pipeline_run(p); + const char *logs = pipeline_capture_logs_end(); + ASSERT_EQ(run_rc, 0); + ASSERT(strstr(logs, "msg=incremental.exact.skip reason=changed_batch_too_large") != NULL); + ASSERT(strstr(logs, "pass=incr_semantic_edges") == NULL); + ASSERT_EQ(cbm_pipeline_publish_kind(p), CBM_PIPELINE_PUBLISH_INCREMENTAL_CONTAINMENT); + ASSERT_STR_EQ(cbm_pipeline_publish_reason(p), "changed_batch_too_large"); + cbm_pipeline_free(p); + + cbm_store_t *s = cbm_store_open_path(g_incr_dbpath); + ASSERT_NOT_NULL(s); + ASSERT_TRUE(pipeline_test_derived_status_is(s, project, + CBM_STORE_DERIVED_VIEW_SEMANTIC_EDGES, + CBM_STORE_DERIVED_STATUS_STALE)); + cbm_store_close(s); + + free(project); + cbm_config_close(cfg); + cleanup_incremental_repo(); + PASS(); +} + TEST(incremental_fast_mixed_unowned_edge_frontier_falls_back_to_full_rebuild) { enum { PIPELINE_CONFIGURED_AFFECTED_CAP = CBM_SZ_8 }; if (setup_incremental_repo() != 0) { @@ -14637,9 +14712,11 @@ TEST(config_registry_includes_incremental_derived_refresh_policy) { ASSERT_NOT_NULL(entry); ASSERT_STR_EQ(entry->default_val, CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_EAGER); ASSERT_STR_EQ(entry->category, "Indexing"); - ASSERT_STR_EQ(entry->range, "eager|stale_on_exact"); + ASSERT_STR_EQ(entry->range, "eager|stale_on_exact|stale_on_incremental"); ASSERT_NOT_NULL(strstr(entry->guidance, CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_STALE_ON_EXACT)); + ASSERT_NOT_NULL(strstr(entry->guidance, + CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_STALE_ON_INCREMENTAL)); PASS(); } @@ -15631,6 +15708,7 @@ SUITE(pipeline) { RUN_TEST(incremental_fast_configured_frontier_cap_allows_bounded_exact); RUN_TEST(incremental_full_stale_on_exact_defers_global_derived_refresh); RUN_TEST(incremental_full_stale_on_exact_mixed_delete_upsert_marks_semantic_stale); + RUN_TEST(incremental_full_stale_on_incremental_defers_containment_semantic_refresh); RUN_TEST(incremental_fast_mixed_unowned_edge_frontier_falls_back_to_full_rebuild); RUN_TEST(incremental_fast_expands_small_inbound_frontier_and_matches_full); RUN_TEST(incremental_fast_three_file_batch_falls_back_to_full_rebuild_parity);