Measure the floor cost distribution near the 500ms ceiling: the plateau is real, the promotion mechanism is not, and #10133 was an outlier

dag/gunbc/floor_cost_distribution.dag

@@ -0,0 +1,226 @@
+module gunbc.floor_cost_distribution
+
+import std.types { Bool, Int, String, NonEmptyStr }
+import std.measure { Millisecond, millisecond, millisecond_count }
+
+// THE DERIVATIONS BEHIND docs/plans/floor-cost-distribution-near-the-ceiling.md, SO THAT DOCUMENT
+// CITES A PRODUCER INSTEAD OF CARRYING COPIED FIGURES.
+//
+// DESIGN section 6: name the instrument, never transcribe its output -- and if a measurement is
+// worth re-deriving it is worth an entry point. The first version of that memo named
+// `gunbc.witness_floor_workflow`, which produces the RAW ROWS and performs none of the analysis
+// below; the ten-run intersection, the band histogram, the counterfactual replay and the inflation
+// percentiles existed only as prose plus numbers. Review 58983 refused it on exactly that ground
+// and was right to: a reader could check the arithmetic of nothing.
+//
+// EVERY FUNCTION HERE IS PURE AND TAKES ITS ROWS AS AN ARGUMENT. Reading the artifacts off disk is
+// a separate fact and lives in `tools.floor_cost_distribution_instrument`, which is the §3 split
+// between the interface (what the analysis IS) and the realization (where the bytes come from).
+// The consequence that matters for evidence: every derivation below is witnessable on a hand-built
+// fixture with no run artifacts present, which is what
+// `test.claim.floor_cost_distribution_witness` does.
+//
+// THE INPUT IS THE `required-floor-claim-cost` ARTIFACT that every required-floor run uploads --
+// the COMPLETE executed population, not the 25-row `[over-cost]` preview the job log prints.
+
+type FloorCostRow {
+  identity: String
+  module: String
+  cpu_ms: Millisecond
+}
+
+// One run's population, named by its run id so a finding can cite which runs it rests on.
+type RunCost {
+  run: NonEmptyStr
+  rows: List<FloorCostRow>
+}
+
+// A TSV LINE IS EITHER A ROW OR IT IS NOT, AND "NOT" IS NEVER A FABRICATED ROW. The artifact
+// carries a `#`-prefixed summary line and a header line before the data, and a short or
+// unparseable line is refused rather than defaulted to zero -- a zero would sit silently at the
+// bottom of every band and inflate the safe population.
+fn parse_claim_cost_line(line: String) -> FloorCostRow? {
+  if starts_with(s: line, prefix: "#") {
+    none
+  } else {
+    let fields = split(s: line, delimiter: "\t")
+    if count(fields) < 6 {
+      none
+    } else if fields[0] == "identity" {
+      none
+    } else {
+      match parse_int(s: fields[5]) {
+        Absent => none
+        Present { value: cpu } => Present {
+          value: FloorCostRow {
+            identity: fields[0],
+            module: fields[1],
+            cpu_ms: millisecond(count: cpu)
+          }
+        }
+      }
+    }
+  }
+}
+
+fn parse_claim_cost_tsv(text: String) -> List<FloorCostRow> {
+  split(s: text, delimiter: "\n")
+    |> flat_map(line => match parse_claim_cost_line(line: line) {
+         Absent => []
+         Present { value: row } => [row]
+       })
+}
+
+// THE BAND HISTOGRAM (question 1: is the population near the ceiling dense).
+type CostBand {
+  lower_ms: Millisecond
+  upper_ms: Millisecond
+  rows: Int
+}
+
+// COMPARISON GOES THROUGH `millisecond_count` RATHER THAN OVER THE CARRIER, because the carrier is
+// what stops a figure read from one clock being compared against a figure read from another; the
+// count is the projection a comparison is entitled to.
+fn rows_in_band(rows: List<FloorCostRow>, lower_ms: Millisecond, upper_ms: Millisecond) -> Int {
+  count(rows |> filter(r =>
+    millisecond_count(m: r.cpu_ms) >= millisecond_count(m: lower_ms)
+      && millisecond_count(m: r.cpu_ms) < millisecond_count(m: upper_ms)))
+}
+
+fn rows_at_or_above(rows: List<FloorCostRow>, ms: Millisecond) -> Int {
+  count(rows |> filter(r => millisecond_count(m: r.cpu_ms) >= millisecond_count(m: ms)))
+}
+
+fn cost_bands(rows: List<FloorCostRow>, edges: List<Millisecond>) -> List<CostBand> {
+  enumerate(edges)
+    |> flat_map(e =>
+         if e.first + 1 < count(edges) {
+           [CostBand {
+              lower_ms: e.second,
+              upper_ms: edges[e.first + 1],
+              rows: rows_in_band(rows: rows, lower_ms: e.second, upper_ms: edges[e.first + 1])
+            }]
+         } else { [] })
+}
+
+// CROSSINGS AND THE COUNTERFACTUAL REPLAY.
+//
+// `excluded_modules` is what makes the replay possible: it answers "what would this run have done
+// if these modules had been repaired", which is the only way to separate a repair's effect from
+// the ambient contention of the run it landed on. It is NOT a claim that excluded rows cost zero;
+// it is a claim about which rows could have crossed.
+fn row_is_excluded(row: FloorCostRow, excluded_modules: List<String>) -> Bool {
+  any(excluded_modules, m => m == row.module)
+}
+
+fn run_crossings(run: RunCost, ceiling_ms: Millisecond, excluded_modules: List<String>) -> List<String> {
+  run.rows
+    |> filter(r => millisecond_count(m: r.cpu_ms) >= millisecond_count(m: ceiling_ms)
+         && !row_is_excluded(row: r, excluded_modules: excluded_modules))
+    |> map(r => r.identity)
+}
+
+fn run_worst_cpu(run: RunCost, excluded_modules: List<String>) -> Millisecond {
+  fold(
+    run.rows |> filter(r => !row_is_excluded(row: r, excluded_modules: excluded_modules)),
+    init: millisecond(count: 0),
+    f: (worst, r) => if millisecond_count(m: r.cpu_ms) > millisecond_count(m: worst) { r.cpu_ms } else { worst }
+  )
+}
+
+// A FLOOR RUN REFUSES IF ANY ONE ROW CROSSES, so the run-level verdict is existential over rows
+// and the fleet-level figure is a COUNT OF RUNS, never a count of rows: fifteen crossings in one
+// run and one crossing in another are the same number of refused merges.
+fn runs_refused(runs: List<RunCost>, ceiling_ms: Millisecond, excluded_modules: List<String>) -> Int {
+  count(runs |> filter(run => count(run_crossings(run: run, ceiling_ms: ceiling_ms, excluded_modules: excluded_modules)) > 0))
+}
+
+fn worst_over_runs(runs: List<RunCost>, excluded_modules: List<String>) -> Millisecond {
+  fold(runs, init: millisecond(count: 0), f: (worst, run) => {
+    let w = run_worst_cpu(run: run, excluded_modules: excluded_modules)
+    if millisecond_count(m: w) > millisecond_count(m: worst) { w } else { worst }
+  })
+}
+
+// MAX-OVER-RUNS, the contention-independent ranking. Membership of any single run's crossing set is
+// a function of that run's load as much as of the row, so a candidate population must be ranked on
+// the worst each row has been observed to cost, never on one run's sample.
+fn identity_max_cpu(runs: List<RunCost>, identity: String) -> Millisecond {
+  fold(runs, init: millisecond(count: 0), f: (worst, run) =>
+    fold(run.rows |> filter(r => r.identity == identity), init: worst, f: (w, r) =>
+      if millisecond_count(m: r.cpu_ms) > millisecond_count(m: w) { r.cpu_ms } else { w }))
+}
+
+fn module_max_cpu(runs: List<RunCost>, module: String) -> Millisecond {
+  fold(runs, init: millisecond(count: 0), f: (worst, run) =>
+    fold(run.rows |> filter(r => r.module == module), init: worst, f: (w, r) =>
+      if millisecond_count(m: r.cpu_ms) > millisecond_count(m: w) { r.cpu_ms } else { w }))
+}
+
+// THE REPAIR LADDER: what removing one more module actually buys, which is the quantity that
+// decides whether per-producer repair is the right unit of work.
+type LadderStep {
+  removed: String
+  worst_ms: Millisecond
+  runs_refused: Int
+}
+
+fn repair_ladder(
+  runs: List<RunCost>,
+  ceiling_ms: Millisecond,
+  already_excluded: List<String>,
+  order: List<String>
+) -> List<LadderStep> {
+  enumerate(order)
+    |> map(e => {
+         let excluded = concat(already_excluded, order |> take(n: e.first + 1))
+         LadderStep {
+           removed: e.second,
+           worst_ms: worst_over_runs(runs: runs, excluded_modules: excluded),
+           runs_refused: runs_refused(runs: runs, ceiling_ms: ceiling_ms, excluded_modules: excluded)
+         }
+       })
+}
+
+// RUN-TO-RUN INFLATION, PAIRED ON IDENTITY.
+//
+// CARRIED IN PERMILLE AS AN Int RATHER THAN AS A RATIO, deliberately. The quantity is a ratio, but
+// an exact integer permille is comparable and orderable without a float, and the percentile below
+// is an ORDER STATISTIC -- it selects an observed value and never interpolates between two, so no
+// figure it returns is one this repository did not measure.
+//
+// `min_base_ms` exists because the artifact records whole milliseconds: at a 2ms baseline a
+// one-millisecond scheduling difference is a 50% "inflation" that says nothing about contention.
+// Rows below the floor are dropped rather than smoothed.
+fn identity_inflation_permille(base: RunCost, other: RunCost, min_base_ms: Millisecond) -> List<Int> {
+  base.rows
+    |> filter(b => millisecond_count(m: b.cpu_ms) >= millisecond_count(m: min_base_ms))
+    |> flat_map(b => match other.rows |> filter(o => o.identity == b.identity) |> first {
+         Absent => []
+         Present { value: o } => [(millisecond_count(m: o.cpu_ms) * 1000) / millisecond_count(m: b.cpu_ms)]
+       })
+}
+
+// An order statistic over the observed values: `percent` is 0-100 and the result is always a
+// member of the input, so it is a measurement and not an interpolation.
+fn permille_percentile(values: List<Int>, percent: Int) -> Int? {
+  let sorted = values |> sort_by(v => v)
+  let n = count(sorted)
+  if n == 0 {
+    none
+  } else {
+    let idx = (percent * n) / 100
+    if idx >= n { Present { value: sorted[n - 1] } } else { Present { value: sorted[idx] } }
+  }
+}
+
+// THE MARGIN THE MEASURED INFLATION IMPLIES: the largest clean-run cost that still lands under the
+// ceiling after inflation. Integer division floors, which rounds the budget DOWN -- the safe
+// direction for a limit.
+fn implied_clean_run_budget(ceiling_ms: Millisecond, inflation_permille: Int) -> Millisecond {
+  if inflation_permille <= 0 {
+    ceiling_ms
+  } else {
+    millisecond(count: (millisecond_count(m: ceiling_ms) * 1000) / inflation_permille)
+  }
+}