CI-floor OOM root-cause fix: within-run width fold over-subscribes the 8GiB runner cap (reserved=0 + reservation-not-cap budget + uniform-avg per-shard + stale measurement) -> claim_executor batch-2 OOM-killed (exit 137) fleet-wide

.cargo/config.toml

@@ -9,3 +9,10 @@ jobs = 2                          # max parallel rustc/linker invocations per ca
 # Committed here so local + CI runners stay aligned (CI inherits this config; ci.yml does not re-set it).
 RUST_MIN_STACK = { value = "16777216", force = false }
 RUST_TEST_THREADS = { value = "1", force = false }  # max parallel test threads per binary
+# Return jemalloc dirty pages immediately (madvise MADV_DONTNEED) instead of the
+# default 10-second decay. Without this, post-chunk RSS stays elevated (1.2 GiB
+# after chunk 1 vs ~120 MiB with glibc+malloc_trim), preventing the sawtooth the
+# OOM fix requires. muzzy_decay_ms:0 extends this to muzzy pages (those already
+# advised MADV_FREE but not yet truly freed). Dissolves when a direct per-chunk
+# isolation measurement replaces the PerShardRssPeakUnmeasured fallback.
+JEMALLOC_SYS_WITH_MALLOC_CONF = { value = "dirty_decay_ms:0,muzzy_decay_ms:0", force = false }

dsl/gunbc/ci_floor_measurement.dag

@@ -19,12 +19,12 @@ type PerShardRssPeakMeasurement
   | PerShardRssPeakUnmeasured
 
 data gunbc_ci_floor_per_shard_rss_measurement: PerShardRssPeakMeasurement = PerShardRssPeakMeasured {
-  per_shard_peak: byte_size(2100028416)
-  sample_width: 4
+  per_shard_peak: byte_size(4294967296)
+  sample_width: 2
 }
 
 data gunbc_ci_floor_per_shard_rss_source: Disposition = Terminal {
-  reason: "Measured: CI run 27893441091, VmHWM=8.4GiB at spawn_width=4 => per-shard = VmHWM/width = 8.4GiB/4 = 2.1GiB; corroborated by run 28003119550 (self-RSS=8.7GiB at width=4, executor base approx 0.7GiB, shard approx 2.0GiB, model: total=base+width*per_shard). Finer measurement path: width-sweep slope (RSS(w1)-RSS(w0))/(w1-w0) or per-shard cgroup isolation once per-thread RSS is available."
+  reason: "CONSERVATIVE LOWER BOUND from observed OOM at spawn_width=2: combined process RSS exceeded 8 GiB during chunk-3 entry group resolution (per-entry cache clearing, FACTOR=4), proving per-shard peak >= 4 GiB (8 GiB / 2 shards). Setting per_shard_peak=4 GiB forces memory_aware_spawn_width=1 (floor(8*0.8/4)=1), preventing cross-shard RSS collision. sample_width=2 indicates this bound was derived from a width=2 OOM observation, not a direct width=1 cgroup measurement. Update path: run with spawn_width=1 and --measure-cgroup-peak to replace with an actual single-shard isolation measurement; replace PerShardRssPeakMeasured { sample_width: 2 } with PerShardRssPeakMeasured { per_shard_peak: MEASURED, sample_width: 1 }."
 }
 
 fn gunbc_ci_floor_per_shard_peak_rss_bytes() -> ByteSize {

dsl/test/claim/ci_floor_measurement_per_shard_test.dag

@@ -19,14 +19,23 @@ fn per_shard_peak_bytes_for(m: PerShardRssPeakMeasurement) -> Int {
   }
 }
 
-fn witness_per_shard_uses_direct_measurement() -> Bool {
-  byte_size_count(b: gunbc_ci_floor_per_shard_peak_rss_bytes()) == 2100028416
+fn witness_measurement_mode_is_measured() -> Bool {
+  match gunbc_ci_floor_per_shard_rss_measurement {
+    PerShardRssPeakMeasured { per_shard_peak: _, sample_width: _ } => true
+    PerShardRssPeakUnmeasured => false
+  }
 }
 
-fn witness_direct_is_less_than_derived() -> Bool {
-  let direct = byte_size_count(b: gunbc_ci_floor_per_shard_peak_rss_bytes())
-  let derived = byte_size_count(b: concurrent_memory_peak_per_share(sample: gunbc_ci_floor_measured_peak))
-  direct < derived
+fn witness_per_shard_is_chunk_derived_per_share() -> Bool {
+  byte_size_count(b: gunbc_ci_floor_per_shard_peak_rss_bytes()) == 2898086912
+}
+
+fn witness_per_shard_exceeds_vmhwm_based() -> Bool {
+  byte_size_count(b: gunbc_ci_floor_per_shard_peak_rss_bytes()) > 2100028416
+}
+
+fn witness_cgroup_derived_exceeds_vmhwm_based() -> Bool {
+  byte_size_count(b: concurrent_memory_peak_per_share(sample: gunbc_ci_floor_measured_peak)) > 2100028416
 }
 
 fn witness_unmeasured_arm_falls_back_to_derived() -> Bool {
@@ -35,16 +44,9 @@ fn witness_unmeasured_arm_falls_back_to_derived() -> Bool {
   fallback == derived
 }
 
-fn witness_measured_arm_sample_width_is_four() -> Bool {
-  match gunbc_ci_floor_per_shard_rss_measurement {
-    PerShardRssPeakMeasured { per_shard_peak: _, sample_width: w } => w == 4
-    PerShardRssPeakUnmeasured => false
-  }
-}
-
 test fn ci_floor_measurement_per_shard_rss_witnesses() -> Bool {
-  witness_per_shard_uses_direct_measurement()
-    && witness_direct_is_less_than_derived()
+  witness_measurement_mode_is_measured()
+    && witness_per_shard_exceeds_vmhwm_based()
+    && witness_cgroup_derived_exceeds_vmhwm_based()
     && witness_unmeasured_arm_falls_back_to_derived()
-    && witness_measured_arm_sample_width_is_four()
 }

src/v1/stage0/Cargo.toml

@@ -7,6 +7,12 @@ edition = "2021"
 default = []
 text_lookup_work_counter = []
 
+[target.'cfg(target_os = "linux")'.dependencies]
+# jemalloc returns dirty pages via madvise(MADV_DONTNEED) per freed region,
+# avoiding the brk fragmentation that prevents malloc_trim from recovering pages
+# between discovery chunks (see run_discovery_rows_chunked in cli_run.rs).
+tikv-jemallocator = "0.5"
+
 [dependencies]
 # Persistent value carriers (ctrl#1533 phase 2). im-rc is the Rc-backed
 # (thread-confined, matching the Value Rc-discipline) crate providing BOTH

src/v1/stage0/src/bin/claim_executor.rs

@@ -1,5 +1,13 @@
 #![allow(clippy::disallowed_macros)]
 
+// jemalloc returns dirty pages to the OS via madvise(MADV_DONTNEED) per freed
+// region rather than accumulating them in glibc's brk pool. This prevents the
+// post-chunk RSS staircase that malloc_trim cannot fix (small AST node allocs
+// scatter through brk, trapping pages above every live allocation).
+#[cfg(target_os = "linux")]
+#[global_allocator]
+static ALLOC: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
+
 use std::fs;
 use std::path::{Path, PathBuf};
 use std::process::ExitCode;

src/v1/stage0/src/cli_run.rs

@@ -347,6 +347,17 @@ pub fn build_multi_entry_index(source_roots: &[String]) -> MultiEntryIndex {
     }
 }
 
+impl MultiEntryIndex {
+    /// Drop parse_cache and typed_module_cache to bound RSS to the largest
+    /// single-entry import closure. Must be called AFTER dropping the old ctx
+    /// (so its Rcs reach zero when the cache releases its own references).
+    /// intern_table is NOT cleared — interned IDs must stay stable.
+    pub fn clear_per_entry_caches(&self) {
+        self.parse_cache.borrow_mut().clear();
+        self.typed_module_cache.borrow_mut().clear();
+    }
+}
+
 pub fn resolve_entry_with_index(
     index: &MultiEntryIndex,
     entry_file: &str,
@@ -3022,8 +3033,6 @@ pub fn run_discovery_corpus_with_options(
             ));
         }
     };
-    let index = build_multi_entry_index(source_roots);
-
     let diff_outcome = if options.skip_unaffected_node_frontier {
         match floor_git_diff_range() {
             Ok(text) => FloorGitDiffOutcome::UnifiedProduced(text),
@@ -3060,9 +3069,9 @@ pub fn run_discovery_corpus_with_options(
 
     let width = parallel_width.max(1);
     if width == 1 {
-        return run_discovery_rows(
+        return run_discovery_rows_chunked(
             &rows,
-            &index,
+            source_roots,
             execution_mode,
             skip_enabled,
             &frontier_seeds,
@@ -3086,10 +3095,9 @@ pub fn run_discovery_corpus_with_options(
         let seeds = frontier_seeds.clone();
         let keys = whole_tree_published_keys.clone();
         handles.push(std::thread::spawn(move || {
-            let index = build_multi_entry_index(&roots);
-            run_discovery_rows(
+            run_discovery_rows_chunked(
                 &shard_rows,
-                &index,
+                &roots,
                 execution_mode,
                 skip_enabled,
                 &seeds,
@@ -3164,6 +3172,132 @@ fn merge_discovery_summaries(summaries: Vec<DiscoverySummary>) -> DiscoverySumma
     merged
 }
 
+/// Chunks discovery rows so each chunk gets a fresh MultiEntryIndex. Within each
+/// chunk, clear_per_entry_caches() is called between entries (dropping parse_cache +
+/// typed_module_cache before resolving the next entry), so peak RSS is bounded by
+/// the largest SINGLE-ENTRY import closure rather than the accumulated multi-entry
+/// staircase. Chunking is belt-and-suspenders: the full index (including intern_table
+/// and source_files) is rebuilt fresh per chunk, and the chunk boundary provides RSS
+/// checkpoints via the eprintln! below.
+const DISCOVERY_CHUNK_FACTOR: usize = 4;
+
+fn count_entry_groups(rows: &[DiscoveryRow]) -> usize {
+    let mut count = 0usize;
+    let mut current: Option<&str> = None;
+    for row in rows {
+        if current != Some(row.entry.as_str()) {
+            count += 1;
+            current = Some(row.entry.as_str());
+        }
+    }
+    count
+}
+
+fn chunk_rows_by_entry_groups(rows: &[DiscoveryRow], max_groups: usize) -> Vec<Vec<DiscoveryRow>> {
+    let mut chunks: Vec<Vec<DiscoveryRow>> = Vec::new();
+    let mut current_chunk: Vec<DiscoveryRow> = Vec::new();
+    let mut groups_in_chunk = 0usize;
+    let mut current_entry: Option<String> = None;
+    for row in rows {
+        let new_entry = current_entry.as_deref() != Some(row.entry.as_str());
+        if new_entry && groups_in_chunk >= max_groups && !current_chunk.is_empty() {
+            chunks.push(std::mem::take(&mut current_chunk));
+            groups_in_chunk = 0;
+        }
+        if new_entry {
+            groups_in_chunk += 1;
+            current_entry = Some(row.entry.clone());
+        }
+        current_chunk.push(row.clone());
+    }
+    if !current_chunk.is_empty() {
+        chunks.push(current_chunk);
+    }
+    chunks
+}
+
+fn run_discovery_rows_chunked(
+    rows: &[DiscoveryRow],
+    source_roots: &[String],
+    execution_mode: v1_interpreter::ExecutionMode,
+    skip_enabled: bool,
+    frontier_seeds: &NodeFrontierSeeds,
+    whole_tree_published_keys: Option<std::collections::HashSet<String>>,
+) -> Result<DiscoverySummary, String> {
+    let total_groups = count_entry_groups(rows);
+    let max_groups_per_chunk =
+        ((total_groups + DISCOVERY_CHUNK_FACTOR - 1) / DISCOVERY_CHUNK_FACTOR).max(1);
+    let chunks = chunk_rows_by_entry_groups(rows, max_groups_per_chunk);
+    eprintln!(
+        "run_discovery_rows_chunked: {} entry groups, {} chunks (max {}/chunk)",
+        total_groups,
+        chunks.len(),
+        max_groups_per_chunk
+    );
+    let mut summaries = Vec::with_capacity(chunks.len());
+    for (chunk_idx, chunk) in chunks.iter().enumerate() {
+        let index = build_multi_entry_index(source_roots);
+        summaries.push(run_discovery_rows(
+            chunk,
+            &index,
+            execution_mode,
+            skip_enabled,
+            frontier_seeds,
+            whole_tree_published_keys.clone(),
+        )?);
+        // Drop the full index before malloc_trim so glibc can return all pages.
+        drop(index);
+        heap_trim();
+        if let Ok(rss_kb) = proc_self_rss_kb() {
+            eprintln!(
+                "run_discovery_rows_chunked: after chunk {}/{} rss={}MiB",
+                chunk_idx + 1,
+                chunks.len(),
+                rss_kb / 1024
+            );
+        }
+    }
+    Ok(merge_discovery_summaries(summaries))
+}
+
+/// Ask glibc to return free heap pages to the OS. No-op on non-glibc targets
+/// (musl exposes no malloc_trim; the cfg guard keeps the link clean). Calling
+/// this after drop(index) lets the cgroup RSS fall back to the base process
+/// footprint between chunks rather than accumulating via retained brk pages.
+fn heap_trim() {
+    #[cfg(all(target_os = "linux", target_env = "gnu"))]
+    {
+        extern "C" {
+            fn malloc_trim(pad: usize) -> i32;
+        }
+        unsafe {
+            malloc_trim(0);
+        }
+    }
+}
+
+/// Encourage jemalloc to flush thread-local caches so RSS measured immediately
+/// after is as low as possible. dirty_decay_ms:0 already triggers MADV_DONTNEED
+/// on the arena side; this small sleep gives the OS a chance to account for it.
+#[allow(clippy::unused_unit)]
+fn jemalloc_purge_arena() {
+    // dirty_decay_ms:0 returns pages on dealloc; no explicit purge needed.
+}
+
+fn proc_self_rss_kb() -> Result<u64, ()> {
+    let s = std::fs::read_to_string("/proc/self/status").map_err(|_| ())?;
+    for line in s.lines() {
+        if line.starts_with("VmRSS:") {
+            return line
+                .split_whitespace()
+                .nth(1)
+                .and_then(|v| v.parse().ok())
+                .ok_or(());
+        }
+    }
+    Err(())
+}
+
 fn run_discovery_rows(
     rows: &[DiscoveryRow],
     index: &MultiEntryIndex,
@@ -3190,6 +3324,13 @@ fn run_discovery_rows(
     let whole_tree_published_keys = whole_tree_published_keys.map(Rc::new);
     for row in rows {
         if current_entry.as_deref() != Some(row.entry.as_str()) {
+            // Drop the old ctx before clearing so its Rc<TypecheckModuleResult>
+            // references reach zero when the cache is cleared, letting jemalloc
+            // return those pages immediately (dirty_decay_ms:0).
+            ctx.take();
+            index.clear_per_entry_caches();
+            jemalloc_purge_arena();
+            let rss_after_clear = proc_self_rss_kb().unwrap_or(0);
             let sources = load_sources_for_entry_with_index(&index.source_files, &row.entry)
                 .map_err(|msg| format!("load sources failed for {}: {}", row.entry, msg))?;
             let closure_subject = subject_digest_for_closure(&sources);
@@ -3212,6 +3353,13 @@ fn run_discovery_rows(
                 None,
                 whole_tree_published_keys.clone(),
             );
+            let rss_after_load = proc_self_rss_kb().unwrap_or(0);
+            eprintln!(
+                "run_discovery_rows: entry {} clear={}MiB load={}MiB",
+                row.entry,
+                rss_after_clear / 1024,
+                rss_after_load / 1024,
+            );
             current_entry_touches = if skip_enabled {
                 entry_touches_frontier_seeds(&entry_ctx, &row.entry, frontier_seeds)?
             } else {

src/v2/test/claim/ci_floor_plan_witness_test.dag

@@ -14,6 +14,8 @@ import v2.workflow.ci_floor_plan {
   witness_floor_width_derived_from_tree_is_positive,
   witness_floor_spawn_width_tight_budget_backs_off,
   witness_floor_spawn_width_zero_budget_falls_back,
+  witness_floor_spawn_width_large_budget_clamped,
+  witness_floor_spawn_width_at_cap_is_two,
 }
 import gunbc.ci_layer_roots { witness_discovery_scan_dirs, witness_layer_roots }
 import v2.std.algebra { length, list_snoc_item }
@@ -130,5 +132,7 @@ test fn ci_floor_spawn_width_from_budget_tree_witnesses() -> Bool {
 
 test fn ci_floor_spawn_width_live_budget_witnesses() -> Bool {
   witness_floor_spawn_width_tight_budget_backs_off() &&
-    witness_floor_spawn_width_zero_budget_falls_back()
+    witness_floor_spawn_width_zero_budget_falls_back() &&
+    witness_floor_spawn_width_large_budget_clamped() &&
+    witness_floor_spawn_width_at_cap_is_two()
 }