diff --git a/src/graph_buffer/graph_buffer.c b/src/graph_buffer/graph_buffer.c index 7f44a2d19..f8272e63a 100644 --- a/src/graph_buffer/graph_buffer.c +++ b/src/graph_buffer/graph_buffer.c @@ -835,6 +835,39 @@ void cbm_gbuf_set_next_id(cbm_gbuf_t *gb, int64_t next_id) { gb->next_id = next_id; } +void cbm_gbuf_raise_next_id(cbm_gbuf_t *gb, int64_t next_id) { + if (gb && next_id > gb->next_id) { + gb->next_id = next_id; + } +} + +static int gb_id_cmp(const void *a, const void *b) { + int64_t x = *(const int64_t *)a; + int64_t y = *(const int64_t *)b; + return (x > y) - (x < y); +} + +int cbm_gbuf_duplicate_edge_ids(const cbm_gbuf_t *gb) { + if (!gb || gb->edges.count < 2) { + return 0; + } + int n = gb->edges.count; + int64_t *ids = cbm_calloc(CBM_MEM_CLASS_GBUF_INDEX, (size_t)n * sizeof(*ids)); + if (!ids) { + return CBM_NOT_FOUND; + } + for (int i = 0; i < n; i++) { + ids[i] = gb->edges.items[i]->id; + } + qsort(ids, (size_t)n, sizeof(*ids), gb_id_cmp); + int dups = 0; + for (int i = 1; i < n; i++) { + dups += ids[i] == ids[i - SKIP_ONE]; + } + cbm_free(CBM_MEM_CLASS_GBUF_INDEX, ids); + return dups; +} + /* ── Definition variants ───────────────────────────────────────────── * One name defined more than once under mutually exclusive conditions — * C-family #if/#else, Rust #[cfg], build-tagged files, a definition under an diff --git a/src/graph_buffer/graph_buffer.h b/src/graph_buffer/graph_buffer.h index ffbff0469..0553f47f0 100644 --- a/src/graph_buffer/graph_buffer.h +++ b/src/graph_buffer/graph_buffer.h @@ -100,9 +100,16 @@ int cbm_gbuf_node_count(const cbm_gbuf_t *gb); /* Get the next ID that would be assigned. Used to initialize shared atomic counters. */ int64_t cbm_gbuf_next_id(const cbm_gbuf_t *gb); -/* Set the next ID counter. Used after merging worker gbufs to sync the main counter. */ +/* Set the next ID counter. Pins it: the delta preseed sets it to the DB's next id. */ void cbm_gbuf_set_next_id(cbm_gbuf_t *gb, int64_t next_id); +/* Raise the next ID counter to at least next_id; never lowers it, so no id + * already handed out comes back. Used to sync after merging worker gbufs. */ +void cbm_gbuf_raise_next_id(cbm_gbuf_t *gb, int64_t next_id); + +/* Number of edges whose id an earlier edge already carries (0 = ids unique). */ +int cbm_gbuf_duplicate_edge_ids(const cbm_gbuf_t *gb); + /* Delete all nodes with a label. Cascade-deletes referencing edges. */ int cbm_gbuf_delete_by_label(cbm_gbuf_t *gb, const char *label); diff --git a/src/pipeline/pipeline.c b/src/pipeline/pipeline.c index dd93d63de..7fe2f419f 100644 --- a/src/pipeline/pipeline.c +++ b/src/pipeline/pipeline.c @@ -283,6 +283,9 @@ struct cbm_pipeline { * per pipeline so concurrent test/process activity cannot cross-trigger. */ void (*before_publish_hook)(cbm_pipeline_t *, const char *, void *); void *before_publish_hook_ctx; + /* Test-only: sees the complete graph buffer before the dump releases its indexes. */ + void (*before_dump_hook)(const cbm_gbuf_t *, void *); + void *before_dump_hook_ctx; int (*rename_hook)(const char *, const char *, void *); void *rename_hook_ctx; }; @@ -1047,6 +1050,32 @@ void cbm_pipeline_set_before_publish_hook_for_tests( } } +void cbm_pipeline_set_before_dump_hook_for_tests(cbm_pipeline_t *p, + void (*hook)(const cbm_gbuf_t *, void *), + void *ctx) { + if (p) { + p->before_dump_hook = hook; + p->before_dump_hook_ctx = ctx; + } +} + +static CBM_TLS void (*g_id_sync_hook)(cbm_gbuf_t *, void *); +static CBM_TLS void *g_id_sync_hook_ctx; + +void cbm_pipeline_set_id_sync_hook_for_tests(void (*hook)(cbm_gbuf_t *, void *), void *ctx) { + g_id_sync_hook = hook; + g_id_sync_hook_ctx = ctx; +} + +void cbm_pipeline_sync_next_id(cbm_gbuf_t *gb, _Atomic int64_t *shared_ids) { + if (g_id_sync_hook) { + g_id_sync_hook(gb, g_id_sync_hook_ctx); + } + /* Raise only: lowering it to the shared value handed the merged edges' + * ids out a second time. */ + cbm_gbuf_raise_next_id(gb, atomic_load(shared_ids)); +} + void cbm_pipeline_set_rename_hook_for_tests(cbm_pipeline_t *p, int (*hook)(const char *, const char *, void *), void *ctx) { @@ -2291,7 +2320,7 @@ static int run_parallel_pipeline(cbm_pipeline_t *p, cbm_pipeline_ctx_t *ctx, cbm_pipeline_spill_close(ctx); return rc != 0 ? rc : CBM_NOT_FOUND; } - cbm_gbuf_set_next_id(p->gbuf, atomic_load(&shared_ids)); + cbm_pipeline_sync_next_id(p->gbuf, &shared_ids); /* extract -> registry handoff: return the extract phase's freed-but-retained * allocator pages to the OS before registry_build allocates. On a 2x Linux * index the extract peak holds ~13 GB of reclaimable pages (peak_mb 20.7 vs @@ -2465,7 +2494,7 @@ static int run_parallel_pipeline(cbm_pipeline_t *p, cbm_pipeline_ctx_t *ctx, } free(def_modules); } - cbm_gbuf_set_next_id(p->gbuf, atomic_load(&shared_ids)); + cbm_pipeline_sync_next_id(p->gbuf, &shared_ids); cbm_pipeline_extract_infra_routes(ctx, p->gbuf, files, cache, file_count); cbm_pipeline_process_infra_bindings(ctx, p->gbuf, files, cache, file_count); for (int i = 0; i < file_count; i++) { @@ -3415,6 +3444,9 @@ static int dump_and_persist_hashes(cbm_pipeline_t *p, const cbm_file_hash_t *bas * release_gbuf_indexes(), which frees node_by_qn (graph_buffer.c), after * which cbm_gbuf_node_count() returns 0. Reading these post-dump left * committed_nodes at 0, so the #334 plausibility gate never fired. */ + if (p->before_dump_hook) { + p->before_dump_hook(p->gbuf, p->before_dump_hook_ctx); + } p->committed_nodes = cbm_gbuf_node_count(p->gbuf); p->committed_edges = cbm_gbuf_edge_count(p->gbuf); int rc = publish_generation_impl(&generation, p->frozen ? p : NULL); diff --git a/src/pipeline/pipeline_incremental.c b/src/pipeline/pipeline_incremental.c index 330d36d89..9f5bbf546 100644 --- a/src/pipeline/pipeline_incremental.c +++ b/src/pipeline/pipeline_incremental.c @@ -1742,7 +1742,7 @@ static int run_extract_resolve(cbm_pipeline_ctx_t *ctx, cbm_file_info_t *changed free_incremental_result_cache(cache, ci); return rc; } - cbm_gbuf_set_next_id(ctx->gbuf, atomic_load(&shared_ids)); + cbm_pipeline_sync_next_id(ctx->gbuf, &shared_ids); cbm_clock_gettime(CLOCK_MONOTONIC, &t); rc = cbm_build_registry_from_cache(ctx, changed_files, ci, cache); @@ -1902,7 +1902,7 @@ static int run_extract_resolve(cbm_pipeline_ctx_t *ctx, cbm_file_info_t *changed } cbm_log_info("pass.timing", "pass", "incr_resolve", "elapsed_ms", itoa_buf((int)elapsed_ms(t))); - cbm_gbuf_set_next_id(ctx->gbuf, atomic_load(&shared_ids)); + cbm_pipeline_sync_next_id(ctx->gbuf, &shared_ids); free_incremental_result_cache(cache, ci); if (legacy_cross_arena_live) { free(all_defs); /* array returned by the existing collect helper */ diff --git a/src/pipeline/pipeline_internal.h b/src/pipeline/pipeline_internal.h index f7f759be8..92e7e315d 100644 --- a/src/pipeline/pipeline_internal.h +++ b/src/pipeline/pipeline_internal.h @@ -1056,6 +1056,14 @@ int cbm_parallel_extract(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *files, CBMFileResult **result_cache, _Atomic int64_t *shared_ids, int worker_count); +/* Sync the main buffer's id counter after a parallel extract or resolve. + * Merged worker nodes keep their shared ids, and merged edges take fresh ids + * from the main counter after the highest of them, so the main counter can + * end above the shared one. */ +void cbm_pipeline_sync_next_id(cbm_gbuf_t *gb, _Atomic int64_t *shared_ids); +/* Test seam: runs on the syncing thread right before each such sync. */ +void cbm_pipeline_set_id_sync_hook_for_tests(void (*hook)(cbm_gbuf_t *, void *), void *ctx); + /* Phase 3B: Serial registry build from cached extraction results. * Creates DEFINES, DEFINES_METHOD, and IMPORTS edges in ctx->gbuf. * Registers callable symbols (Function/Method/Class) in ctx->registry. */ @@ -1440,6 +1448,11 @@ void cbm_pipeline_set_committed_counts(cbm_pipeline_t *p, int nodes, int edges); * before the cancellation check + atomic replace. Not part of the public API. */ void cbm_pipeline_set_before_publish_hook_for_tests( cbm_pipeline_t *p, void (*hook)(cbm_pipeline_t *, const char *, void *), void *ctx); +/* Test seam: invoked with the complete graph buffer just before the full + * route dumps it (the dump releases the buffer's indexes). */ +void cbm_pipeline_set_before_dump_hook_for_tests(cbm_pipeline_t *p, + void (*hook)(const cbm_gbuf_t *, void *), + void *ctx); void cbm_pipeline_set_rename_hook_for_tests(cbm_pipeline_t *p, int (*hook)(const char *, const char *, void *), void *ctx); diff --git a/tests/test_parallel.c b/tests/test_parallel.c index 2cf6038f0..f2eb59681 100644 --- a/tests/test_parallel.c +++ b/tests/test_parallel.c @@ -958,6 +958,157 @@ TEST(gbuf_next_id_accessors) { PASS(); } +TEST(gbuf_raise_next_id_never_lowers) { + cbm_gbuf_t *gb = cbm_gbuf_new("proj", "/"); + cbm_gbuf_raise_next_id(gb, 100); + ASSERT_EQ(cbm_gbuf_next_id(gb), 100); + cbm_gbuf_raise_next_id(gb, 40); + ASSERT_EQ(cbm_gbuf_next_id(gb), 100); + int64_t id = cbm_gbuf_upsert_node(gb, "Function", "foo", "proj.foo", "f.go", 1, 5, "{}"); + ASSERT_EQ(id, 100); + cbm_gbuf_free(gb); + PASS(); +} + +/* Edge ids in the main buffer stay unique through the parallel phases. + * Merged worker nodes keep their shared ids, and merged edges take fresh ids + * from the main counter after the highest of them, so the main counter can + * end above the shared one (12 of 17 infra corpora, deckhouse by 229,233). + * The sync after each parallel phase used to set the main counter to the + * shared value anyway, and later passes reused the merged edges' ids. Where + * the counter ends depends on worker timing, so the seam pins it: right before + * each sync the hook takes ID_SYNC_PUSH edge ids from the main buffer. */ +enum { ID_FILES = 64, ID_SYNC_PUSH = 3 }; + +typedef struct { + bool push; + int syncs; +} id_sync_probe_t; + +static void id_sync_hook(cbm_gbuf_t *gb, void *ctx) { + id_sync_probe_t *probe = ctx; + probe->syncs++; + const cbm_gbuf_node_t **fns = NULL; + int n = 0; + if (!probe->push || cbm_gbuf_find_by_label(gb, "Function", &fns, &n) != 0 || n < 2) { + return; + } + char props[CBM_SZ_64]; + for (int i = 0; i < ID_SYNC_PUSH; i++) { + snprintf(props, sizeof(props), "{\"id_sync_probe\":%d}", probe->syncs * ID_SYNC_PUSH + i); + cbm_gbuf_insert_edge(gb, fns[0]->id, fns[1]->id, "CALLS", props); + } +} + +typedef struct { + int edges; + int duplicates; +} edge_id_probe_t; + +static void probe_edge_ids(const cbm_gbuf_t *gb, void *ctx) { + edge_id_probe_t *probe = ctx; + probe->edges = cbm_gbuf_edge_count(gb); + probe->duplicates = cbm_gbuf_duplicate_edge_ids(gb); +} + +/* `count` Go files; each function calls the next file's. */ +static bool write_id_fixture(const char *repo, int count) { + char path[512]; + char source[256]; + for (int i = 0; i < count; i++) { + snprintf(path, sizeof(path), "%s/f%02d.go", repo, i); + snprintf(source, sizeof(source), + "package main\n\nfunc F%02d() int {\n\treturn F%02d() + 1\n}\n", i, + (i + 1) % count); + if (th_write_file(path, source) != 0) { + return false; + } + } + return true; +} + +TEST(parallel_full_index_keeps_edge_ids_unique) { + char repo[256]; + snprintf(repo, sizeof(repo), "/tmp/cbm_par_ids_XXXXXX"); + ASSERT_NOT_NULL(cbm_mkdtemp(repo)); + /* More files than MIN_FILES_FOR_PARALLEL (50): the parallel route. */ + if (!write_id_fixture(repo, ID_FILES)) { + th_rmtree(repo); + FAIL("failed to write the edge-id fixture"); + } + char db_path[512]; + snprintf(db_path, sizeof(db_path), "%s/ids.db", repo); + + char *old_workers = getenv("CBM_WORKERS"); + char *saved_workers = old_workers ? strdup(old_workers) : NULL; + cbm_setenv("CBM_WORKERS", "4", 1); + id_sync_probe_t syncs = {.push = true}; + edge_id_probe_t probe = {0}; + cbm_pipeline_set_id_sync_hook_for_tests(id_sync_hook, &syncs); + cbm_pipeline_t *p = cbm_pipeline_new(repo, db_path, CBM_MODE_FULL); + int rc = CBM_NOT_FOUND; + if (p) { + cbm_pipeline_set_before_dump_hook_for_tests(p, probe_edge_ids, &probe); + rc = cbm_pipeline_run(p); + cbm_pipeline_free(p); + } + cbm_pipeline_set_id_sync_hook_for_tests(NULL, NULL); + if (saved_workers) { + cbm_setenv("CBM_WORKERS", saved_workers, 1); + free(saved_workers); + } else { + cbm_unsetenv("CBM_WORKERS"); + } + th_rmtree(repo); + + ASSERT_EQ(rc, 0); + /* Both full-route syncs (after extract, after resolve) go through the helper. */ + ASSERT_EQ(syncs.syncs, 2); + ASSERT_GT(probe.edges, ID_FILES); + ASSERT_EQ(probe.duplicates, 0); + PASS(); +} + +/* The closure-repair route always runs the parallel phases; both of its + * syncs go through the same helper. */ +TEST(parallel_closure_repair_syncs_ids_through_the_helper) { + char repo[256]; + snprintf(repo, sizeof(repo), "/tmp/cbm_par_ids_XXXXXX"); + ASSERT_NOT_NULL(cbm_mkdtemp(repo)); + if (!write_id_fixture(repo, 3)) { + th_rmtree(repo); + FAIL("failed to write the edge-id fixture"); + } + char db_path[512]; + snprintf(db_path, sizeof(db_path), "%s/ids.db", repo); + cbm_pipeline_incremental_test_reset_faults(); + cbm_pipeline_t *first = cbm_pipeline_new(repo, db_path, CBM_MODE_FULL); + int first_rc = first ? cbm_pipeline_run(first) : CBM_NOT_FOUND; + cbm_pipeline_free(first); + + /* A body-only edit (no new names) routes to closure repair. */ + char path[512]; + snprintf(path, sizeof(path), "%s/f01.go", repo); + int write_rc = + th_write_file(path, "package main\n\nfunc F01() int {\n\treturn F02() + 40 + 2\n}\n"); + id_sync_probe_t syncs = {.push = false}; + cbm_pipeline_set_id_sync_hook_for_tests(id_sync_hook, &syncs); + cbm_pipeline_incremental_test_reset_faults(); + cbm_pipeline_t *second = cbm_pipeline_new(repo, db_path, CBM_MODE_FULL); + int second_rc = second ? cbm_pipeline_run(second) : CBM_NOT_FOUND; + cbm_incremental_route_t route = cbm_pipeline_incremental_test_last_route(); + cbm_pipeline_free(second); + cbm_pipeline_set_id_sync_hook_for_tests(NULL, NULL); + th_rmtree(repo); + + ASSERT_EQ(first_rc, 0); + ASSERT_EQ(write_rc, 0); + ASSERT_EQ(second_rc, 0); + ASSERT_EQ(route, CBM_INCREMENTAL_ROUTE_CLOSURE_REPAIR); + ASSERT_EQ(syncs.syncs, 2); + PASS(); +} + /* ── Parallel-pipeline LSP-override regression ────────────────────── */ /* Pin the wiring fix that unified pass_calls.c (sequential) and * pass_parallel.c (parallel) on cbm_pipeline_find_lsp_resolution + @@ -4768,6 +4919,9 @@ SUITE(parallel) { RUN_TEST(gbuf_merge_empty_src); RUN_TEST(gbuf_merge_src_free_safe); RUN_TEST(gbuf_next_id_accessors); + RUN_TEST(gbuf_raise_next_id_never_lowers); + RUN_TEST(parallel_full_index_keeps_edge_ids_unique); + RUN_TEST(parallel_closure_repair_syncs_ids_through_the_helper); /* Parallel pipeline parity tests */ RUN_TEST(parallel_node_count);