racer: refactoring/DSTs for core Racer logic

Makefile

@@ -367,6 +367,7 @@ help: ## Show this help
 	@echo "Racer:"
 	@echo "  racer-test                       Test the Rust Racer workspace and Go Racer API and wire contracts"
 	@echo "  racer-gpu-test                   Build the CUDA shim with kernels and run GPU hardware tests"
+	@echo "  racer-sim                        Run the Racer simulator (ARGS='--scenario smoke --nodes 16')"
 	@echo "  racer-envtest                    Run Racer API-server tests with KUBEBUILDER_ASSETS"
 	@echo "  racer-envtest-ci                 Provision pinned assets and run Racer API-server tests"
 	@echo "  racer-generate                   Generate Racer deepcopy and CRD artifacts"
@@ -610,6 +611,12 @@ racer-python-dist: ## Build sdists and wheels for racer-client and racer-vllm in
 	timeout --signal=TERM --kill-after=10s 300s "$(RACER_PY_VENV)/bin/python" -m build -o tmp/racer-python-dist clients/python/racer
 	timeout --signal=TERM --kill-after=10s 300s "$(RACER_PY_VENV)/bin/python" -m build -o tmp/racer-python-dist clients/python/racer-vllm
 
+RACER_SIM_TIMEOUT ?= 7200s
+
+.PHONY: racer-sim
+racer-sim: ## Run the Racer discrete-event simulator; pass options with ARGS='--nodes 50000 --seed 1'
+	timeout --signal=TERM --kill-after=10s $(RACER_SIM_TIMEOUT) cargo run --release --locked --manifest-path cmd/racer-dataplane/Cargo.toml -p racer-sim -- $(ARGS)
+
 $(SETUP_ENVTEST):
 	@mkdir -p bin tmp/envtest-tools
 	TMPDIR="$(CURDIR)/tmp/envtest-tools" GOBIN="$(CURDIR)/bin" timeout --signal=TERM --kill-after=10s 300s $(GOCMD) install sigs.k8s.io/controller-runtime/tools/setup-envtest@$(SETUP_ENVTEST_VERSION)

cmd/racer-dataplane/Cargo.toml

@@ -1,5 +1,5 @@
 [workspace]
-members = [".", "runtime", "alloc", "crypto", "verbs", "gpu", "http", "topology", "telemetry", "flow", "control-wire", "controlplane", "uds-endpoint", "wire-codec", "peer-wire", "object-wire", "cluster-control", "admission"]
+members = [".", "runtime", "alloc", "crypto", "verbs", "gpu", "http", "topology", "telemetry", "flow", "control-wire", "controlplane", "uds-endpoint", "wire-codec", "peer-wire", "object-wire", "cluster-control", "admission", "core", "sim"]
 resolver = "3"
 
 [package]
@@ -28,6 +28,7 @@ required-features = ["subscription-interop"]
 controlplane = { path = "controlplane" }
 peer-wire = { path = "peer-wire" }
 racer-object-wire = { path = "object-wire" }
+racer-core = { path = "core" }
 envy = { version = "0.4.2", default-features = false }
 uds-endpoint = { path = "uds-endpoint" }
 wire-codec = { path = "wire-codec", features = ["rest"] }

cmd/racer-dataplane/alloc/src/lib.rs

@@ -168,6 +168,8 @@ pub use demand::{
     DEFAULT_DECAY_WINDOWS, DEFAULT_INITIAL_TARGET, DEFAULT_MAX_DEPTH, DEFAULT_MIN_DEPTH,
     DEFAULT_WINDOW, PoolDemand, PoolStats,
 };
+#[cfg(feature = "simulation")]
+pub use segments::PhantomEntries;
 pub use segments::{
     FreezeGuard, Generation, SegmentClock, SegmentEntries, SegmentId, SegmentLease,
     SegmentSnapshot, SegmentState, Segments,

cmd/racer-dataplane/alloc/src/retention.rs

@@ -7,8 +7,11 @@ use std::time::{Duration, Instant};
 const GENERATIONS: usize = 4;
 
 /// Bounded rotating Bloom history. A hit is a hint, never identity or ownership.
-pub struct SecondSight {
-    hash: RandomState,
+///
+/// Production uses the randomly keyed [`RandomState`]; simulators pass a seeded
+/// [`BuildHasher`] through [`SecondSight::with_hasher`] so false positives replay.
+pub struct SecondSight<S = RandomState> {
+    hash: S,
 
     pages: [Vec<u64>; GENERATIONS],
 
@@ -24,12 +27,19 @@ pub struct SecondSight {
 impl SecondSight {
     /// Construct four generations within the byte budget, starting at `now`.
     pub fn new(bytes: usize, period: Duration, now: Instant) -> Option<Self> {
+        Self::with_hasher(bytes, period, now, RandomState::new())
+    }
+}
+
+impl<S: BuildHasher> SecondSight<S> {
+    /// Like [`SecondSight::new`] with an explicit, possibly deterministic, hasher.
+    pub fn with_hasher(bytes: usize, period: Duration, now: Instant, hash: S) -> Option<Self> {
         let words = bytes / (GENERATIONS * 8);
         if words == 0 || period.is_zero() || period.as_nanos() > u64::MAX as u128 {
             return None;
         }
         Some(Self {
-            hash: RandomState::new(),
+            hash,
             pages: std::array::from_fn(|_| vec![0; words]),
             occupancy: [0; GENERATIONS],
             current: 0,
@@ -87,8 +97,11 @@ impl SecondSight {
 }
 
 /// Exact, capacity-bounded resident heat. Only explicitly tracked keys can heat up.
-pub struct Heat<K> {
-    entries: HashMap<K, Temperature>,
+///
+/// The table is never iterated, so the hasher only affects performance; it is a
+/// parameter so simulators can avoid OS randomness entirely.
+pub struct Heat<K, S = RandomState> {
+    entries: HashMap<K, Temperature, S>,
 
     capacity: usize,
 }
@@ -112,8 +125,15 @@ impl Temperature {
 impl<K: Eq + Hash + Clone> Heat<K> {
     /// Create an empty exact table; zero capacity is invalid.
     pub fn new(capacity: usize) -> Option<Self> {
+        Self::with_hasher(capacity, RandomState::new())
+    }
+}
+
+impl<K: Eq + Hash + Clone, S: BuildHasher> Heat<K, S> {
+    /// Like [`Heat::new`] with an explicit hasher.
+    pub fn with_hasher(capacity: usize, hash: S) -> Option<Self> {
         (capacity != 0).then(|| Self {
-            entries: HashMap::new(),
+            entries: HashMap::with_hasher(hash),
             capacity,
         })
     }
@@ -189,6 +209,32 @@ mod tests {
         assert!(SecondSight::new(32, Duration::ZERO, start).is_none());
     }
 
+    /// Fixed-key hashers make Bloom collisions identical across instances.
+    #[test]
+    fn seeded_hasher_replays_identically() {
+        use std::hash::BuildHasherDefault;
+        type Fixed = BuildHasherDefault<std::collections::hash_map::DefaultHasher>;
+        let start = Instant::now();
+        let run = || {
+            let mut history =
+                SecondSight::with_hasher(64, Duration::from_secs(60), start, Fixed::default())
+                    .unwrap();
+            (0..200u64)
+                .map(|key| history.observe(&key, start))
+                .collect::<Vec<_>>()
+        };
+        let first = run();
+        assert_eq!(first, run());
+        assert!(first.iter().any(|seen| *seen), "tiny filter must collide");
+        assert!(
+            SecondSight::with_hasher(31, Duration::from_secs(1), start, Fixed::default()).is_none()
+        );
+        let mut heat = Heat::with_hasher(1, Fixed::default()).unwrap();
+        assert!(heat.track(&1u64, start));
+        assert!(!heat.track(&2u64, start));
+        assert!(Heat::<u64, Fixed>::with_hasher(0, Fixed::default()).is_none());
+    }
+
     #[test]
     fn exact_heat_is_bounded_saturates_decays_and_forgets() {
         let now = Instant::now();