rmg-core/scheduler: fail-fast drain; u32 radix histogram; keep PendingTx private

Makefile

@@ -2,6 +2,7 @@ SHELL := /bin/bash
 
 # Default docs port; override with: make docs PORT=5180
 PORT ?= 5173
+BENCH_PORT ?= 8000
 
 .PHONY: hooks docs docs-build docs-ci echo-total
 hooks:
@@ -43,3 +44,75 @@ docs-ci:
 echo-total:
 	@chmod +x scripts/gen-echo-total.sh
 	@./scripts/gen-echo-total.sh
+# Benchmarks and reports
+.PHONY: bench-report vendor-d3 bench-serve bench-open
+
+vendor-d3:
+	@mkdir -p docs/benchmarks/vendor
+	@if [ ! -f docs/benchmarks/vendor/d3.v7.min.js ]; then \
+	  echo "Downloading D3 v7 to docs/benchmarks/vendor..."; \
+	  curl -fsSL https://unpkg.com/d3@7/dist/d3.min.js -o docs/benchmarks/vendor/d3.v7.min.js; \
+	  echo "D3 saved to docs/benchmarks/vendor/d3.v7.min.js"; \
+	else \
+	  echo "D3 already present (docs/benchmarks/vendor/d3.v7.min.js)"; \
+	fi
+
+bench-serve:
+	@echo "Serving repo at http://localhost:$(BENCH_PORT) (Ctrl+C to stop)"
+	@python3 -m http.server $(BENCH_PORT)
+
+bench-open:
+	@open "http://localhost:$(BENCH_PORT)/docs/benchmarks/"
+
+bench-report: vendor-d3
+	@echo "Running benches (rmg-benches)..."
+	cargo bench -p rmg-benches
+	@echo "Starting local server on :$(BENCH_PORT) and opening dashboard..."
+	@mkdir -p target
+	@if [ -f target/bench_http.pid ] && ps -p $$(cat target/bench_http.pid) >/dev/null 2>&1; then \
+	  echo "[bench] Stopping previous server (pid $$(cat target/bench_http.pid))"; \
+	  kill $$(cat target/bench_http.pid) >/dev/null 2>&1 || true; \
+	  rm -f target/bench_http.pid; \
+	fi
+	@/bin/sh -c 'nohup python3 -m http.server $(BENCH_PORT) >/dev/null 2>&1 & echo $$! > target/bench_http.pid'
+	@echo "[bench] Waiting for server to become ready..."
+	@for i in {1..80}; do \
+	  if curl -sSf "http://localhost:$(BENCH_PORT)/" >/dev/null ; then \
+	    echo "[bench] Server is up at http://localhost:$(BENCH_PORT)/" ; \
+	    break ; \
+	  fi ; \
+	  sleep 0.25 ; \
+	done
+	@open "http://localhost:$(BENCH_PORT)/docs/benchmarks/"
+
+.PHONY: bench-status bench-stop
+
+bench-status:
+	@if [ -f target/bench_http.pid ] && ps -p $$(cat target/bench_http.pid) >/dev/null 2>&1; then \
+	  echo "[bench] Server running (pid $$(cat target/bench_http.pid)) at http://localhost:$(BENCH_PORT)"; \
+	else \
+	  echo "[bench] Server not running"; \
+	fi
+
+bench-stop:
+	@if [ -f target/bench_http.pid ]; then \
+	  kill $$(cat target/bench_http.pid) >/dev/null 2>&1 || true; \
+	  rm -f target/bench_http.pid; \
+	  echo "[bench] Server stopped"; \
+	else \
+	  echo "[bench] No PID file at target/bench_http.pid"; \
+	fi
+
+.PHONY: bench-bake bench-open-inline
+
+# Bake a standalone HTML with inline data that works over file://
+bench-bake: vendor-d3
+	@echo "Running benches (rmg-benches)..."
+	cargo bench -p rmg-benches
+	@echo "Baking inline report..."
+	@python3 scripts/bench_bake.py --out docs/benchmarks/report-inline.html
+	@echo "Opening inline report..."
+	@open docs/benchmarks/report-inline.html
+
+bench-open-inline:
+	@open docs/benchmarks/report-inline.html

README.md

@@ -40,8 +40,10 @@ It’s the core of the Echo engine: runtime, assets, networking, and tools all o
 
 ## Developer: Running Benchmarks
 
-- Command: `cargo bench -p rmg-benches`
-- Purpose: Runs Criterion micro-benchmarks for the benches crate (`crates/rmg-benches`).
+- Command (live dashboard): `make bench-report`
+  - Runs `cargo bench -p rmg-benches`, starts a local server, and opens the dashboard at `http://localhost:8000/docs/benchmarks/`.
+- Command (offline static file): `make bench-bake`
+  - Runs benches and bakes `docs/benchmarks/report-inline.html` with results injected so it works over `file://` (no server required).
 - Docs: see `crates/rmg-benches/benches/README.md` for details, tips, and report paths.
 
 ### Core Principles

crates/rmg-benches/Cargo.toml

@@ -10,8 +10,8 @@ description = "Microbenchmarks for Echo (rmg-core): snapshot hashing and schedul
 criterion = { version = "0.5", default-features = false, features = ["html_reports"] }
 # Pin version alongside path to satisfy cargo-deny wildcard bans
 rmg-core = { version = "0.1.0", path = "../rmg-core" }
-# Minor-pin for semver compatibility; benches do not rely on a specific patch.
-blake3 = "1.8"
+# Exact pin and trimmed features to avoid rayon/parallelism in benches.
+blake3 = { version = "=1.8.2", default-features = false, features = ["std"] }
 
 [[bench]]
 name = "motion_throughput"

crates/rmg-benches/benches/README.md

@@ -38,6 +38,12 @@ cargo bench -p rmg-benches --bench scheduler_drain
 
 Criterion HTML reports are written under `target/criterion/<group>/report/index.html`.
 
+### Charts & Reports
+
+- Live server + dashboard: `make bench-report` opens `http://localhost:8000/docs/benchmarks/`.
+- Offline static report (no server): `make bench-bake` writes `docs/benchmarks/report-inline.html` with results injected.
+  - Open the file directly (Finder or `open docs/benchmarks/report-inline.html`).
+
 ## Interpreting Results
 
 - Use the throughput value to sanity‑check the scale of work per iteration.
@@ -48,7 +54,9 @@ Criterion HTML reports are written under `target/criterion/<group>/report/index.
 ## Environment Notes
 
 - Toolchain: `stable` Rust (see `rust-toolchain.toml`).
-- Dependency policy: avoid wildcards; benches use a minor pin for `blake3`.
+- Dependency policy: avoid wildcards; benches use an exact patch pin for `blake3`
+  with trimmed features to avoid incidental parallelism:
+  `blake3 = { version = "=1.8.2", default-features = false, features = ["std"] }`.
 - Repro: keep your machine under minimal background load; prefer `--quiet` and 
   close other apps.
 
@@ -62,4 +70,3 @@ cargo flamegraph -p rmg-benches --bench snapshot_hash -- --sample-size 50
 ```
 
 These tools are not required for CI and are optional for local analysis.
-

crates/rmg-benches/benches/reserve_scaling.rs

@@ -0,0 +1,56 @@
+#![allow(missing_docs)]
+//! Benchmark: reserve() scaling with footprint size and number of reserved rewrites
+//!
+//! Measures how reserve() performance scales with:
+//! 1. Number of previously reserved rewrites (k)
+//! 2. Size of footprint being reserved (m)
+//!
+//! The current GenSet-based implementation should scale as O(m), independent of k.
+//! A naive Vec<Footprint> implementation would scale as O(k × m).
+
+use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion, Throughput};
+use rmg_core::Hash;
+use std::time::Duration;
+
+// Import the scheduler - it's crate-private so we can't access it directly
+// Instead we'll use it through the Engine API
+// Actually, we need direct access for this micro-benchmark, so we'll create
+// a test module inside rmg-core and expose it via a feature flag or just
+// write an integration test instead.
+
+// For now, let's write a simpler benchmark that measures reserve through the Engine API
+
+fn make_hash(val: u8) -> Hash {
+    let mut h = [0u8; 32];
+    h[0] = val;
+    h
+}
+
+// Note: This benchmark requires access to DeterministicScheduler which is crate-private.
+// Moving this to rmg-core/src/scheduler.rs tests module or using a pub(crate) test harness.
+
+fn bench_reserve_scaling(_c: &mut Criterion) {
+    // This is a placeholder - the actual benchmark needs to be in rmg-core
+    // where we can access the scheduler directly.
+
+    // TODO: Implement this properly by either:
+    // 1. Adding a test-only public API to DeterministicScheduler
+    // 2. Moving this benchmark into rmg-core as a test module
+    // 3. Using Engine API indirectly (less precise)
+
+    let _ = (
+        BenchmarkId::new("placeholder", "reserve_scaling"),
+        Throughput::Elements(1),
+        make_hash(0),
+    );
+}
+
+criterion_group! {
+    name = benches;
+    config = Criterion::default()
+        .warm_up_time(Duration::from_secs(1))
+        .measurement_time(Duration::from_secs(5))
+        .sample_size(50);
+    targets = bench_reserve_scaling
+}
+criterion_main!(benches);

crates/rmg-benches/benches/scheduler_drain.rs

@@ -15,6 +15,7 @@ use rmg_core::{
     make_node_id, make_type_id, ApplyResult, ConflictPolicy, Engine, Footprint, Hash, NodeId,
     NodeRecord, PatternGraph, RewriteRule,
 };
+use std::time::Duration;
 
 // Bench constants to avoid magic strings.
 const BENCH_NOOP_RULE_NAME: &str = "bench/noop";
@@ -70,29 +71,71 @@ fn build_engine_with_entities(n: usize) -> (Engine, Vec<NodeId>) {
 
 fn bench_scheduler_drain(c: &mut Criterion) {
     let mut group = c.benchmark_group("scheduler_drain");
-    for &n in &[10usize, 100, 1_000] {
+    // Stabilize CI runs: explicit warmup/measurement and sample size.
+    group
+        .warm_up_time(Duration::from_secs(3))
+        .measurement_time(Duration::from_secs(10))
+        .sample_size(60);
+    for &n in &[10usize, 100, 1_000, 3_000, 10_000, 30_000] {
         // Throughput: number of rule applications in this run (n entities).
         group.throughput(Throughput::Elements(n as u64));
+
+        // Full apply+commit cycle (original benchmark)
         group.bench_with_input(BenchmarkId::from_parameter(n), &n, |b, &n| {
             b.iter_batched(
                 || build_engine_with_entities(n),
                 |(mut engine, ids)| {
                     // Apply the no-op rule to all entities, then commit.
                     let tx = engine.begin();
                     for id in &ids {
-                        let res = engine
-                            .apply(tx, BENCH_NOOP_RULE_NAME, id)
-                            .expect("Failed to apply noop bench rule");
+                        let res = engine.apply(tx, BENCH_NOOP_RULE_NAME, id).unwrap();
                         // Avoid affecting timing; check only in debug builds.
                         debug_assert!(matches!(res, ApplyResult::Applied));
                     }
-                    let snap = engine.commit(tx).expect("Failed to commit benchmark tx");
+                    let snap = engine.commit(tx).unwrap();
                     // Ensure the commit work is not optimized away.
                     criterion::black_box(snap);
                 },
                 BatchSize::PerIteration,
             )
         });
+
+        // Enqueue phase only (apply without commit)
+        group.bench_function(BenchmarkId::new("enqueue", n), |b| {
+            b.iter_batched(
+                || build_engine_with_entities(n),
+                |(mut engine, ids)| {
+                    let tx = engine.begin();
+                    for id in &ids {
+                        let res = engine.apply(tx, BENCH_NOOP_RULE_NAME, id).unwrap();
+                        debug_assert!(matches!(res, ApplyResult::Applied));
+                    }
+                    criterion::black_box(tx);
+                },
+                BatchSize::PerIteration,
+            )
+        });
+
+        // Drain phase only (commit with pre-enqueued rewrites)
+        group.bench_function(BenchmarkId::new("drain", n), |b| {
+            b.iter_batched(
+                || {
+                    let (mut engine, ids) = build_engine_with_entities(n);
+                    let tx = engine.begin();
+                    // Pre-enqueue all rewrites (not timed)
+                    for id in &ids {
+                        let _ = engine.apply(tx, BENCH_NOOP_RULE_NAME, id).unwrap();
+                    }
+                    (engine, tx)
+                },
+                |(mut engine, tx)| {
+                    // Only measure the commit (drain + execute)
+                    let snap = engine.commit(tx).unwrap();
+                    criterion::black_box(snap);
+                },
+                BatchSize::PerIteration,
+            )
+        });
     }
     group.finish();
 }

crates/rmg-benches/benches/snapshot_hash.rs

@@ -14,6 +14,7 @@ use criterion::{criterion_group, criterion_main, BatchSize, BenchmarkId, Criteri
 use rmg_core::{
     make_edge_id, make_node_id, make_type_id, EdgeRecord, Engine, GraphStore, NodeRecord,
 };
+use std::time::Duration;
 
 // String constants to avoid magic literals drifting silently.
 const ROOT_ID_STR: &str = "root";
@@ -71,7 +72,12 @@ fn build_chain_engine(n: usize) -> Engine {
 
 fn bench_snapshot_hash(c: &mut Criterion) {
     let mut group = c.benchmark_group("snapshot_hash");
-    for &n in &[10usize, 100, 1_000] {
+    // Stabilize CI runs across environments.
+    group
+        .warm_up_time(Duration::from_secs(3))
+        .measurement_time(Duration::from_secs(10))
+        .sample_size(80);
+    for &n in &[10usize, 100, 1_000, 3_000, 10_000, 30_000] {
         // Throughput: total nodes in reachable set (n entities + 1 root).
         group.throughput(Throughput::Elements(n as u64 + 1));
         group.bench_with_input(BenchmarkId::from_parameter(n), &n, |b, &n| {
@@ -82,7 +88,7 @@ fn bench_snapshot_hash(c: &mut Criterion) {
                     let snap = engine.snapshot();
                     criterion::black_box(snap.hash);
                 },
-                BatchSize::SmallInput,
+                BatchSize::PerIteration,
             )
         });
     }

crates/rmg-core/Cargo.toml

@@ -19,6 +19,7 @@ hex = { version = "0.4", optional = true }
 serde = { version = "1.0", features = ["derive"], optional = true }
 serde_json = { version = "1.0", optional = true }
 once_cell = "1.19"
+rustc-hash = "2.0"
 
 [dev-dependencies]
 serde = { version = "1.0", features = ["derive"] }

crates/rmg-core/src/engine_impl.rs

@@ -157,8 +157,8 @@ impl Engine {
                 "missing compact rule id for a registered rule",
             ));
         };
-        self.scheduler.pending.entry(tx).or_default().insert(
-            (scope_fp, rule.id),
+        self.scheduler.enqueue(
+            tx,
             PendingRewrite {
                 rule_id: rule.id,
                 compact_rule,

crates/rmg-core/src/footprint.rs

@@ -37,6 +37,10 @@ impl IdSet {
     pub fn insert_edge(&mut self, id: &EdgeId) {
         self.0.insert(id.0);
     }
+    /// Returns an iterator over the identifiers in the set.
+    pub fn iter(&self) -> impl Iterator<Item = &Hash> {
+        self.0.iter()
+    }
     /// Returns true if any element is shared with `other`.
     pub fn intersects(&self, other: &Self) -> bool {
         // Early‑exit by zipping ordered sets.