Break-glass merge PR #2607

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