From cfab5d75f635c9e430865f92270e2fb3ab01dd2d Mon Sep 17 00:00:00 2001 From: Martin Vogel Date: Mon, 12 Oct 2026 19:06:56 +0200 Subject: [PATCH] fix(graph): the id counter is never lowered after a parallel phase After a parallel extract or resolve, the main buffer's counter was set to the shared allocator's value. 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. Setting it down handed the merged edges' ids out a second time. All four sync points (full route and closure repair, after extract and after resolve) now go through cbm_pipeline_sync_next_id, which only raises the counter. The delta preseed still pins it with cbm_gbuf_set_next_id, as before. On the 17 infra corpora the old sync left 94,239 duplicate edge ids in the buffer (11 corpora; deckhouse 44,959), and the fix leaves none. The persisted graph does not change, because the dump numbers edges itself. The duplicates matter to anything that finds buffer edges by id: cascade_delete_edges unindexes by id and then frees the edge, so a duplicate can leave a freed edge in an index. Tests: a parallel full index with the counters pinned by a test seam fails with a duplicate edge id without the fix. Both full-route and both closure-repair syncs are checked to run through the helper. gbuf_raise_next_id_never_lowers covers the new buffer call. Signed-off-by: Martin Vogel --- src/graph_buffer/graph_buffer.c | 33 ++++++ src/graph_buffer/graph_buffer.h | 9 +- src/pipeline/pipeline.c | 36 ++++++- src/pipeline/pipeline_incremental.c | 4 +- src/pipeline/pipeline_internal.h | 13 +++ tests/test_parallel.c | 154 ++++++++++++++++++++++++++++ 6 files changed, 244 insertions(+), 5 deletions(-) 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);