From cd93ae40ac12fa8a6093291399264760119f8ad7 Mon Sep 17 00:00:00 2001 From: Lewis Date: Sat, 9 May 2026 10:59:06 +0300 Subject: [PATCH] feat(bobbin-sim): pre-work, runtime, determinism Lewis: May this revision serve well! --- Cargo.lock | 26 ++++ Cargo.toml | 2 + crates/bobbin-sim/Cargo.toml | 38 +++++ crates/bobbin-sim/src/determinism.rs | 200 +++++++++++++++++++++++++ crates/bobbin-sim/src/lib.rs | 13 ++ crates/bobbin-sim/src/report.rs | 32 ++++ crates/bobbin-sim/src/runtime.rs | 168 +++++++++++++++++++++ crates/bobbin-sim/src/trace_capture.rs | 96 ++++++++++++ crates/bobbin-sim/src/workload.rs | 36 +++++ 9 files changed, 611 insertions(+) create mode 100644 crates/bobbin-sim/Cargo.toml create mode 100644 crates/bobbin-sim/src/determinism.rs create mode 100644 crates/bobbin-sim/src/lib.rs create mode 100644 crates/bobbin-sim/src/report.rs create mode 100644 crates/bobbin-sim/src/runtime.rs create mode 100644 crates/bobbin-sim/src/trace_capture.rs create mode 100644 crates/bobbin-sim/src/workload.rs diff --git a/Cargo.lock b/Cargo.lock index 75f050b..0ed5699 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -417,6 +417,32 @@ dependencies = [ "tracing", ] +[[package]] +name = "bobbin-sim" +version = "0.0.1" +dependencies = [ + "bobbin-edge-index", + "bobbin-ingest", + "bobbin-record-lru", + "bobbin-runtime", + "bobbin-search", + "bobbin-slingshot-client", + "bobbin-types", + "bytes", + "clap", + "futures", + "http", + "jacquard-common", + "serde", + "serde_json", + "thiserror 2.0.18", + "tokio", + "tokio-util", + "tracing", + "tracing-subscriber", + "url", +] + [[package]] name = "bobbin-slingshot-client" version = "0.0.1" diff --git a/Cargo.toml b/Cargo.toml index ada6f56..9308b74 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,6 +2,7 @@ resolver = "2" members = [ "crates/bobbin", + "crates/bobbin-sim", "crates/types", "crates/edge-index", "crates/ingest", @@ -28,6 +29,7 @@ bobbin-record-lru = { path = "crates/record-lru" } bobbin-knot-proxy = { path = "crates/knot-proxy" } bobbin-runtime = { path = "crates/runtime" } bobbin-search = { path = "crates/search" } +bobbin-sim = { path = "crates/bobbin-sim" } bobbin-xrpc = { path = "crates/xrpc" } jacquard-common = "0.12.0-beta.2" diff --git a/crates/bobbin-sim/Cargo.toml b/crates/bobbin-sim/Cargo.toml new file mode 100644 index 0000000..7e053f9 --- /dev/null +++ b/crates/bobbin-sim/Cargo.toml @@ -0,0 +1,38 @@ +[package] +name = "bobbin-sim" +version.workspace = true +edition.workspace = true +license.workspace = true +rust-version.workspace = true + +[lib] +path = "src/lib.rs" + +[[bin]] +name = "bobbin-sim" +path = "src/main.rs" + +[dependencies] +bobbin-edge-index = { workspace = true } +bobbin-ingest = { workspace = true } +bobbin-record-lru = { workspace = true } +bobbin-runtime = { workspace = true } +bobbin-search = { workspace = true } +bobbin-slingshot-client = { workspace = true } +bobbin-types = { workspace = true } + +bytes = { workspace = true } +clap = { workspace = true } +futures = { workspace = true } +http = { workspace = true } +jacquard-common = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +thiserror = { workspace = true } +tokio = { workspace = true, features = ["test-util"] } +tokio-util = { workspace = true } +tracing = { workspace = true } +tracing-subscriber = { workspace = true } +url = { workspace = true } + +[dev-dependencies] diff --git a/crates/bobbin-sim/src/determinism.rs b/crates/bobbin-sim/src/determinism.rs new file mode 100644 index 0000000..2dfec0f --- /dev/null +++ b/crates/bobbin-sim/src/determinism.rs @@ -0,0 +1,200 @@ +use std::num::NonZeroUsize; +use std::time::Duration; + +use tokio::runtime::Builder as TokioBuilder; +use tracing_subscriber::Registry; +use tracing_subscriber::layer::SubscriberExt; + +use crate::report::{SimOutcome, SimReport}; +use crate::runtime::{Sim, SimConfig}; +use crate::trace_capture::TraceCapture; +use crate::workload::Workload; + +#[derive(Clone, Debug)] +pub struct LeakRunConfig { + pub seed: u64, + pub parallelism: NonZeroUsize, + pub max_virtual_runtime: Duration, + pub mem_ws_capacity: usize, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum LeakOutcome { + Match, + OutcomeMismatch { + first: SimOutcome, + second: SimOutcome, + }, + EventCountMismatch { + first: u64, + second: u64, + }, + EdgeCountMismatch { + first: u64, + second: u64, + }, + LastCursorMismatch { + first: u64, + second: u64, + }, + ResolverHitsMismatch { + first: u64, + second: u64, + }, + ResolverMissesMismatch { + first: u64, + second: u64, + }, + ConsumerTooSlowMismatch { + first: u64, + second: u64, + }, + TraceLengthMismatch { + first: usize, + second: usize, + }, + TraceLineMismatch { + index: usize, + first: String, + second: String, + }, +} + +#[derive(Debug)] +pub struct LeakRunResult { + pub config: LeakRunConfig, + pub workload: &'static str, + pub outcome: LeakOutcome, + pub first_report: SimReport, + pub second_report: SimReport, +} + +impl LeakRunResult { + pub fn passed(&self) -> bool { + matches!(self.outcome, LeakOutcome::Match) + } +} + +pub fn run_leak_check( + config: LeakRunConfig, + workload_factory: F, +) -> LeakRunResult +where + F: Fn() -> Box, +{ + let workload_name = { + let probe = workload_factory(); + probe.name() + }; + + let (first_report, first_lines) = run_once(&config, workload_factory()); + let (second_report, second_lines) = run_once(&config, workload_factory()); + + let outcome = compare_runs( + &first_report, + &first_lines, + &second_report, + &second_lines, + ); + + LeakRunResult { + config, + workload: workload_name, + outcome, + first_report, + second_report, + } +} + +fn run_once( + config: &LeakRunConfig, + workload: Box, +) -> (SimReport, Vec) { + let capture = TraceCapture::new(); + let subscriber = Registry::default().with(capture.layer()); + + let mut sim_config = SimConfig::new(config.seed); + sim_config.max_virtual_runtime = config.max_virtual_runtime; + sim_config.parallelism = config.parallelism; + sim_config.mem_ws_capacity = config.mem_ws_capacity; + + let runtime = TokioBuilder::new_current_thread() + .enable_all() + .start_paused(true) + .build() + .expect("build current_thread runtime with paused time"); + + let report = tracing::subscriber::with_default(subscriber, || { + runtime.block_on(Sim::new(sim_config, workload).run()) + }); + drop(runtime); + + let lines = capture.into_lines(); + (report, lines) +} + +fn compare_runs( + a: &SimReport, + a_lines: &[String], + b: &SimReport, + b_lines: &[String], +) -> LeakOutcome { + if a.outcome != b.outcome { + return LeakOutcome::OutcomeMismatch { + first: a.outcome, + second: b.outcome, + }; + } + if a.events_processed != b.events_processed { + return LeakOutcome::EventCountMismatch { + first: a.events_processed, + second: b.events_processed, + }; + } + if a.edge_count != b.edge_count { + return LeakOutcome::EdgeCountMismatch { + first: a.edge_count, + second: b.edge_count, + }; + } + if a.last_cursor != b.last_cursor { + return LeakOutcome::LastCursorMismatch { + first: a.last_cursor, + second: b.last_cursor, + }; + } + if a.resolver_hits != b.resolver_hits { + return LeakOutcome::ResolverHitsMismatch { + first: a.resolver_hits, + second: b.resolver_hits, + }; + } + if a.resolver_misses != b.resolver_misses { + return LeakOutcome::ResolverMissesMismatch { + first: a.resolver_misses, + second: b.resolver_misses, + }; + } + if a.consumer_too_slow_count != b.consumer_too_slow_count { + return LeakOutcome::ConsumerTooSlowMismatch { + first: a.consumer_too_slow_count, + second: b.consumer_too_slow_count, + }; + } + if a_lines.len() != b_lines.len() { + return LeakOutcome::TraceLengthMismatch { + first: a_lines.len(), + second: b_lines.len(), + }; + } + for (i, (x, y)) in a_lines.iter().zip(b_lines.iter()).enumerate() { + if x != y { + return LeakOutcome::TraceLineMismatch { + index: i, + first: x.clone(), + second: y.clone(), + }; + } + } + LeakOutcome::Match +} diff --git a/crates/bobbin-sim/src/lib.rs b/crates/bobbin-sim/src/lib.rs new file mode 100644 index 0000000..260443c --- /dev/null +++ b/crates/bobbin-sim/src/lib.rs @@ -0,0 +1,13 @@ +mod determinism; +mod report; +mod runtime; +mod trace_capture; +mod workload; + +pub mod workloads; + +pub use determinism::{LeakOutcome, LeakRunConfig, LeakRunResult, run_leak_check}; +pub use report::{SimOutcome, SimReport}; +pub use runtime::{Sim, SimConfig}; +pub use trace_capture::{StageLayer, TraceCapture}; +pub use workload::{Workload, WorkloadCtx, WorkloadHooks}; diff --git a/crates/bobbin-sim/src/report.rs b/crates/bobbin-sim/src/report.rs new file mode 100644 index 0000000..21053c0 --- /dev/null +++ b/crates/bobbin-sim/src/report.rs @@ -0,0 +1,32 @@ +use std::time::Duration; + +use bobbin_runtime::UnixMicros; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum SimOutcome { + Passed, + Failed, + TimedOut, +} + +#[derive(Clone, Debug)] +pub struct SimReport { + pub workload: &'static str, + pub seed: u64, + pub outcome: SimOutcome, + pub virtual_runtime: Duration, + pub virtual_clock_end: UnixMicros, + pub events_processed: u64, + pub last_cursor: u64, + pub edge_count: u64, + pub resolver_hits: u64, + pub resolver_misses: u64, + pub consumer_too_slow_count: u64, + pub failure_reason: Option, +} + +impl SimReport { + pub fn passed(self) -> bool { + matches!(self.outcome, SimOutcome::Passed) + } +} diff --git a/crates/bobbin-sim/src/runtime.rs b/crates/bobbin-sim/src/runtime.rs new file mode 100644 index 0000000..15b422e --- /dev/null +++ b/crates/bobbin-sim/src/runtime.rs @@ -0,0 +1,168 @@ +use std::num::NonZeroUsize; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; + +use bobbin_edge_index::{CoverageWatch, EdgeStore}; +use bobbin_ingest::{ + DEFAULT_INGEST_PARALLELISM, IngestConfig, IngestRuntime, RepoIdResolver, run as run_ingest, +}; +use bobbin_record_lru::NoopRecordStore; +use bobbin_runtime::{ + Clock, DEFAULT_MEM_WS_CAPACITY, MemHttpTransport, MemWsTransport, RuntimeHasher, + SeededEntropy, SimClock, UnixMicros, +}; +use bobbin_slingshot_client::SlingshotClient; +use bobbin_types::search::NoopSearchSink; +use tokio_util::sync::CancellationToken; +use url::Url; + +use crate::report::{SimOutcome, SimReport}; +use crate::workload::{Workload, WorkloadCtx}; + +#[derive(Clone, Debug)] +pub struct SimConfig { + pub seed: u64, + pub base_unix: UnixMicros, + pub max_virtual_runtime: Duration, + pub parallelism: NonZeroUsize, + pub hydrant_base: Url, + pub slingshot_base: Url, + pub mem_ws_capacity: usize, +} + +impl SimConfig { + pub fn new(seed: u64) -> Self { + Self { + seed, + base_unix: UnixMicros::new(1_700_000_000_000_000), + max_virtual_runtime: Duration::from_secs(60), + parallelism: DEFAULT_INGEST_PARALLELISM, + hydrant_base: Url::parse("ws://hydrant.sim/").unwrap(), + slingshot_base: Url::parse("http://slingshot.sim/").unwrap(), + mem_ws_capacity: DEFAULT_MEM_WS_CAPACITY, + } + } +} + +pub struct Sim { + config: SimConfig, + workload: Box, +} + +impl Sim { + pub fn new(config: SimConfig, workload: Box) -> Self { + Self { config, workload } + } + + pub async fn run(self) -> SimReport { + let SimConfig { + seed, + base_unix, + max_virtual_runtime, + parallelism, + hydrant_base, + slingshot_base, + mem_ws_capacity, + } = self.config; + + let entropy = Arc::new(SeededEntropy::new(seed)); + let hasher = RuntimeHasher::from_entropy(&*entropy); + let clock: Arc = Arc::new(SimClock::at(base_unix)); + + let store = Arc::new(EdgeStore::new(hasher.clone())); + let coverage = Arc::new(CoverageWatch::new()); + let records = Arc::new(NoopRecordStore); + let cancel = CancellationToken::new(); + let consumer_too_slow_count = Arc::new(AtomicU64::new(0)); + + let workload_name = self.workload.name(); + let ctx = WorkloadCtx { + seed, + clock: clock.clone(), + entropy: entropy.clone(), + store: store.clone(), + coverage: coverage.clone(), + records: records.clone(), + cancel: cancel.clone(), + consumer_too_slow_count: consumer_too_slow_count.clone(), + }; + let hooks = self.workload.build(ctx); + + let slingshot_http = + MemHttpTransport::shared(hooks.slingshot.clone(), clock.clone()); + let slingshot_client = SlingshotClient::new(slingshot_base, slingshot_http) + .expect("slingshot base url is valid"); + let resolver = Arc::new(RepoIdResolver::with_slingshot( + slingshot_client, + clock.clone(), + hasher.clone(), + )); + + let mem_ws = MemWsTransport::shared_with_capacity(hooks.hydrant.clone(), mem_ws_capacity); + + let ingest_runtime: IngestRuntime = IngestRuntime { + store: store.clone(), + coverage: coverage.clone(), + search: Arc::new(NoopSearchSink), + records: records.clone() as Arc, + resolver: resolver.clone(), + clock: clock.clone(), + entropy: entropy.clone(), + ws: mem_ws, + cancel: cancel.clone(), + }; + let ingest_config = IngestConfig { + hydrant_base, + start_cursor: bobbin_edge_index::HydrantCursor::new(0), + parallelism, + }; + let mut ingest_handle = tokio::spawn(async move { + let _ = run_ingest(ingest_config, ingest_runtime).await; + }); + + let script = hooks.script; + let report = tokio::select! { + biased; + _ = clock.sleep(max_virtual_runtime) => SimReport { + workload: workload_name, + seed, + outcome: SimOutcome::TimedOut, + virtual_runtime: max_virtual_runtime, + virtual_clock_end: clock.now_unix_micros(), + events_processed: coverage.snapshot().events_processed(), + last_cursor: coverage.snapshot().last_cursor().raw(), + edge_count: store.key_count() as u64, + resolver_hits: resolver.stats().hits, + resolver_misses: resolver.stats().miss_count(), + consumer_too_slow_count: consumer_too_slow_count.load(Ordering::Relaxed), + failure_reason: Some(format!( + "max_virtual_runtime {max_virtual_runtime:?} exhausted", + )), + }, + r = script => { + let stats = resolver.stats(); + SimReport { + resolver_hits: stats.hits, + resolver_misses: stats.miss_count(), + consumer_too_slow_count: consumer_too_slow_count.load(Ordering::Relaxed), + ..r + } + } + }; + + cancel.cancel(); + let drain_deadline = clock.sleep(Duration::from_secs(5)); + tokio::pin!(drain_deadline); + tokio::select! { + _ = &mut drain_deadline => { + ingest_handle.abort(); + let _ = ingest_handle.await; + } + res = &mut ingest_handle => { + let _ = res; + } + } + report + } +} diff --git a/crates/bobbin-sim/src/trace_capture.rs b/crates/bobbin-sim/src/trace_capture.rs new file mode 100644 index 0000000..0184268 --- /dev/null +++ b/crates/bobbin-sim/src/trace_capture.rs @@ -0,0 +1,96 @@ +use std::sync::Arc; +use std::sync::Mutex; + +use tracing::field::{Field, Visit}; +use tracing::{Event, Subscriber}; +use tracing_subscriber::Layer; +use tracing_subscriber::layer::Context; +use tracing_subscriber::registry::LookupSpan; + +#[derive(Clone, Default)] +pub struct TraceCapture { + inner: Arc>>, +} + +impl TraceCapture { + pub fn new() -> Self { + Self::default() + } + + pub fn lines(&self) -> Vec { + self.inner.lock().unwrap().clone() + } + + pub fn into_lines(self) -> Vec { + std::mem::take(&mut *self.inner.lock().unwrap()) + } + + pub fn layer(&self) -> StageLayer { + StageLayer { + sink: self.inner.clone(), + } + } +} + +pub struct StageLayer { + sink: Arc>>, +} + +impl Layer for StageLayer +where + S: Subscriber + for<'a> LookupSpan<'a>, +{ + fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) { + let meta = event.metadata(); + if meta.target() != "bobbin_ingest::stage" { + return; + } + let mut visitor = StageVisitor::default(); + event.record(&mut visitor); + let line = format!( + "cursor={} regime={} nsid={} edge_count={} parallelism={}", + visitor.cursor.unwrap_or(u64::MAX), + visitor.regime.as_deref().unwrap_or(""), + visitor.nsid.as_deref().unwrap_or(""), + visitor.edge_count.unwrap_or(0), + visitor.parallelism.unwrap_or(0), + ); + self.sink.lock().unwrap().push(line); + } +} + +#[derive(Default)] +struct StageVisitor { + cursor: Option, + regime: Option, + nsid: Option, + edge_count: Option, + parallelism: Option, +} + +impl Visit for StageVisitor { + fn record_u64(&mut self, field: &Field, value: u64) { + match field.name() { + "cursor" => self.cursor = Some(value), + "edge_count" => self.edge_count = Some(value), + "parallelism" => self.parallelism = Some(value as usize), + _ => {} + } + } + + fn record_i64(&mut self, field: &Field, value: i64) { + if field.name() == "cursor" { + self.cursor = Some(value as u64); + } + } + + fn record_str(&mut self, field: &Field, value: &str) { + match field.name() { + "regime" => self.regime = Some(value.to_owned()), + "nsid" => self.nsid = Some(value.to_owned()), + _ => {} + } + } + + fn record_debug(&mut self, _field: &Field, _value: &dyn std::fmt::Debug) {} +} diff --git a/crates/bobbin-sim/src/workload.rs b/crates/bobbin-sim/src/workload.rs new file mode 100644 index 0000000..64c3ef3 --- /dev/null +++ b/crates/bobbin-sim/src/workload.rs @@ -0,0 +1,36 @@ +use std::future::Future; +use std::pin::Pin; +use std::sync::Arc; +use std::sync::atomic::AtomicU64; + +use bobbin_edge_index::{CoverageWatch, EdgeStore}; +use bobbin_record_lru::NoopRecordStore; +use bobbin_runtime::{Clock, Entropy, MemHttpResponder, MemWsResponder}; +use tokio_util::sync::CancellationToken; + +use crate::report::SimReport; + +pub struct WorkloadCtx { + pub seed: u64, + pub clock: Arc, + pub entropy: Arc, + pub store: Arc, + pub coverage: Arc, + pub records: Arc, + pub cancel: CancellationToken, + pub consumer_too_slow_count: Arc, +} + +pub type WorkloadScript = + Pin + Send + 'static>>; + +pub struct WorkloadHooks { + pub slingshot: Arc, + pub hydrant: Arc, + pub script: WorkloadScript, +} + +pub trait Workload: Send + 'static { + fn name(&self) -> &'static str; + fn build(self: Box, ctx: WorkloadCtx) -> WorkloadHooks; +} -- 2.51.2