fix(pipeline): bound cross-file preparation memory

docs/INDEX_RESOURCE_LIMITS.md

@@ -78,6 +78,20 @@ its measurement is skipped, because a total without a live member would
 under-report. On Linux the probe reads every `/proc/<pid>/stat` to find the
 tree's members, so a `/proc` mounted with `hidepid=1` makes it fail closed.
 
+## Internal cross-file memory budget
+
+The internal `CBM_MEM_BUDGET_MB` budget also applies when extraction results
+are gathered for cross-file resolution. The collector checks the remaining
+charged-memory budget before allocating the project-wide definition array and
+periodically while copying definitions. A budget refusal aborts the staged
+update and preserves the serving index. The parallel driver also checks the
+budget between shared registry builders.
+
+C# and Java/Kotlin registry scratch arrays scale with definitions of their
+own languages, rather than every definition in the repository. This avoids
+large unrelated data files multiplying those temporary allocations. These
+checks remain cooperative; they do not replace the process-tree watchdog.
+
 ## Counting and failure semantics
 
 `index_max_files` counts a file only after it passes directory pruning, ignore

internal/cbm/lsp/cs_lsp.c

@@ -3939,8 +3939,12 @@ CBMTypeRegistry *cbm_cs_build_cross_registry(CBMArena *arena, CBMLSPDef *defs, i
      * NULL, which must not be mistaken for OOM — the empty registry is still
      * built, finalized, and shared. */
     CBMLSPDef *cs = NULL;
-    if (def_count > 0) {
-        cs = (CBMLSPDef *)cbm_arena_alloc(arena, (size_t)def_count * sizeof(*cs));
+    int cs_count = 0;
+    for (int i = 0; i < def_count; i++) {
+        cs_count += defs[i].lang == CBM_LANG_CSHARP;
+    }
+    if (cs_count > 0) {
+        cs = (CBMLSPDef *)cbm_arena_alloc(arena, (size_t)cs_count * sizeof(*cs));
         if (!cs)
             return NULL;
     }

internal/cbm/lsp/java_lsp.c

@@ -3850,8 +3850,12 @@ CBMTypeRegistry *cbm_java_build_cross_registry(CBMArena *arena, CBMLSPDef *defs,
      * NULL, which must not be mistaken for OOM — the empty registry is still
      * built, finalized, and shared. */
     CBMLSPDef *jvm = NULL;
-    if (def_count > 0) {
-        jvm = (CBMLSPDef *)cbm_arena_alloc(arena, (size_t)def_count * sizeof(*jvm));
+    int jvm_count = 0;
+    for (int i = 0; i < def_count; i++) {
+        jvm_count += defs[i].lang == CBM_LANG_JAVA || defs[i].lang == CBM_LANG_KOTLIN;
+    }
+    if (jvm_count > 0) {
+        jvm = (CBMLSPDef *)cbm_arena_alloc(arena, (size_t)jvm_count * sizeof(*jvm));
         if (!jvm) {
             return NULL;
         }

src/foundation/mem.c

@@ -649,6 +649,14 @@ bool cbm_mem_over_budget(void) {
     return charged > g_budget;
 }
 
+bool cbm_mem_would_exceed_budget(size_t additional_bytes) {
+    if (g_budget == 0) {
+        return false;
+    }
+    size_t charged = cbm_mem_charged();
+    return charged > g_budget || additional_bytes > g_budget - charged;
+}
+
 /* "Give memory back now" — true when our own budget is exceeded OR the MACHINE
  * is short, whichever comes first.
  *

src/foundation/mem.h

@@ -104,6 +104,10 @@ void cbm_mem_set_charged_for_tests(size_t bytes);
 /* Returns true if current RSS exceeds the budget. */
 bool cbm_mem_over_budget(void);
 
+/* Reject a known upcoming allocation before touching its pages. A zero
+ * budget means initialization has not run yet. Overflow-safe. */
+bool cbm_mem_would_exceed_budget(size_t additional_bytes);
+
 /* True when memory should be handed back NOW: our budget is exceeded, or the
  * machine itself is short. Use this for spill/reclaim decisions; use
  * cbm_mem_over_budget() for the decision to stop a run. System pressure must

src/pipeline/pass_lsp_cross.c

@@ -17,6 +17,7 @@
 
 #include "pipeline/lsp_surface.h"
 #include "foundation/mem_core.h"
+#include "foundation/mem.h"
 #include "result_spill.h"
 #include "pipeline/pipeline_internal.h"
 #include "pipeline/lsp_resolve.h"
@@ -40,6 +41,7 @@
 
 #include <stdio.h>
 #include <stdint.h>
+#include <limits.h>
 #include <stdlib.h>
 #include <stdatomic.h>
 #include <string.h>
@@ -1186,27 +1188,28 @@ CBMLSPDef *cbm_pxc_collect_all_defs(const cbm_pipeline_ctx_t *ctx, CBMArena *are
                                     int *out_count, int *out_def_starts) {
     const cbm_result_spill_t *spill = ctx ? ctx->spill : NULL;
     bool owner_walk_required = ctx && atomic_load(&ctx->test_definition_owners_seen);
-    int total = 0;
+    CBMLSPDef *defs = NULL;
+    size_t total = 0;
     for (int i = 0; i < file_count; i++) {
-        int defs = 0;
+        int file_defs = 0;
         int impls = 0;
         int records = 0; /* a Python module's namespace records (re-exports) */
         if (cache[i]) {
             owner_walk_required = owner_walk_required || cache[i]->has_test_definition_owners;
-            defs = cache[i]->defs.count;
+            file_defs = cache[i]->defs.count;
             impls = cache[i]->impl_traits.count;
             records = pxc_py_namespace_record_bound(cache[i]);
         } else if (spill && cbm_result_spill_has(spill, i)) {
             /* The header carries the counts (and the fact stream's status and
              * size), so the bound needs no full load. */
             CBMFileResult hdr;
             if (cbm_result_spill_peek_header(spill, i, &hdr)) {
-                defs = hdr.defs.count;
+                file_defs = hdr.defs.count;
                 impls = hdr.impl_traits.count;
                 records = pxc_py_namespace_record_bound(&hdr);
             }
         }
-        total += defs;
+        total += file_defs;
         if (files[i].language == CBM_LANG_RUST) {
             total += impls;
         }
@@ -1221,7 +1224,14 @@ CBMLSPDef *cbm_pxc_collect_all_defs(const cbm_pipeline_ctx_t *ctx, CBMArena *are
         }
         return NULL;
     }
-    CBMLSPDef *defs = (CBMLSPDef *)calloc((size_t)total, sizeof(CBMLSPDef));
+    /* Spilling extraction results does not bound this project-wide array or
+     * its owned strings. Refuse the lower bound before calloc can turn a
+     * data archive into several more gigabytes of anonymous memory. */
+    if (total > INT_MAX || total > SIZE_MAX / sizeof(CBMLSPDef) ||
+        cbm_mem_would_exceed_budget(total * sizeof(CBMLSPDef))) {
+        goto over_budget;
+    }
+    defs = (CBMLSPDef *)calloc(total, sizeof(CBMLSPDef));
     if (!defs) {
         *out_count = 0;
         if (out_def_starts) {
@@ -1232,6 +1242,9 @@ CBMLSPDef *cbm_pxc_collect_all_defs(const cbm_pipeline_ctx_t *ctx, CBMArena *are
     int capacity = total;
     int idx = 0;
     for (int fi = 0; fi < file_count; fi++) {
+        if (cbm_mem_would_exceed_budget(0)) {
+            goto over_budget;
+        }
         if (out_def_starts) {
             out_def_starts[fi] = idx;
         }
@@ -1251,12 +1264,8 @@ CBMLSPDef *cbm_pxc_collect_all_defs(const cbm_pipeline_ctx_t *ctx, CBMArena *are
 #endif
             fr_loaded = fr != NULL;
             if (!fr && owner_walk_required) {
-                free(defs);
                 *out_count = -1;
-                if (out_def_starts) {
-                    memset(out_def_starts, 0, (size_t)(file_count + 1) * sizeof(int));
-                }
-                return NULL;
+                goto collect_failure;
             }
         }
         if (!fr)
@@ -1289,6 +1298,15 @@ CBMLSPDef *cbm_pxc_collect_all_defs(const cbm_pipeline_ctx_t *ctx, CBMArena *are
                                         &bs.imp_count);
         }
         for (int di = 0; di < fr->defs.count; di++) {
+            /* One file can contain millions of JSON entries; a per-file
+             * check alone cannot bound the strings copied into the arena. */
+            if ((di & 255) == 0 && cbm_mem_would_exceed_budget(0)) {
+                cbm_pxc_free_import_map(bs.imp_keys, bs.imp_vals, bs.imp_count);
+                if (fr_loaded) {
+                    cbm_free_result(fr);
+                }
+                goto over_budget;
+            }
             if (pxc_build_lsp_def(arena, &fr->defs.items[di], def_modules[fi], namespace_name,
                                   files[fi].language, &defs[idx], bs.reg ? &bs : NULL) == 0) {
                 idx++;
@@ -1330,6 +1348,16 @@ CBMLSPDef *cbm_pxc_collect_all_defs(const cbm_pipeline_ctx_t *ctx, CBMArena *are
     }
     *out_count = idx;
     return defs;
+
+over_budget:
+    cbm_log_error("mem.budget.exceeded", "phase", "lsp_cross_collect", "action", "fail_whole");
+    *out_count = CBM_PXC_DEFS_OVER_BUDGET;
+collect_failure:
+    free(defs);
+    if (out_def_starts) {
+        memset(out_def_starts, 0, ((size_t)file_count + 1) * sizeof(int));
+    }
+    return NULL;
 }
 
 /* Return the one source import path bound to `local_name`, or NULL when the
@@ -2547,7 +2575,10 @@ int cbm_pipeline_pass_lsp_cross(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *
         cbm_pxc_collect_all_defs(ctx, &ctx->seq_cross_arena, cache, files, file_count,
                                  ctx->project_name, def_modules, &def_count, def_starts);
     if (def_count < 0 || (owner_walk_required && (!all_defs || def_count <= 0))) {
-        atomic_store(&ctx->test_declarations_failed, 1);
+        bool over_budget = def_count == CBM_PXC_DEFS_OVER_BUDGET;
+        if (!over_budget) {
+            atomic_store(&ctx->test_declarations_failed, 1);
+        }
         cbm_free(CBM_MEM_CLASS_EXTRACT, def_starts);
         free(all_defs);
         for (int i = 0; i < file_count; i++) {
@@ -2558,7 +2589,7 @@ int cbm_pipeline_pass_lsp_cross(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *
             cbm_pxc_set_rust_manifest(NULL);
             cbm_arena_destroy(&cargo_arena);
         }
-        return CBM_PIPELINE_ABORT_PRESERVE_DB;
+        return over_budget ? CBM_PIPELINE_ABORT_OVER_BUDGET : CBM_PIPELINE_ABORT_PRESERVE_DB;
     }
     /* Same seam as the parallel driver: serialize per-file surfaces while the
      * result cache is alive. Failure only degrades to a full rebuild on the

src/pipeline/pass_lsp_cross.h

@@ -43,13 +43,18 @@
 /* True iff this language has a cbm_run_X_lsp_cross resolver wired up. */
 bool cbm_pxc_has_cross_lsp(CBMLanguage lang);
 
+/* Collector failure distinct from missing configured test-owner definitions.
+ * Every driver must fail the attempt, preserving the serving generation. */
+#define CBM_PXC_DEFS_OVER_BUDGET (-2)
+
 /* Collect a project-wide CBMLSPDef[] from every cached file result.
  * def_modules[i] receives the module QN for files[i] (malloc'd; the
  * caller frees each entry then the array). String fields in the
  * returned CBMLSPDef[] are borrowed from cache[i]->arena and from
  * def_modules[i] — caller must keep both alive while the array is in
  * use. Returns the malloc'd array (free() it) and writes the entry
- * count to *out_count. Returns NULL on alloc failure or when no defs
+ * count to *out_count. A budget refusal returns NULL with
+ * *out_count = CBM_PXC_DEFS_OVER_BUDGET. Returns NULL on alloc failure or when no defs
  * exist. out_def_starts (optional, file_count + 1 entries, caller-owned)
  * receives per-file prefix offsets: file i's defs occupy
  * [out_def_starts[i], out_def_starts[i+1]) — the LSP-surface serializer

src/pipeline/pipeline.c

@@ -2894,7 +2894,10 @@ static int run_sequential_pipeline(cbm_pipeline_t *p, cbm_pipeline_ctx_t *ctx,
     for (int si = 0; si < PL_SEQ_PASSES && rc == 0; si++) {
         cbm_clock_gettime(CLOCK_MONOTONIC, t);
         int pr = seq_passes[si].fn(ctx, files, file_count);
-        if (pr == CBM_PIPELINE_ABORT_PRESERVE_DB || atomic_load(&ctx->test_declarations_failed)) {
+        if (pr == CBM_PIPELINE_ABORT_OVER_BUDGET) {
+            rc = pr;
+        } else if (pr == CBM_PIPELINE_ABORT_PRESERVE_DB ||
+                   atomic_load(&ctx->test_declarations_failed)) {
             rc = CBM_PIPELINE_ABORT_PRESERVE_DB;
         } else if (pr != 0 && !seq_passes[si].ignore_err) {
             rc = pr;
@@ -3051,6 +3054,7 @@ static int run_parallel_pipeline(cbm_pipeline_t *p, cbm_pipeline_ctx_t *ctx,
     int def_count = 0;
     CBMLSPDef *all_defs = NULL;
     int *def_starts = NULL;
+    CBMModuleDefIndex *module_def_index = NULL;
     /* The collected defs own their strings in this arena (results may be on
      * disk by now); the per-language cross registries share it. */
     CBMArena cross_lsp_arena;
@@ -3064,6 +3068,10 @@ static int run_parallel_pipeline(cbm_pipeline_t *p, cbm_pipeline_ctx_t *ctx,
                                                           def_modules, &def_count, def_starts)
                                : NULL;
     }
+    if (def_count == CBM_PXC_DEFS_OVER_BUDGET) {
+        rc = CBM_PIPELINE_ABORT_OVER_BUDGET;
+        goto cross_lsp_cleanup;
+    }
     if (def_count < 0) {
         atomic_store(&ctx->test_declarations_failed, 1);
         def_count = 0; /* keep metric counts valid; the fatal latch survives */
@@ -3085,23 +3093,25 @@ static int run_parallel_pipeline(cbm_pipeline_t *p, cbm_pipeline_ctx_t *ctx,
         }
     }
     CBM_PROF_END_N("lsp_cross_prepare", "2_surface_rows", t_surfaces, file_count);
-    free(def_starts);
     /* Build inverted index: module_qn → defs. The fused resolve_worker
      * uses this to filter the global all_defs[] down to just the defs
      * each file actually needs (own_module + imported modules) — the
      * gopls "package summary" pattern. Drops per-file registry build
      * cost from O(all_defs) to O(relevant_defs), typically 50-100×
      * smaller per file. */
     CBM_PROF_START(t_module_index);
-    CBMModuleDefIndex *module_def_index =
-        all_defs ? cbm_pxc_build_module_def_index(all_defs, def_count) : NULL;
+    module_def_index = all_defs ? cbm_pxc_build_module_def_index(all_defs, def_count) : NULL;
     CBM_PROF_END_N("lsp_cross_prepare", "3_module_def_index", t_module_index, def_count);
     /* Tier 2 full: pre-build per-language cross-LSP registries.
      * Built ONCE here; shared READ-ONLY across all files of that language
      * during resolve. Per-file work is then: parse + AST walk + O(1) lookups
      * — no registry build, no Phase 1b mutations. Languages added so far:
      * Go, Python, C/C++, C#, TS/JS, Java. Others (Kotlin, PHP) fall back to per-file. */
     CBMCrossLspRegistries cross_registries = {0};
+    if (cbm_mem_would_exceed_budget(0)) {
+        rc = CBM_PIPELINE_ABORT_OVER_BUDGET;
+        goto cross_lsp_cleanup;
+    }
     if (all_defs) {
         /* Per-builder split of lsp_cross_prepare — attributes a slow prepare to
          * ONE language instead of re-diagnosing the whole pass (the cs builder
@@ -3110,23 +3120,47 @@ static int run_parallel_pipeline(cbm_pipeline_t *p, cbm_pipeline_ctx_t *ctx,
         long b_ms[6];
         cbm_clock_gettime(CLOCK_MONOTONIC, &t_b);
         cross_registries.go = cbm_go_build_cross_registry(&cross_lsp_arena, all_defs, def_count);
+        if (cbm_mem_would_exceed_budget(0)) {
+            rc = CBM_PIPELINE_ABORT_OVER_BUDGET;
+            goto cross_lsp_cleanup;
+        }
         b_ms[0] = (long)elapsed_ms(t_b);
         cbm_clock_gettime(CLOCK_MONOTONIC, &t_b);
         cross_registries.python =
             cbm_py_build_cross_registry(&cross_lsp_arena, all_defs, def_count);
+        if (cbm_mem_would_exceed_budget(0)) {
+            rc = CBM_PIPELINE_ABORT_OVER_BUDGET;
+            goto cross_lsp_cleanup;
+        }
         b_ms[1] = (long)elapsed_ms(t_b);
         cbm_clock_gettime(CLOCK_MONOTONIC, &t_b);
         cross_registries.c = cbm_c_build_cross_registry(&cross_lsp_arena, all_defs, def_count);
+        if (cbm_mem_would_exceed_budget(0)) {
+            rc = CBM_PIPELINE_ABORT_OVER_BUDGET;
+            goto cross_lsp_cleanup;
+        }
         b_ms[2] = (long)elapsed_ms(t_b);
         cbm_clock_gettime(CLOCK_MONOTONIC, &t_b);
         cross_registries.cs = cbm_cs_build_cross_registry(&cross_lsp_arena, all_defs, def_count);
+        if (cbm_mem_would_exceed_budget(0)) {
+            rc = CBM_PIPELINE_ABORT_OVER_BUDGET;
+            goto cross_lsp_cleanup;
+        }
         b_ms[3] = (long)elapsed_ms(t_b);
         cbm_clock_gettime(CLOCK_MONOTONIC, &t_b);
         cross_registries.ts = cbm_ts_build_cross_registry(&cross_lsp_arena, all_defs, def_count);
+        if (cbm_mem_would_exceed_budget(0)) {
+            rc = CBM_PIPELINE_ABORT_OVER_BUDGET;
+            goto cross_lsp_cleanup;
+        }
         b_ms[4] = (long)elapsed_ms(t_b);
         cbm_clock_gettime(CLOCK_MONOTONIC, &t_b);
         cross_registries.java =
             cbm_java_build_cross_registry(&cross_lsp_arena, all_defs, def_count);
+        if (cbm_mem_would_exceed_budget(0)) {
+            rc = CBM_PIPELINE_ABORT_OVER_BUDGET;
+            goto cross_lsp_cleanup;
+        }
         b_ms[5] = (long)elapsed_ms(t_b);
         char b_buf[6][CBM_SZ_16];
         const char *b_name[6] = {"go", "python", "c", "cs", "ts", "java"};
@@ -3157,6 +3191,11 @@ static int run_parallel_pipeline(cbm_pipeline_t *p, cbm_pipeline_ctx_t *ctx,
                  itoa_buf((int)elapsed_ms(*t)));
     pipeline_phase_mark("parallel_resolve");
     log_result_arenas("post_resolve", cache, file_count);
+cross_lsp_cleanup:
+    if (rc == CBM_PIPELINE_ABORT_OVER_BUDGET) {
+        cbm_log_error("mem.budget.exceeded", "phase", "lsp_cross_prepare", "action", "fail_whole");
+    }
+    free(def_starts);
     cbm_pxc_free_module_def_index(module_def_index);
     cbm_arena_destroy(&cross_lsp_arena); /* releases all per-lang registries */
     free(all_defs);
@@ -3167,8 +3206,10 @@ static int run_parallel_pipeline(cbm_pipeline_t *p, cbm_pipeline_ctx_t *ctx,
         free(def_modules);
     }
     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);
+    if (rc == 0) {
+        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++) {
         if (cache[i]) {
             cbm_free_result(cache[i]);