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 <martin.vogel.tech@gmail.com>

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

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);
 

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);

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 */

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);

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);