From 78e1f25b1ddd345ccf2309d5795cb102802182b1 Mon Sep 17 00:00:00 2001 From: Shubham Mishra Date: Fri, 31 Jul 2026 13:38:57 +0530 Subject: [PATCH 1/4] runtime enter --- crates/runtime/src/lib.rs | 65 ++++++++++++++++++++++++++ crates/runtime/src/sim/executor/mod.rs | 1 + crates/runtime/src/sim_std.rs | 41 +++++++++------- 3 files changed, 89 insertions(+), 18 deletions(-) diff --git a/crates/runtime/src/lib.rs b/crates/runtime/src/lib.rs index 9511cf52c50..b7b589e51c3 100644 --- a/crates/runtime/src/lib.rs +++ b/crates/runtime/src/lib.rs @@ -64,6 +64,11 @@ pub struct AbortHandle { inner: AbortHandleInner, } +#[must_use = "runtime context exits immediately unless the guard is held"] +pub struct EnterGuard<'a> { + _inner: EnterGuardInner<'a>, +} + enum JoinHandleInner { Tokio(tokio::task::JoinHandle), #[cfg(feature = "simulation")] @@ -89,6 +94,16 @@ enum AbortHandleInner { Simulation(sim::AbortHandle), } +enum EnterGuardInner<'a> { + Tokio { + _guard: tokio::runtime::EnterGuard<'a>, + }, + #[cfg(feature = "simulation")] + Simulation { + _guard: sim_std::EnterGuard, + }, +} + #[derive(Debug)] pub struct JoinError { inner: JoinErrorInner, @@ -219,6 +234,18 @@ impl fmt::Display for RuntimeTimeout { impl std::error::Error for RuntimeTimeout {} +pub fn spawn(future: impl Future + Send + 'static) -> JoinHandle { + Handle::current().spawn(future) +} + +pub async fn spawn_blocking(f: F) -> R +where + F: FnOnce() -> R + Send + 'static, + R: Send + 'static, +{ + Handle::current().spawn_blocking(f).await +} + impl Handle { pub fn tokio(handle: TokioHandle) -> Self { Self::Tokio(handle) @@ -227,6 +254,15 @@ impl Handle { pub fn tokio_current() -> Self { Self::tokio(TokioHandle::current()) } + + pub fn current() -> Self { + #[cfg(feature = "simulation")] + if let Some(handle) = sim_std::try_current() { + return Self::Simulation(handle); + } + + Self::tokio_current() + } } #[cfg(feature = "simulation")] @@ -237,6 +273,20 @@ impl Handle { } impl Handle { + pub fn enter(&self) -> EnterGuard<'_> { + match self { + Self::Tokio(handle) => EnterGuard { + _inner: EnterGuardInner::Tokio { _guard: handle.enter() }, + }, + #[cfg(feature = "simulation")] + Self::Simulation(handle) => EnterGuard { + _inner: EnterGuardInner::Simulation { + _guard: sim_std::enter(handle.clone()), + }, + }, + } + } + pub fn spawn(&self, future: impl Future + Send + 'static) -> JoinHandle { match self { Self::Tokio(handle) => JoinHandle { @@ -358,4 +408,19 @@ mod tests { assert!(!flag.load(Ordering::Acquire)); }); } + + #[cfg(feature = "simulation")] + #[test] + fn free_spawn_apis_use_current_simulation_runtime() { + use crate::sim::Runtime; + let mut rt = Runtime::new(4); + + let output = rt.block_on(async { + let spawned = crate::spawn(async { 31 }).await.unwrap(); + let blocking = crate::spawn_blocking(|| 41).await; + spawned + blocking + }); + + assert_eq!(output, 72); + } } diff --git a/crates/runtime/src/sim/executor/mod.rs b/crates/runtime/src/sim/executor/mod.rs index 4a1a8394701..c6518bbb832 100644 --- a/crates/runtime/src/sim/executor/mod.rs +++ b/crates/runtime/src/sim/executor/mod.rs @@ -150,6 +150,7 @@ impl Runtime { /// While the future runs, spawned tasks share the same deterministic /// scheduler, timer wheel, and runtime RNG. pub fn block_on(&mut self, future: F) -> F::Output { + let _guard = crate::sim_std::enter(self.handle()); self.executor.block_on(future) } diff --git a/crates/runtime/src/sim_std.rs b/crates/runtime/src/sim_std.rs index 59293f1635b..39d200eab43 100644 --- a/crates/runtime/src/sim_std.rs +++ b/crates/runtime/src/sim_std.rs @@ -8,7 +8,7 @@ #![allow(clippy::disallowed_macros)] use alloc::boxed::Box; -use core::{cell::Cell, future::Future}; +use core::{cell::RefCell, future::Future}; use std::sync::OnceLock; use crate::sim; @@ -21,7 +21,6 @@ use crate::sim; /// tests that execute inside a hosted process. While the future runs, this /// marks the thread as inside simulation so OS thread spawns can be rejected. pub fn block_on(runtime: &mut sim::Runtime, future: F) -> F::Output { - let _system_thread_context = enter_simulation_thread(); runtime.block_on(future) } @@ -68,34 +67,40 @@ fn panic_with_seed(seed: u64, payload: Box) -> ! { // Simulation thread context. -// Ambient state used only while `sim_std::block_on` is driving a simulation. +// Ambient state used only while hosted code has entered a simulation runtime. // -// The simulator itself stays explicit-handle based. This thread-local only -// marks whether the current OS thread is owned by a running simulation so -// host thread creation can be rejected. +// The simulator itself stays explicit-handle based. This thread-local stores +// the current handle for runtime-generic code that uses [`crate::Handle::current`], +// and its presence marks the current OS thread as simulation-owned so host +// thread creation can be rejected. thread_local! { - // Marks the current OS thread as simulation-owned so thread creation hooks - // can reject accidental escapes to the host scheduler. - static IN_SIMULATION: Cell = const { Cell::new(false) }; + static CURRENT_RUNTIME: RefCell> = const { RefCell::new(None) }; } -struct SimulationThreadGuard { - previous: bool, +#[must_use = "simulation runtime context exits immediately unless the guard is held"] +pub struct EnterGuard { + previous_runtime: Option, } -fn enter_simulation_thread() -> SimulationThreadGuard { - let previous = IN_SIMULATION.with(|state| state.replace(true)); - SimulationThreadGuard { previous } +/// Enter a simulation runtime on the current OS thread until the returned guard drops. +pub fn enter(handle: sim::Handle) -> EnterGuard { + let previous_runtime = CURRENT_RUNTIME.with(|current| current.replace(Some(handle))); + EnterGuard { previous_runtime } +} + +/// Return the simulation runtime entered on this OS thread, if any. +pub(crate) fn try_current() -> Option { + CURRENT_RUNTIME.with(|current| current.borrow().clone()) } fn in_simulation() -> bool { - IN_SIMULATION.with(Cell::get) + CURRENT_RUNTIME.with(|current| current.borrow().is_some()) } -impl Drop for SimulationThreadGuard { +impl Drop for EnterGuard { fn drop(&mut self) { - IN_SIMULATION.with(|state| { - state.set(self.previous); + CURRENT_RUNTIME.with(|current| { + current.replace(self.previous_runtime.take()); }); } } From d05c8faf25389c51d8ded14eb94d4f6b3054ccd1 Mon Sep 17 00:00:00 2001 From: Shubham Mishra Date: Fri, 31 Jul 2026 16:44:36 +0530 Subject: [PATCH 2/4] split runtime-core crate --- Cargo.lock | 11 +- Cargo.toml | 2 + crates/runtime-core/Cargo.toml | 14 ++ crates/runtime-core/LICENSE | 1 + crates/runtime-core/src/lib.rs | 7 + .../src/sim/buggify.rs | 2 +- .../src/sim/executor/mod.rs | 56 ++------ .../src/sim/executor/task.rs | 6 +- .../{runtime => runtime-core}/src/sim/mod.rs | 3 +- .../{runtime => runtime-core}/src/sim/rng.rs | 16 ++- .../src/sim/time/mod.rs | 3 +- .../src/sim/time/sleep.rs | 0 crates/runtime/Cargo.toml | 5 +- crates/runtime/src/sim.rs | 129 ++++++++++++++++++ crates/runtime/tests/sim_e2e.rs | 26 ++-- 15 files changed, 210 insertions(+), 71 deletions(-) create mode 100644 crates/runtime-core/Cargo.toml create mode 120000 crates/runtime-core/LICENSE create mode 100644 crates/runtime-core/src/lib.rs rename crates/{runtime => runtime-core}/src/sim/buggify.rs (98%) rename crates/{runtime => runtime-core}/src/sim/executor/mod.rs (92%) rename crates/{runtime => runtime-core}/src/sim/executor/task.rs (96%) rename crates/{runtime => runtime-core}/src/sim/mod.rs (83%) rename crates/{runtime => runtime-core}/src/sim/rng.rs (93%) rename crates/{runtime => runtime-core}/src/sim/time/mod.rs (99%) rename crates/{runtime => runtime-core}/src/sim/time/sleep.rs (100%) create mode 100644 crates/runtime/src/sim.rs diff --git a/Cargo.lock b/Cargo.lock index 41920bbdebd..ef0f375ecec 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8553,13 +8553,20 @@ dependencies = [ name = "spacetimedb-runtime" version = "2.7.1" dependencies = [ - "async-task", "futures", "libc", - "spin", + "spacetimedb-runtime-core", "tokio", ] +[[package]] +name = "spacetimedb-runtime-core" +version = "2.7.1" +dependencies = [ + "async-task", + "spin", +] + [[package]] name = "spacetimedb-sats" version = "2.7.1" diff --git a/Cargo.toml b/Cargo.toml index 1af80d1498b..9c5d473fd7f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -28,6 +28,7 @@ members = [ "crates/primitives", "crates/query", "crates/runtime", + "crates/runtime-sim", "crates/sats", "crates/schema", "crates/smoketests", @@ -152,6 +153,7 @@ spacetimedb-primitives = { path = "crates/primitives", version = "=2.7.1" } spacetimedb-query = { path = "crates/query", version = "=2.7.1" } spacetimedb-query-builder = { path = "crates/query-builder", version = "=2.7.1" } spacetimedb-runtime = { path = "crates/runtime", version = "=2.7.1" } +spacetimedb-runtime-sim = { path = "crates/runtime-sim", version = "=2.7.1" } spacetimedb-sats = { path = "crates/sats", version = "=2.7.1" } spacetimedb-schema = { path = "crates/schema", version = "=2.7.1" } spacetimedb-snapshot = { path = "crates/snapshot", version = "=2.7.1" } diff --git a/crates/runtime-core/Cargo.toml b/crates/runtime-core/Cargo.toml new file mode 100644 index 00000000000..6d82b8f03aa --- /dev/null +++ b/crates/runtime-core/Cargo.toml @@ -0,0 +1,14 @@ +[package] +name = "spacetimedb-runtime-core" +version.workspace = true +edition.workspace = true +license-file = "LICENSE" +description = "Portable no_std runtime core utilities for SpacetimeDB" +rust-version.workspace = true + +[lints] +workspace = true + +[dependencies] +async-task = { version = "4.4", default-features = false } +spin = { version = "0.9", default-features = false, features = ["mutex", "spin_mutex"] } diff --git a/crates/runtime-core/LICENSE b/crates/runtime-core/LICENSE new file mode 120000 index 00000000000..8540cf8a991 --- /dev/null +++ b/crates/runtime-core/LICENSE @@ -0,0 +1 @@ +../../licenses/BSL.txt \ No newline at end of file diff --git a/crates/runtime-core/src/lib.rs b/crates/runtime-core/src/lib.rs new file mode 100644 index 00000000000..8296506b759 --- /dev/null +++ b/crates/runtime-core/src/lib.rs @@ -0,0 +1,7 @@ +#![no_std] + +extern crate alloc; +#[cfg(test)] +extern crate std; + +pub mod sim; diff --git a/crates/runtime/src/sim/buggify.rs b/crates/runtime-core/src/sim/buggify.rs similarity index 98% rename from crates/runtime/src/sim/buggify.rs rename to crates/runtime-core/src/sim/buggify.rs index 07188c6c207..ca4558162a6 100644 --- a/crates/runtime/src/sim/buggify.rs +++ b/crates/runtime-core/src/sim/buggify.rs @@ -1,4 +1,4 @@ -use crate::sim::Runtime; +use super::Runtime; /// Probabilistic fault-injection helpers for simulation code. /// diff --git a/crates/runtime/src/sim/executor/mod.rs b/crates/runtime-core/src/sim/executor/mod.rs similarity index 92% rename from crates/runtime/src/sim/executor/mod.rs rename to crates/runtime-core/src/sim/executor/mod.rs index c6518bbb832..fbb7f7c0cf2 100644 --- a/crates/runtime/src/sim/executor/mod.rs +++ b/crates/runtime-core/src/sim/executor/mod.rs @@ -10,7 +10,7 @@ use core::{ use spin::Mutex; -use crate::sim::{time::TimeHandle, Rng}; +use super::{time::TimeHandle, Rng}; mod task; use task::Abortable; @@ -150,7 +150,6 @@ impl Runtime { /// While the future runs, spawned tasks share the same deterministic /// scheduler, timer wheel, and runtime RNG. pub fn block_on(&mut self, future: F) -> F::Output { - let _guard = crate::sim_std::enter(self.handle()); self.executor.block_on(future) } @@ -221,22 +220,26 @@ impl Runtime { } #[allow(dead_code)] - pub(crate) fn enable_determinism_log(&self) { + #[doc(hidden)] + pub fn enable_determinism_log(&self) { self.executor.rng.enable_determinism_log(); } #[allow(dead_code)] - pub(crate) fn enable_determinism_check(&self, log: crate::sim::DeterminismLog) { + #[doc(hidden)] + pub fn enable_determinism_check(&self, log: super::DeterminismLog) { self.executor.rng.enable_determinism_check(log); } #[allow(dead_code)] - pub(crate) fn take_determinism_log(&self) -> Option { + #[doc(hidden)] + pub fn take_determinism_log(&self) -> Option { self.executor.rng.take_determinism_log() } #[allow(dead_code)] - pub(crate) fn finish_determinism_check(&self) -> Result<(), alloc::string::String> { + #[doc(hidden)] + pub fn finish_determinism_check(&self) -> Result<(), alloc::string::String> { self.executor.rng.finish_determinism_check() } } @@ -307,7 +310,7 @@ impl Handle { } /// Create a future that becomes ready after `duration` of virtual time. - pub fn sleep(&self, duration: Duration) -> crate::sim::time::Sleep { + pub fn sleep(&self, duration: Duration) -> super::time::Sleep { self.executor.time.sleep(duration) } @@ -316,7 +319,7 @@ impl Handle { &self, duration: Duration, future: impl Future, - ) -> Result { + ) -> Result { self.executor.time.timeout(duration, future).await } @@ -646,7 +649,7 @@ impl Receiver { #[cfg(test)] mod tests { use std::sync::{ - atomic::{AtomicBool, AtomicUsize, Ordering}, + atomic::{AtomicUsize, Ordering}, Arc, }; @@ -735,13 +738,12 @@ mod tests { assert_eq!(err, JoinError); } - #[cfg(feature = "simulation")] #[test] - fn sim_std_block_on_can_spawn_local_task_with_explicit_handle() { + fn block_on_can_spawn_local_task_with_explicit_handle() { let mut runtime = Runtime::new(5); let handle = runtime.handle(); let node = handle.create_node().name("local").build(); - let value = crate::sim_std::block_on(&mut runtime, async move { + let value = runtime.block_on(async move { let captured = std::rc::Rc::new(17); node.spawn_local(async move { yield_now().await; @@ -764,34 +766,4 @@ mod tests { assert_eq!(named.name(), Some("replica-1")); assert_ne!(unnamed.id(), named.id()); } - - #[cfg(feature = "simulation")] - #[test] - fn check_determinism_runs_future_twice() { - static CALLS: AtomicUsize = AtomicUsize::new(0); - CALLS.store(0, Ordering::SeqCst); - - let value = crate::sim_std::check_determinism(3, || async { - CALLS.fetch_add(1, Ordering::SeqCst); - yield_now().await; - 13 - }); - - assert_eq!(value, 13); - assert_eq!(CALLS.load(Ordering::SeqCst), 2); - } - - #[cfg(feature = "simulation")] - #[test] - #[should_panic(expected = "non-determinism detected")] - fn check_determinism_rejects_different_scheduler_sequence() { - static FIRST_RUN: AtomicBool = AtomicBool::new(true); - FIRST_RUN.store(true, Ordering::SeqCst); - - crate::sim_std::check_determinism(4, || async { - if FIRST_RUN.swap(false, Ordering::SeqCst) { - yield_now().await; - } - }); - } } diff --git a/crates/runtime/src/sim/executor/task.rs b/crates/runtime-core/src/sim/executor/task.rs similarity index 96% rename from crates/runtime/src/sim/executor/task.rs rename to crates/runtime-core/src/sim/executor/task.rs index bf03a8293d3..3d52349baa1 100644 --- a/crates/runtime/src/sim/executor/task.rs +++ b/crates/runtime-core/src/sim/executor/task.rs @@ -38,7 +38,8 @@ impl JoinHandle { } /// Poll the underlying async_task::Task for its output. - pub(crate) fn poll_join(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + #[doc(hidden)] + pub fn poll_join(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { // async_task::Task implements Future. Polling it drives the wrapped // Abortable future inside the executor. Pin::new(&mut self.task).poll(cx) @@ -96,8 +97,7 @@ impl fmt::Display for JoinError { } } -#[cfg(feature = "simulation")] -impl std::error::Error for JoinError {} +impl core::error::Error for JoinError {} // Shared state between AbortHandle and Abortable. struct AbortState { diff --git a/crates/runtime/src/sim/mod.rs b/crates/runtime-core/src/sim/mod.rs similarity index 83% rename from crates/runtime/src/sim/mod.rs rename to crates/runtime-core/src/sim/mod.rs index ccdcc104991..e2c231828a1 100644 --- a/crates/runtime/src/sim/mod.rs +++ b/crates/runtime-core/src/sim/mod.rs @@ -6,5 +6,6 @@ pub mod time; pub use executor::{ yield_now, AbortHandle, Handle, JoinError, JoinHandle, Node, NodeBuilder, NodeId, Runtime, RuntimeConfig, }; -pub(crate) use rng::DeterminismLog; +#[doc(hidden)] +pub use rng::DeterminismLog; pub use rng::{GlobalRng, Rng}; diff --git a/crates/runtime/src/sim/rng.rs b/crates/runtime-core/src/sim/rng.rs similarity index 93% rename from crates/runtime/src/sim/rng.rs rename to crates/runtime-core/src/sim/rng.rs index c210ff8b781..3555953e764 100644 --- a/crates/runtime/src/sim/rng.rs +++ b/crates/runtime-core/src/sim/rng.rs @@ -9,7 +9,7 @@ pub type Rng = GlobalRng; /// The simulator owns one runtime-wide RNG handle and uses it for scheduler /// choices, probabilistic fault injection, and determinism checks. Hosted /// conveniences such as thread-local current-RNG access and libc random hooks -/// live in `crate::sim_std`, not here. +/// live in the hosted runtime facade, not here. #[derive(Clone, Debug)] pub struct GlobalRng { inner: Arc>, @@ -143,21 +143,24 @@ impl GlobalRng { } #[allow(dead_code)] - pub(crate) fn enable_determinism_log(&self) { + #[doc(hidden)] + pub fn enable_determinism_log(&self) { let mut inner = self.inner.lock(); inner.log = Some(Vec::new()); inner.check = None; } #[allow(dead_code)] - pub(crate) fn enable_determinism_check(&self, log: DeterminismLog) { + #[doc(hidden)] + pub fn enable_determinism_check(&self, log: DeterminismLog) { let mut inner = self.inner.lock(); inner.check = Some((log.0, 0)); inner.log = None; } #[allow(dead_code)] - pub(crate) fn take_determinism_log(&self) -> Option { + #[doc(hidden)] + pub fn take_determinism_log(&self) -> Option { let mut inner = self.inner.lock(); inner .log @@ -167,7 +170,8 @@ impl GlobalRng { } #[allow(dead_code)] - pub(crate) fn finish_determinism_check(&self) -> Result<(), String> { + #[doc(hidden)] + pub fn finish_determinism_check(&self) -> Result<(), String> { let inner = self.inner.lock(); if let Some((log, consumed)) = &inner.check && *consumed != log.len() @@ -183,7 +187,7 @@ impl GlobalRng { } #[derive(Debug, Clone, Eq, PartialEq)] -pub(crate) struct DeterminismLog(Vec); +pub struct DeterminismLog(Vec); fn probability_sample(value: u64, probability: f64) -> bool { if probability <= 0.0 { diff --git a/crates/runtime/src/sim/time/mod.rs b/crates/runtime-core/src/sim/time/mod.rs similarity index 99% rename from crates/runtime/src/sim/time/mod.rs rename to crates/runtime-core/src/sim/time/mod.rs index e454e84561b..242ff0ab84a 100644 --- a/crates/runtime/src/sim/time/mod.rs +++ b/crates/runtime-core/src/sim/time/mod.rs @@ -207,10 +207,11 @@ impl fmt::Display for TimeoutElapsed { } } -impl std::error::Error for TimeoutElapsed {} +impl core::error::Error for TimeoutElapsed {} #[cfg(test)] mod tests { + use alloc::{vec, vec::Vec}; use std::{sync::Arc, time::Duration}; use crate::sim; diff --git a/crates/runtime/src/sim/time/sleep.rs b/crates/runtime-core/src/sim/time/sleep.rs similarity index 100% rename from crates/runtime/src/sim/time/sleep.rs rename to crates/runtime-core/src/sim/time/sleep.rs diff --git a/crates/runtime/Cargo.toml b/crates/runtime/Cargo.toml index 13b3e94e7e7..f36d5a66cee 100644 --- a/crates/runtime/Cargo.toml +++ b/crates/runtime/Cargo.toml @@ -11,12 +11,11 @@ workspace = true [dependencies] tokio.workspace = true -async-task = { version = "4.4", default-features = false, optional = true } -spin = { version = "0.9", default-features = false, features = ["mutex", "spin_mutex"], optional = true } +spacetimedb-runtime-core = { workspace = true, optional = true } libc = { version = "0.2", optional = true } [dev-dependencies] futures.workspace = true [features] -simulation = ["dep:async-task", "dep:spin", "dep:libc"] +simulation = ["dep:spacetimedb-runtime-core", "dep:libc"] diff --git a/crates/runtime/src/sim.rs b/crates/runtime/src/sim.rs new file mode 100644 index 00000000000..339e788363b --- /dev/null +++ b/crates/runtime/src/sim.rs @@ -0,0 +1,129 @@ +//! Std facade for the portable deterministic simulation runtime. +//! +//! The simulator core lives in `spacetimedb-runtime-core` and stays `no_std`. +//! This module preserves the `spacetimedb_runtime::sim` API surface while +//! adding the hosted context behavior that belongs in `spacetimedb-runtime`. + +use core::{ + future::Future, + ops::{Deref, DerefMut}, + time::Duration, +}; + +pub use spacetimedb_runtime_core::sim::{ + buggify, time, yield_now, AbortHandle, GlobalRng, Handle, JoinError, JoinHandle, Node, NodeBuilder, NodeId, Rng, + RuntimeConfig, +}; + +/// Hosted wrapper around the portable simulation runtime. +pub struct Runtime { + inner: spacetimedb_runtime_core::sim::Runtime, +} + +impl Runtime { + /// Create a simulation runtime seeded for deterministic scheduling and RNG. + pub fn new(seed: u64) -> Self { + Self::with_config(RuntimeConfig::new(seed)) + } + + /// Create a simulation runtime from an explicit runtime configuration. + pub fn with_config(config: RuntimeConfig) -> Self { + Self { + inner: spacetimedb_runtime_core::sim::Runtime::with_config(config), + } + } + + /// Drive a top-level future to completion with hosted simulation context installed. + pub fn block_on(&mut self, future: F) -> F::Output { + let _guard = crate::sim_std::enter(self.handle()); + self.inner.block_on(future) + } + + /// Return the amount of virtual time elapsed in this runtime. + pub fn elapsed(&self) -> Duration { + self.inner.elapsed() + } + + /// Get a cloneable handle for spawning tasks and accessing runtime services. + pub fn handle(&self) -> Handle { + self.inner.handle() + } + + /// Create a new simulated node. + pub fn create_node(&self) -> NodeBuilder { + self.inner.create_node() + } + + /// Pause scheduling for a node. + pub fn pause(&self, node: NodeId) { + self.inner.pause(node); + } + + /// Resume scheduling for a previously paused node. + pub fn resume(&self, node: NodeId) { + self.inner.resume(node); + } + + /// Spawn a `Send` future onto a specific simulated node. + pub fn spawn_on(&self, node: NodeId, future: F) -> JoinHandle + where + F: Future + Send + 'static, + F::Output: Send + 'static, + { + self.inner.spawn_on(node, future) + } + + pub fn enable_buggify(&self) { + self.inner.enable_buggify(); + } + + /// Disable probabilistic fault injection for this runtime. + pub fn disable_buggify(&self) { + self.inner.disable_buggify(); + } + + /// Return whether buggify is enabled for this runtime. + pub fn is_buggify_enabled(&self) -> bool { + self.inner.is_buggify_enabled() + } + + /// Sample the default runtime buggify probability. + pub fn buggify(&self) -> bool { + self.inner.buggify() + } + + /// Sample a caller-provided runtime buggify probability. + pub fn buggify_with_prob(&self, probability: f64) -> bool { + self.inner.buggify_with_prob(probability) + } + + pub(crate) fn enable_determinism_log(&self) { + self.inner.enable_determinism_log(); + } + + pub(crate) fn enable_determinism_check(&self, log: spacetimedb_runtime_core::sim::DeterminismLog) { + self.inner.enable_determinism_check(log); + } + + pub(crate) fn take_determinism_log(&self) -> Option { + self.inner.take_determinism_log() + } + + pub(crate) fn finish_determinism_check(&self) -> Result<(), alloc::string::String> { + self.inner.finish_determinism_check() + } +} + +impl Deref for Runtime { + type Target = spacetimedb_runtime_core::sim::Runtime; + + fn deref(&self) -> &Self::Target { + &self.inner + } +} + +impl DerefMut for Runtime { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.inner + } +} diff --git a/crates/runtime/tests/sim_e2e.rs b/crates/runtime/tests/sim_e2e.rs index 973b9ccac15..46270d208ce 100644 --- a/crates/runtime/tests/sim_e2e.rs +++ b/crates/runtime/tests/sim_e2e.rs @@ -1,14 +1,16 @@ #![cfg(feature = "simulation")] #![allow(clippy::disallowed_macros)] -use std::{sync::Arc, time::Duration}; +use std::{ + sync::{Arc, Mutex}, + time::Duration, +}; use futures::{ channel::{mpsc, oneshot}, StreamExt, }; use spacetimedb_runtime::sim::{buggify, Rng, Runtime}; -use spin::Mutex; /// One reply produced by the simulated server. #[derive(Clone, Copy, Debug, Eq, PartialEq)] @@ -156,7 +158,7 @@ fn run_buggified_client_server(seed: u64) -> ClientServerRun { // Receive all 5 requests before processing any replies for _ in 0..5 { let request = request_rx.next().await.expect("client should send request"); - server_events_for_server.lock().push(ServerEvent::Received { + server_events_for_server.lock().unwrap().push(ServerEvent::Received { id: request.id, at: server_handle.now(), }); @@ -169,7 +171,7 @@ fn run_buggified_client_server(seed: u64) -> ClientServerRun { worker_handle.sleep(Duration::from_millis(request.id + 1)).await; // buggify decides whether to drop this request (40% probability) if worker_handle.buggify_with_prob(0.4) { - worker_events.lock().push(ServerEvent::Dropped { + worker_events.lock().unwrap().push(ServerEvent::Dropped { id: request.id, at: worker_handle.now(), }); @@ -182,7 +184,7 @@ fn run_buggified_client_server(seed: u64) -> ClientServerRun { value: request.input * 10, at: worker_handle.now(), }; - worker_events.lock().push(ServerEvent::Replied { + worker_events.lock().unwrap().push(ServerEvent::Replied { id: request.id, at: response.at, }); @@ -230,7 +232,7 @@ fn run_buggified_client_server(seed: u64) -> ClientServerRun { // Drive both client and server to completion let responses = client.await.expect("client task should complete"); server.await.expect("server task should complete"); - (responses, server_events.lock().clone()) + (responses, server_events.lock().unwrap().clone()) }); // --- package the results: client responses, server trace, and total virtual time --- @@ -260,21 +262,21 @@ fn multi_node_runtime_coordinates_pause_resume_and_virtual_time() { let a_handle = handle.clone(); let a_events = Arc::clone(&events); let a = node_a.spawn(async move { - a_events.lock().push(("a_started", a_handle.now())); + a_events.lock().unwrap().push(("a_started", a_handle.now())); a_handle.sleep(Duration::from_millis(3)).await; - a_events.lock().push(("a_finished", a_handle.now())); + a_events.lock().unwrap().push(("a_finished", a_handle.now())); }); let b_handle = handle.clone(); let b_events = Arc::clone(&events); let b = node_b.spawn(async move { - b_events.lock().push(("b_started", b_handle.now())); + b_events.lock().unwrap().push(("b_started", b_handle.now())); b_handle.sleep(Duration::from_millis(2)).await; - b_events.lock().push(("b_finished", b_handle.now())); + b_events.lock().unwrap().push(("b_finished", b_handle.now())); }); handle.sleep(Duration::from_millis(1)).await; - events.lock().push(("main_resumed_b", handle.now())); + events.lock().unwrap().push(("main_resumed_b", handle.now())); node_b.resume(); a.await.expect("node a task should complete"); @@ -282,7 +284,7 @@ fn multi_node_runtime_coordinates_pause_resume_and_virtual_time() { } }); - let events = events.lock(); + let events = events.lock().unwrap(); let get = |name: &str| events.iter().find(|(n, _)| *n == name).map(|(_, t)| *t).unwrap(); // a starts with only per-task overhead accumulated before its first poll From e69b8292eb74f6e8631d4078e6e0707e512886e1 Mon Sep 17 00:00:00 2001 From: Shubham Mishra Date: Fri, 31 Jul 2026 17:11:18 +0530 Subject: [PATCH 3/4] remove EnterGaurd --- Cargo.toml | 4 +- crates/runtime/src/lib.rs | 70 +----------------- crates/runtime/src/sim.rs | 129 ---------------------------------- crates/runtime/src/sim_std.rs | 46 ++++++------ 4 files changed, 23 insertions(+), 226 deletions(-) delete mode 100644 crates/runtime/src/sim.rs diff --git a/Cargo.toml b/Cargo.toml index 9c5d473fd7f..d0b567fecac 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -28,7 +28,7 @@ members = [ "crates/primitives", "crates/query", "crates/runtime", - "crates/runtime-sim", + "crates/runtime-core", "crates/sats", "crates/schema", "crates/smoketests", @@ -153,7 +153,7 @@ spacetimedb-primitives = { path = "crates/primitives", version = "=2.7.1" } spacetimedb-query = { path = "crates/query", version = "=2.7.1" } spacetimedb-query-builder = { path = "crates/query-builder", version = "=2.7.1" } spacetimedb-runtime = { path = "crates/runtime", version = "=2.7.1" } -spacetimedb-runtime-sim = { path = "crates/runtime-sim", version = "=2.7.1" } +spacetimedb-runtime-core = { path = "crates/runtime-core", version = "=2.7.1" } spacetimedb-sats = { path = "crates/sats", version = "=2.7.1" } spacetimedb-schema = { path = "crates/schema", version = "=2.7.1" } spacetimedb-snapshot = { path = "crates/snapshot", version = "=2.7.1" } diff --git a/crates/runtime/src/lib.rs b/crates/runtime/src/lib.rs index b7b589e51c3..c6192e1b738 100644 --- a/crates/runtime/src/lib.rs +++ b/crates/runtime/src/lib.rs @@ -1,6 +1,3 @@ -#[cfg(feature = "simulation")] -extern crate alloc; - use core::{ fmt, future::Future, @@ -11,7 +8,7 @@ use core::{ }; #[cfg(feature = "simulation")] -pub mod sim; +pub use spacetimedb_runtime_core::sim; #[cfg(feature = "simulation")] pub mod sim_std; @@ -64,11 +61,6 @@ pub struct AbortHandle { inner: AbortHandleInner, } -#[must_use = "runtime context exits immediately unless the guard is held"] -pub struct EnterGuard<'a> { - _inner: EnterGuardInner<'a>, -} - enum JoinHandleInner { Tokio(tokio::task::JoinHandle), #[cfg(feature = "simulation")] @@ -94,16 +86,6 @@ enum AbortHandleInner { Simulation(sim::AbortHandle), } -enum EnterGuardInner<'a> { - Tokio { - _guard: tokio::runtime::EnterGuard<'a>, - }, - #[cfg(feature = "simulation")] - Simulation { - _guard: sim_std::EnterGuard, - }, -} - #[derive(Debug)] pub struct JoinError { inner: JoinErrorInner, @@ -234,18 +216,6 @@ impl fmt::Display for RuntimeTimeout { impl std::error::Error for RuntimeTimeout {} -pub fn spawn(future: impl Future + Send + 'static) -> JoinHandle { - Handle::current().spawn(future) -} - -pub async fn spawn_blocking(f: F) -> R -where - F: FnOnce() -> R + Send + 'static, - R: Send + 'static, -{ - Handle::current().spawn_blocking(f).await -} - impl Handle { pub fn tokio(handle: TokioHandle) -> Self { Self::Tokio(handle) @@ -254,15 +224,6 @@ impl Handle { pub fn tokio_current() -> Self { Self::tokio(TokioHandle::current()) } - - pub fn current() -> Self { - #[cfg(feature = "simulation")] - if let Some(handle) = sim_std::try_current() { - return Self::Simulation(handle); - } - - Self::tokio_current() - } } #[cfg(feature = "simulation")] @@ -273,20 +234,6 @@ impl Handle { } impl Handle { - pub fn enter(&self) -> EnterGuard<'_> { - match self { - Self::Tokio(handle) => EnterGuard { - _inner: EnterGuardInner::Tokio { _guard: handle.enter() }, - }, - #[cfg(feature = "simulation")] - Self::Simulation(handle) => EnterGuard { - _inner: EnterGuardInner::Simulation { - _guard: sim_std::enter(handle.clone()), - }, - }, - } - } - pub fn spawn(&self, future: impl Future + Send + 'static) -> JoinHandle { match self { Self::Tokio(handle) => JoinHandle { @@ -408,19 +355,4 @@ mod tests { assert!(!flag.load(Ordering::Acquire)); }); } - - #[cfg(feature = "simulation")] - #[test] - fn free_spawn_apis_use_current_simulation_runtime() { - use crate::sim::Runtime; - let mut rt = Runtime::new(4); - - let output = rt.block_on(async { - let spawned = crate::spawn(async { 31 }).await.unwrap(); - let blocking = crate::spawn_blocking(|| 41).await; - spawned + blocking - }); - - assert_eq!(output, 72); - } } diff --git a/crates/runtime/src/sim.rs b/crates/runtime/src/sim.rs deleted file mode 100644 index 339e788363b..00000000000 --- a/crates/runtime/src/sim.rs +++ /dev/null @@ -1,129 +0,0 @@ -//! Std facade for the portable deterministic simulation runtime. -//! -//! The simulator core lives in `spacetimedb-runtime-core` and stays `no_std`. -//! This module preserves the `spacetimedb_runtime::sim` API surface while -//! adding the hosted context behavior that belongs in `spacetimedb-runtime`. - -use core::{ - future::Future, - ops::{Deref, DerefMut}, - time::Duration, -}; - -pub use spacetimedb_runtime_core::sim::{ - buggify, time, yield_now, AbortHandle, GlobalRng, Handle, JoinError, JoinHandle, Node, NodeBuilder, NodeId, Rng, - RuntimeConfig, -}; - -/// Hosted wrapper around the portable simulation runtime. -pub struct Runtime { - inner: spacetimedb_runtime_core::sim::Runtime, -} - -impl Runtime { - /// Create a simulation runtime seeded for deterministic scheduling and RNG. - pub fn new(seed: u64) -> Self { - Self::with_config(RuntimeConfig::new(seed)) - } - - /// Create a simulation runtime from an explicit runtime configuration. - pub fn with_config(config: RuntimeConfig) -> Self { - Self { - inner: spacetimedb_runtime_core::sim::Runtime::with_config(config), - } - } - - /// Drive a top-level future to completion with hosted simulation context installed. - pub fn block_on(&mut self, future: F) -> F::Output { - let _guard = crate::sim_std::enter(self.handle()); - self.inner.block_on(future) - } - - /// Return the amount of virtual time elapsed in this runtime. - pub fn elapsed(&self) -> Duration { - self.inner.elapsed() - } - - /// Get a cloneable handle for spawning tasks and accessing runtime services. - pub fn handle(&self) -> Handle { - self.inner.handle() - } - - /// Create a new simulated node. - pub fn create_node(&self) -> NodeBuilder { - self.inner.create_node() - } - - /// Pause scheduling for a node. - pub fn pause(&self, node: NodeId) { - self.inner.pause(node); - } - - /// Resume scheduling for a previously paused node. - pub fn resume(&self, node: NodeId) { - self.inner.resume(node); - } - - /// Spawn a `Send` future onto a specific simulated node. - pub fn spawn_on(&self, node: NodeId, future: F) -> JoinHandle - where - F: Future + Send + 'static, - F::Output: Send + 'static, - { - self.inner.spawn_on(node, future) - } - - pub fn enable_buggify(&self) { - self.inner.enable_buggify(); - } - - /// Disable probabilistic fault injection for this runtime. - pub fn disable_buggify(&self) { - self.inner.disable_buggify(); - } - - /// Return whether buggify is enabled for this runtime. - pub fn is_buggify_enabled(&self) -> bool { - self.inner.is_buggify_enabled() - } - - /// Sample the default runtime buggify probability. - pub fn buggify(&self) -> bool { - self.inner.buggify() - } - - /// Sample a caller-provided runtime buggify probability. - pub fn buggify_with_prob(&self, probability: f64) -> bool { - self.inner.buggify_with_prob(probability) - } - - pub(crate) fn enable_determinism_log(&self) { - self.inner.enable_determinism_log(); - } - - pub(crate) fn enable_determinism_check(&self, log: spacetimedb_runtime_core::sim::DeterminismLog) { - self.inner.enable_determinism_check(log); - } - - pub(crate) fn take_determinism_log(&self) -> Option { - self.inner.take_determinism_log() - } - - pub(crate) fn finish_determinism_check(&self) -> Result<(), alloc::string::String> { - self.inner.finish_determinism_check() - } -} - -impl Deref for Runtime { - type Target = spacetimedb_runtime_core::sim::Runtime; - - fn deref(&self) -> &Self::Target { - &self.inner - } -} - -impl DerefMut for Runtime { - fn deref_mut(&mut self) -> &mut Self::Target { - &mut self.inner - } -} diff --git a/crates/runtime/src/sim_std.rs b/crates/runtime/src/sim_std.rs index 39d200eab43..8153482519e 100644 --- a/crates/runtime/src/sim_std.rs +++ b/crates/runtime/src/sim_std.rs @@ -1,14 +1,14 @@ //! Std-hosted entry points for running the deterministic simulator in tests. //! //! The portable simulator lives in [`crate::sim`]. This module is deliberately -//! host-specific: it installs thread-local context while a simulation is -//! running, checks determinism by replaying a seed in fresh OS threads, and -//! intercepts a few libc calls so std code cannot silently escape determinism. +//! host-specific: it marks a thread as simulation-owned while hosted tests run, +//! checks determinism by replaying a seed in fresh OS threads, and intercepts a +//! few libc calls so std code cannot silently escape determinism. #![allow(clippy::disallowed_macros)] -use alloc::boxed::Box; -use core::{cell::RefCell, future::Future}; +use core::{cell::Cell, future::Future}; +use std::boxed::Box; use std::sync::OnceLock; use crate::sim; @@ -21,6 +21,7 @@ use crate::sim; /// tests that execute inside a hosted process. While the future runs, this /// marks the thread as inside simulation so OS thread spawns can be rejected. pub fn block_on(runtime: &mut sim::Runtime, future: F) -> F::Output { + let _guard = enter(); runtime.block_on(future) } @@ -67,40 +68,33 @@ fn panic_with_seed(seed: u64, payload: Box) -> ! { // Simulation thread context. -// Ambient state used only while hosted code has entered a simulation runtime. +// Ambient hosted state used only while sim_std is driving a simulation runtime. // -// The simulator itself stays explicit-handle based. This thread-local stores -// the current handle for runtime-generic code that uses [`crate::Handle::current`], -// and its presence marks the current OS thread as simulation-owned so host -// thread creation can be rejected. +// The simulator itself stays explicit-handle based. This flag only marks the +// current OS thread as simulation-owned so host thread creation and randomness +// hooks can reject accidental escapes from deterministic simulation. thread_local! { - static CURRENT_RUNTIME: RefCell> = const { RefCell::new(None) }; + static IN_SIMULATION: Cell = const { Cell::new(false) }; } #[must_use = "simulation runtime context exits immediately unless the guard is held"] -pub struct EnterGuard { - previous_runtime: Option, -} - -/// Enter a simulation runtime on the current OS thread until the returned guard drops. -pub fn enter(handle: sim::Handle) -> EnterGuard { - let previous_runtime = CURRENT_RUNTIME.with(|current| current.replace(Some(handle))); - EnterGuard { previous_runtime } -} +struct EnterGuard; -/// Return the simulation runtime entered on this OS thread, if any. -pub(crate) fn try_current() -> Option { - CURRENT_RUNTIME.with(|current| current.borrow().clone()) +fn enter() -> EnterGuard { + IN_SIMULATION.with(|current| { + assert!(!current.replace(true), "nested hosted simulation block_on"); + }); + EnterGuard } fn in_simulation() -> bool { - CURRENT_RUNTIME.with(|current| current.borrow().is_some()) + IN_SIMULATION.with(Cell::get) } impl Drop for EnterGuard { fn drop(&mut self) { - CURRENT_RUNTIME.with(|current| { - current.replace(self.previous_runtime.take()); + IN_SIMULATION.with(|current| { + assert!(current.replace(false), "simulation context guard dropped without enter"); }); } } From 589709b9a44cdaa8fae9115a727b88993276de7a Mon Sep 17 00:00:00 2001 From: Shubham Mishra Date: Fri, 31 Jul 2026 17:17:56 +0530 Subject: [PATCH 4/4] sim feature --- crates/runtime-core/Cargo.toml | 8 ++++++-- crates/runtime-core/src/lib.rs | 2 ++ crates/runtime/Cargo.toml | 2 +- 3 files changed, 9 insertions(+), 3 deletions(-) diff --git a/crates/runtime-core/Cargo.toml b/crates/runtime-core/Cargo.toml index 6d82b8f03aa..a3369a69f89 100644 --- a/crates/runtime-core/Cargo.toml +++ b/crates/runtime-core/Cargo.toml @@ -9,6 +9,10 @@ rust-version.workspace = true [lints] workspace = true +[features] +default = [] +sim = ["dep:async-task", "dep:spin"] + [dependencies] -async-task = { version = "4.4", default-features = false } -spin = { version = "0.9", default-features = false, features = ["mutex", "spin_mutex"] } +async-task = { version = "4.4", default-features = false, optional = true } +spin = { version = "0.9", default-features = false, features = ["mutex", "spin_mutex"], optional = true } diff --git a/crates/runtime-core/src/lib.rs b/crates/runtime-core/src/lib.rs index 8296506b759..f7590ada98b 100644 --- a/crates/runtime-core/src/lib.rs +++ b/crates/runtime-core/src/lib.rs @@ -1,7 +1,9 @@ #![no_std] +#[cfg(feature = "sim")] extern crate alloc; #[cfg(test)] extern crate std; +#[cfg(feature = "sim")] pub mod sim; diff --git a/crates/runtime/Cargo.toml b/crates/runtime/Cargo.toml index f36d5a66cee..c8affea0f48 100644 --- a/crates/runtime/Cargo.toml +++ b/crates/runtime/Cargo.toml @@ -18,4 +18,4 @@ libc = { version = "0.2", optional = true } futures.workspace = true [features] -simulation = ["dep:spacetimedb-runtime-core", "dep:libc"] +simulation = ["dep:spacetimedb-runtime-core", "spacetimedb-runtime-core/sim", "dep:libc"]