diff --git a/crates/runtime-core/src/sim/executor/mod.rs b/crates/runtime-core/src/sim/executor/mod.rs index fbb7f7c0cf2..b9ce879e882 100644 --- a/crates/runtime-core/src/sim/executor/mod.rs +++ b/crates/runtime-core/src/sim/executor/mod.rs @@ -10,22 +10,42 @@ use core::{ use spin::Mutex; -use super::{time::TimeHandle, Rng}; +use crate::sim::executor::task::Abortable; + +use super::rng::Ratio; +use super::{net, time::TimeHandle, Rng}; mod task; -use task::Abortable; pub use task::{AbortHandle, JoinError, JoinHandle}; type Runnable = async_task::Runnable; +const READY_TASK_BUDGET: usize = 256; + #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub struct RuntimeConfig { pub seed: u64, + pub network: Option, + pub node_faults: NodeFaultOptions, } impl RuntimeConfig { - pub const fn new(seed: u64) -> Self { - Self { seed } + pub fn new(seed: u64) -> Self { + Self { + seed, + network: None, + node_faults: NodeFaultOptions::default(), + } + } + + pub fn with_network(mut self, network: net::Options) -> Self { + self.network = network.into(); + self + } + + pub fn with_node_faults(mut self, node_faults: NodeFaultOptions) -> Self { + self.node_faults = node_faults; + self } } @@ -35,6 +55,29 @@ impl Default for RuntimeConfig { } } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct NodeFaultOptions { + /// Probability per fault tick that a running node crashes. + pub crash_node_probability: Ratio, + /// Probability per fault tick that a crashed node restarts. + pub restart_node_probability: Ratio, + /// Probability per fault tick that a running node pauses. + pub pause_node_probability: Ratio, + /// Probability per fault tick that a paused node resumes. + pub unpause_node_probability: Ratio, +} + +impl Default for NodeFaultOptions { + fn default() -> Self { + Self { + crash_node_probability: Ratio::ZERO, + restart_node_probability: Ratio::ZERO, + pause_node_probability: Ratio::ZERO, + unpause_node_probability: Ratio::ZERO, + } + } +} + /// A unique identifier for a simulated node. #[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)] pub struct NodeId(u64); @@ -94,6 +137,16 @@ impl Node { self.config.name.as_deref() } + /// Return the simulated network endpoint for this node. + pub fn net(&self) -> Option { + self.handle.network().map(|net| net.on_node(self.id)) + } + + /// Crash this node and invalidate all tasks spawned before the crash. + pub fn crash(&self) { + self.handle.crash_node(self.id); + } + /// Pause scheduling for this node. pub fn pause(&self) { self.handle.pause(self.id); @@ -104,6 +157,11 @@ impl Node { self.handle.resume(self.id); } + /// Restart this node and invalidate all tasks spawned before the restart. + pub fn restart(&self) { + self.handle.restart_node(self.id); + } + /// Spawn a `Send` future onto this simulated node. pub fn spawn(&self, future: F) -> JoinHandle where @@ -251,6 +309,11 @@ pub struct Handle { } impl Handle { + /// Return the shared simulated network for this runtime. + pub fn network(&self) -> Option { + self.executor.net.clone() + } + /// Create a new simulated node owned by this runtime. pub fn create_node(&self) -> NodeBuilder { NodeBuilder { @@ -259,9 +322,13 @@ impl Handle { } } + fn node_config(&self, node: NodeId) -> Arc { + self.executor.node_config(node) + } + fn build_node(&self, config: NodeConfig) -> Node { - let id = self.executor.create_node(config.clone()); - let config = self.executor.node_config(id); + let id = self.executor.create_node(config); + let config = self.node_config(id); Node { id, handle: self.clone(), @@ -274,11 +341,21 @@ impl Handle { self.executor.pause(node); } + /// Crash a node until it is restarted. + pub fn crash_node(&self, node: NodeId) { + self.executor.crash_node(node); + } + /// Resume scheduling for a node and requeue any buffered tasks for it. pub fn resume(&self, node: NodeId) { self.executor.resume(node); } + /// Restart a node and invalidate all tasks spawned before the restart. + pub fn restart_node(&self, node: NodeId) { + self.executor.restart_node(node); + } + /// Spawn a `Send` future onto a specific simulated node. pub fn spawn_on(&self, node: NodeId, future: F) -> JoinHandle where @@ -323,6 +400,15 @@ impl Handle { self.executor.time.timeout(duration, future).await } + /// Yield this task back to the simulation scheduler once. + pub async fn yield_now(&self) { + yield_now().await + } + + pub fn block_on(&self, future: F) -> F::Output { + self.executor.block_on(future) + } + pub fn enable_buggify(&self) { self.executor.enable_buggify(); } @@ -357,9 +443,11 @@ struct Executor { queue: Receiver, sender: Sender, nodes: spin::Mutex>>, + node_faults: NodeFaultOptions, next_node: AtomicU64, rng: Rng, time: TimeHandle, + net: Option, } impl Executor { @@ -367,14 +455,27 @@ impl Executor { fn new(config: RuntimeConfig) -> Self { let queue = Queue::new(); let mut nodes = BTreeMap::new(); - nodes.insert(NodeId::MAIN, Arc::new(NodeRecord::default())); + let time = TimeHandle::new(); + let rng = Rng::new(config.seed); + + nodes.insert(NodeId::MAIN, Arc::new(NodeRecord::new(NodeConfig::default()))); + + let net = config + .network + .map(|config| net::Network::new(time.clone(), rng.clone(), config)); + if let Some(net) = &net { + net.register_node(NodeId::MAIN); + } + Self { queue: queue.receiver(), sender: queue.sender(), nodes: spin::Mutex::new(nodes), + node_faults: config.node_faults, next_node: AtomicU64::new(1), - rng: Rng::new(config.seed), - time: TimeHandle::new(), + rng, + time, + net, } } @@ -404,34 +505,61 @@ impl Executor { fn create_node(&self, config: NodeConfig) -> NodeId { let id = NodeId(self.next_node.fetch_add(1, Ordering::Relaxed)); - self.nodes.lock().insert( - id, - Arc::new(NodeRecord { - config: Arc::new(config), - state: NodeState::default(), - }), - ); - id - } + self.nodes.lock().insert(id, Arc::new(NodeRecord::new(config))); - fn node_config(&self, node: NodeId) -> Arc { - self.node_record(node).config.clone() + if let Some(net) = &self.net { + net.register_node(id); + } + + id } /// Mark a node as paused so newly selected runnables are buffered. fn pause(&self, node: NodeId) { - self.node_record(node).state.paused.store(true, Ordering::Relaxed); + assert_ne!(node, NodeId::MAIN, "cannot pause the main simulation node"); + + self.node_state(node).paused.store(true, Ordering::Relaxed); + if let Some(net) = &self.net { + net.isolate_node(node); + } + } + + /// Mark a node as crashed until it is restarted. + fn crash_node(&self, node: NodeId) { + assert_ne!(node, NodeId::MAIN, "cannot crash the main simulation node"); + + let state = self.node_state(node); + state.crashed.store(true, Ordering::Release); + state.paused.store(false, Ordering::Release); + state.paused_queue.lock().clear(); + if let Some(net) = &self.net { + net.isolate_node(node); + } } /// Mark a node as runnable again and requeue any buffered tasks for it. fn resume(&self, node: NodeId) { - let record = self.node_record(node); - record.state.paused.store(false, Ordering::Relaxed); - - let mut paused = record.state.paused_queue.lock(); - for runnable in paused.drain(..) { + let state = self.node_state(node); + state.paused.store(false, Ordering::Relaxed); + let runnables = core::mem::take(&mut *state.paused_queue.lock()); + for runnable in runnables { self.sender.send(runnable); } + if let Some(net) = &self.net { + net.unisolate_node(node); + } + } + + /// Mark a crashed node as running again. + fn restart_node(&self, node: NodeId) { + assert_ne!(node, NodeId::MAIN, "cannot restart the main simulation node"); + + let state = self.node_state(node); + state.crashed.store(false, Ordering::Release); + state.paused.store(false, Ordering::Release); + if let Some(net) = &self.net { + net.unisolate_node(node); + } } /// Spawn a `Send` task and enqueue its runnable on the shared runtime queue. @@ -477,9 +605,10 @@ impl Executor { #[track_caller] /// Run the top-level future until completion. /// - /// The executor repeatedly drains runnable tasks, then advances virtual - /// time to the next timer when the queue is empty. If neither runnable work - /// nor timers remain, the simulation is considered deadlocked. + /// The executor polls a bounded random batch of runnable tasks, samples + /// simulated fault sources at one captured instant, then advances virtual + /// time only when no current-time source can make progress. If neither + /// runnable work nor timers remain, the simulation is considered deadlocked. fn block_on(&self, future: F) -> F::Output { let sender = self.sender.clone(); let (runnable, mut task) = unsafe { @@ -490,16 +619,24 @@ impl Executor { runnable.schedule(); loop { - self.run_all_ready(); - if task.is_finished() { - let waker = Waker::noop(); - return match Pin::new(&mut task).poll(&mut Context::from_waker(waker)) { - Poll::Ready(output) => output, - Poll::Pending => unreachable!("task.is_finished() was true"), - }; + if let Some(output) = poll_finished_task(&mut task) { + return output; } - if self.time.wake_next_timer() { + let task_progress = self.run_ready_budget(READY_TASK_BUDGET); + if let Some(output) = poll_finished_task(&mut task) { + return output; + } + + let now = self.time.now(); + let node_progress = self.node_fault_ticks(now); + let network_progress = self.net_tick(now); + let timer_progress = self.time.wake_due_timers(); + if task_progress || node_progress || network_progress || timer_progress { + continue; + } + + if self.advance_to_next_network_deadline(now) || self.time.wake_next_timer() { continue; } @@ -507,24 +644,121 @@ impl Executor { } } - /// Drain the runnable queue, selecting tasks in deterministic RNG order. + fn net_tick(&self, now: Duration) -> bool { + if let Some(net) = &self.net { + net.tick(now) + } else { + false + } + } + + fn next_network_delivery_deadline(&self, now: Duration) -> Option { + self.net.as_ref().and_then(|net| net.next_delivery_deadline(now)) + } + + fn advance_to_next_network_deadline(&self, now: Duration) -> bool { + let Some(network_deadline) = self.next_network_delivery_deadline(now) else { + return false; + }; + + if let Some(timer_deadline) = self.time.next_timer_deadline() + && network_deadline > timer_deadline + { + return false; + } + + self.time.advance_to(network_deadline) + } + + fn node_fault_ticks(&self, now: Duration) -> bool { + let options = self.node_faults; + + let nodes = { + let nodes = self.nodes.lock(); + nodes + .keys() + .copied() + .filter(|node| *node != NodeId::MAIN) + .collect::>() + }; + + for node in nodes { + if self.node_crash_tick(node, options, now) || self.node_pause_tick(node, options, now) { + return true; + } + } + + false + } + + fn node_crash_tick(&self, node: NodeId, options: NodeFaultOptions, _now: Duration) -> bool { + let record = self.node_state(node); + if record.crashed.load(Ordering::Acquire) { + if self.rng.buggify_ratio(options.restart_node_probability) { + self.restart_node(node); + return true; + } + } else if !record.paused.load(Ordering::Relaxed) && self.rng.buggify_ratio(options.crash_node_probability) { + self.crash_node(node); + return true; + } + + false + } + + fn node_pause_tick(&self, node: NodeId, options: NodeFaultOptions, _now: Duration) -> bool { + let record = self.node_state(node); + if record.crashed.load(Ordering::Acquire) { + return false; + } + + if record.paused.load(Ordering::Relaxed) { + if self.rng.buggify_ratio(options.unpause_node_probability) { + self.resume(node); + true + } else { + false + } + } else if self.rng.buggify_ratio(options.pause_node_probability) { + self.pause(node); + true + } else { + false + } + } + + /// Poll a bounded batch from the runnable queue in deterministic RNG order. /// - /// Paused-node tasks are diverted into that node's paused buffer instead of - /// being polled immediately. - fn run_all_ready(&self) { - while let Some(runnable) = self.queue.try_recv_random(&self.rng) { + /// Returning to the outer scheduler after a fixed quantum lets timers and + /// network deliveries make progress even when CPU-ready tasks keep + /// re-scheduling themselves. Paused-node tasks are diverted into that node's + /// paused buffer instead of being polled immediately. + fn run_ready_budget(&self, budget: usize) -> bool { + assert!(budget > 0, "ready task budget must be non-zero"); + + let mut progressed = false; + for _ in 0..budget { + let Some(runnable) = self.queue.try_recv_random(&self.rng) else { + break; + }; + + progressed = true; let node = *runnable.metadata(); - let record = self.node_record(node); - if record.state.paused.load(Ordering::Relaxed) { - record.state.paused_queue.lock().push(runnable); + let state = self.node_state(node); + if state.is_crashed() { + continue; + } + if state.is_paused() { + state.paused_queue.lock().push(runnable); continue; } runnable.run(); - // Advance virtual time by 100ns–1μs per task poll to model execution cost. + // Advance virtual time by 100ns-1us per task poll to model execution cost. // Using the runtime RNG keeps overhead deterministic by seed. let nanos = 100 + (self.rng.next_u64() % 901); self.time.advance(Duration::from_nanos(nanos)); } + progressed } /// Look up the record for a node, panicking if the node is unknown. @@ -536,25 +770,74 @@ impl Executor { .unwrap_or_else(|| panic!("unknown simulated node {node}")) } + fn node_config(&self, node: NodeId) -> Arc { + self.node_record(node).config.clone() + } + + fn node_state(&self, node: NodeId) -> Arc { + self.node_record(node).state.clone() + } + fn assert_known_node(&self, node: NodeId) { - let _ = self.node_record(node); + let _ = self.node_state(node); } } -/// One simulated node's immutable metadata plus scheduler state. -#[derive(Clone, Default)] +fn poll_finished_task(task: &mut async_task::Task) -> Option { + if !task.is_finished() { + return None; + } + + let waker = Waker::noop(); + match Pin::new(task).poll(&mut Context::from_waker(waker)) { + Poll::Ready(output) => Some(output), + Poll::Pending => unreachable!("task.is_finished() was true"), + } +} + +/// Complete executor record for a simulated node. struct NodeRecord { config: Arc, - state: NodeState, + state: Arc, +} + +impl NodeRecord { + fn new(config: NodeConfig) -> Self { + Self { + config: Arc::new(config), + state: Arc::new(NodeState::default()), + } + } } /// Per-node scheduler state shared by tasks assigned to that node. -#[derive(Clone, Default)] +#[derive(Clone)] struct NodeState { paused: Arc, + crashed: Arc, paused_queue: Arc>>, } +impl Default for NodeState { + fn default() -> Self { + Self { + paused: Arc::new(AtomicBool::new(false)), + crashed: Arc::new(AtomicBool::new(false)), + paused_queue: Arc::new(Mutex::new(Vec::new())), + } + } +} + +impl NodeState { + fn is_crashed(&self) -> bool { + self.crashed.load(Ordering::Relaxed) + } + + fn is_paused(&self) -> bool { + self.paused.load(Ordering::Relaxed) + } +} + /// Yield back to the scheduler once. /// /// This is the smallest explicit interleaving point available to simulated @@ -756,6 +1039,25 @@ mod tests { assert_eq!(value, 17); } + #[test] + fn block_on_returns_while_background_task_stays_ready() { + let mut runtime = Runtime::new(10); + let handle = runtime.handle(); + let node = runtime.create_node().name("hot").build(); + let _hot = node.spawn(async { + loop { + yield_now().await; + } + }); + + let value = runtime.block_on(async move { + handle.sleep(Duration::from_micros(1)).await; + 42 + }); + + assert_eq!(value, 42); + } + #[test] fn node_builder_sets_name() { let runtime = Runtime::new(9); diff --git a/crates/runtime-core/src/sim/mod.rs b/crates/runtime-core/src/sim/mod.rs index e2c231828a1..eeefd25cfda 100644 --- a/crates/runtime-core/src/sim/mod.rs +++ b/crates/runtime-core/src/sim/mod.rs @@ -1,11 +1,14 @@ pub mod buggify; mod executor; +pub mod net; +mod probability; mod rng; pub mod time; pub use executor::{ - yield_now, AbortHandle, Handle, JoinError, JoinHandle, Node, NodeBuilder, NodeId, Runtime, RuntimeConfig, + yield_now, AbortHandle, Handle, JoinError, JoinHandle, Node, NodeBuilder, NodeFaultOptions, NodeId, Runtime, + RuntimeConfig, }; #[doc(hidden)] pub use rng::DeterminismLog; -pub use rng::{GlobalRng, Rng}; +pub use rng::{GlobalRng, Ratio, Rng}; diff --git a/crates/runtime-core/src/sim/net.rs b/crates/runtime-core/src/sim/net.rs new file mode 100644 index 00000000000..28951368d48 --- /dev/null +++ b/crates/runtime-core/src/sim/net.rs @@ -0,0 +1,874 @@ +use alloc::{ + collections::{BTreeMap, BTreeSet, VecDeque}, + sync::Arc, + vec::Vec, +}; +use core::{ + fmt, + future::Future, + pin::Pin, + task::{Context, Poll, Waker}, + time::Duration, +}; + +use spin::Mutex; + +use super::probability::sample_duration_between; +use super::rng::Ratio; +use super::{time::TimeHandle, NodeId, Rng}; + +/// Directed network path from one simulated node to another. +#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)] +struct Path { + from: NodeId, + to: NodeId, +} + +impl Path { + fn new(from: NodeId, to: NodeId) -> Self { + Self { from, to } + } + + fn crosses_partition(self, left: &BTreeSet) -> bool { + self.from != self.to && left.contains(&self.from) != left.contains(&self.to) + } +} + +/// Per-runtime simulated network options. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct Options { + /// Minimum simulated one-way packet delay. + pub one_way_delay_min: Duration, + /// Mean simulated one-way packet delay. + pub one_way_delay_mean: Duration, + /// Probability that the network drops an outbound packet. + pub packet_loss_probability: Ratio, + /// Maximum number of in-flight packets on a single directed path. + pub path_maximum_capacity: usize, + /// Mean simulated duration for an automatically clogged path. + pub path_clog_duration_mean: Duration, + /// Probability that a send automatically clogs its directed path. + pub path_clog_probability: Ratio, + /// Probability that the network duplicates an outbound packet. + pub packet_replay_probability: Ratio, + /// Probability that an unpartitioned network enters a partitioned state on a tick. + pub partition_probability: Ratio, + /// Probability that a newly created automatic partition blocks only one direction. + pub one_way_partition_probability: Ratio, + /// Probability that an automatically partitioned network heals on a tick. + pub unpartition_probability: Ratio, + /// Minimum ticks an automatic partition remains stable before healing is sampled. + pub partition_stability_ticks: u32, + /// Minimum ticks an automatically healed network remains stable before partitioning is sampled. + pub unpartition_stability_ticks: u32, +} + +const DEFAULT_ONE_WAY_DELAY_MIN: Duration = Duration::from_millis(1); +const DEFAULT_ONE_WAY_DELAY_MEAN: Duration = Duration::from_millis(10); +const DEFAULT_PATH_MAXIMUM_CAPACITY: usize = 1024; +const DEFAULT_PATH_CLOG_DURATION_MEAN: Duration = Duration::from_millis(100); + +impl Default for Options { + fn default() -> Self { + Self { + one_way_delay_min: DEFAULT_ONE_WAY_DELAY_MIN, + one_way_delay_mean: DEFAULT_ONE_WAY_DELAY_MEAN, + packet_loss_probability: Ratio::ZERO, + path_maximum_capacity: DEFAULT_PATH_MAXIMUM_CAPACITY, + path_clog_duration_mean: DEFAULT_PATH_CLOG_DURATION_MEAN, + path_clog_probability: Ratio::ZERO, + packet_replay_probability: Ratio::ZERO, + partition_probability: Ratio::ZERO, + one_way_partition_probability: Ratio::ZERO, + unpartition_probability: Ratio::ZERO, + partition_stability_ticks: 0, + unpartition_stability_ticks: 0, + } + } +} + +/// Shared deterministic network state for one simulation runtime. +#[derive(Clone, Debug)] +pub struct Network { + inner: Arc>, + rng: Rng, + time: TimeHandle, +} + +impl Network { + pub(crate) fn new(time: TimeHandle, rng: Rng, options: Options) -> Self { + Self { + inner: Arc::new(Mutex::new(NetworkState::new(options))), + rng, + time, + } + } + + pub(crate) fn register_node(&self, node: NodeId) { + self.inner.lock().nodes.entry(node).or_default(); + } + + /// Return a handle that sends from and receives for `node`. + pub fn on_node(&self, node: NodeId) -> NodeNetwork { + self.register_node(node); + NodeNetwork { + node, + network: self.clone(), + } + } + + /// Clear network-owned faults while preserving process isolation, inboxes, and in-flight packets. + pub fn clear_faults(&self) { + let mut state = self.inner.lock(); + state.clear_node_faults(); + state.clear_link_faults(); + } + + /// Clear link-level faults while preserving node-level clogs. + pub fn clear_link_faults(&self) { + self.inner.lock().clear_link_faults(); + } + + /// Return whether inbound or outbound delivery is blocked for `node`. + pub fn is_node_clogged(&self, node: NodeId) -> bool { + self.inner.lock().is_node_clogged(node) + } + + /// Block all inbound and outbound delivery for `node`. + pub fn clog_node(&self, node: NodeId) { + let mut state = self.inner.lock(); + state.clogged_node_in.insert(node); + state.clogged_node_out.insert(node); + } + + /// Clear the manual inbound and outbound node clog. + pub fn unclog_node(&self, node: NodeId) { + let mut state = self.inner.lock(); + state.clogged_node_in.remove(&node); + state.clogged_node_out.remove(&node); + } + + /// Isolate a paused or crashed process from all network traffic. + pub(crate) fn isolate_node(&self, node: NodeId) { + self.inner.lock().isolated_nodes.insert(node); + } + + /// Remove process isolation without clearing manually configured node clogs. + pub(crate) fn unisolate_node(&self, node: NodeId) { + self.inner.lock().isolated_nodes.remove(&node); + } + + /// Block all inbound delivery to `node`. + pub fn clog_node_in(&self, node: NodeId) { + self.inner.lock().clogged_node_in.insert(node); + } + + /// Clear the manual inbound node clog. + pub fn unclog_node_in(&self, node: NodeId) { + self.inner.lock().clogged_node_in.remove(&node); + } + + /// Block all outbound delivery from `node`. + pub fn clog_node_out(&self, node: NodeId) { + self.inner.lock().clogged_node_out.insert(node); + } + + /// Clear the manual outbound node clog. + pub fn unclog_node_out(&self, node: NodeId) { + self.inner.lock().clogged_node_out.remove(&node); + } + + /// Block delivery from `from` to `to`. + pub fn clog_link(&self, from: NodeId, to: NodeId) { + self.inner.lock().clogged_links.insert(Path::new(from, to)); + } + + /// Clear the manual directed-link clog. + pub fn unclog_link(&self, from: NodeId, to: NodeId) { + self.inner.lock().clogged_links.remove(&Path::new(from, to)); + } + + /// Return whether a payload from `from` to `to` would currently be blocked. + pub fn is_blocked(&self, from: NodeId, to: NodeId) -> bool { + let mut state = self.inner.lock(); + state.expire_path_clogs(self.time.now()); + state.is_path_blocked(Path::new(from, to)) + } + + /// Enqueue one payload from `from` to `to` into the simulated network. + pub fn send(&self, from: NodeId, to: NodeId, payload: Vec) -> Result<(), SendError> { + let now = self.time.now(); + self.inner.lock().send(Path::new(from, to), now, payload, &self.rng) + } + + /// Drain simulated network deliveries ready at `now`. + pub(crate) fn tick(&self, now: Duration) -> bool { + let Some(wakers) = self.inner.lock().tick(now, &self.rng) else { + return false; + }; + for waker in wakers { + waker.wake(); + } + true + } + + pub(crate) fn next_delivery_deadline(&self, now: Duration) -> Option { + self.inner.lock().next_delivery_deadline(now) + } + + /// Receive one payload addressed to `node`. + pub fn recv(&self, node: NodeId) -> Recv { + self.register_node(node); + Recv { + node, + network: self.clone(), + } + } +} + +/// Per-node network handle. +#[derive(Clone, Debug)] +pub struct NodeNetwork { + node: NodeId, + network: Network, +} + +impl NodeNetwork { + pub fn node(&self) -> NodeId { + self.node + } + + pub fn send(&self, to: NodeId, payload: Vec) -> Result<(), SendError> { + self.network.send(self.node, to, payload) + } + + pub fn recv(&self) -> Recv { + self.network.recv(self.node) + } + + pub fn network(&self) -> &Network { + &self.network + } +} + +/// One delivered simulated network payload. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct Packet { + pub from: NodeId, + pub payload: Vec, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +struct PacketEvent { + path: Path, + deliver_at: Duration, + payload: Vec, +} + +/// Error returned when the simulated network refuses a send submission. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum SendError { + UnknownNode { node: NodeId }, + PathAtCapacity { from: NodeId, to: NodeId, capacity: usize }, +} + +impl fmt::Display for SendError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::UnknownNode { node } => write!(f, "simulated network node {node} is not registered"), + Self::PathAtCapacity { from, to, capacity } => { + write!(f, "simulated network path {from}->{to} reached capacity {capacity}") + } + } + } +} + +#[derive(Debug)] +struct NetworkState { + options: Options, + nodes: BTreeMap, + /// Packets accepted by the network but not delivered yet. + /// + /// Delayed and blocked packets both stay here so path capacity accounting + /// and random delivery order use a single source of truth. + in_flight: VecDeque, + /// Node-level manual faults. These have no local deadline. + clogged_node_in: BTreeSet, + clogged_node_out: BTreeSet, + /// Executor-owned node isolation for paused or crashed processes. + isolated_nodes: BTreeSet, + /// Link-level manual faults. These have no local deadline. + clogged_links: BTreeSet, + /// Link faults created by the automatic partition state machine. + automatic_partition_links: BTreeSet, + /// Temporary directed-path clogs keyed by their expiry time. + path_clogs: BTreeMap, + partition_stability_ticks_remaining: u32, + unpartition_stability_ticks_remaining: u32, +} + +impl NetworkState { + fn new(options: Options) -> Self { + Self { + options, + nodes: BTreeMap::new(), + in_flight: VecDeque::new(), + clogged_node_in: BTreeSet::new(), + clogged_node_out: BTreeSet::new(), + isolated_nodes: BTreeSet::new(), + clogged_links: BTreeSet::new(), + automatic_partition_links: BTreeSet::new(), + path_clogs: BTreeMap::new(), + partition_stability_ticks_remaining: 0, + unpartition_stability_ticks_remaining: 0, + } + } + + /// Clear manually configured node-level network faults. + fn clear_node_faults(&mut self) { + self.clogged_node_in.clear(); + self.clogged_node_out.clear(); + } + + /// Clear faults owned by the network path and partition machinery. + fn clear_link_faults(&mut self) { + self.clogged_links.clear(); + self.automatic_partition_links.clear(); + self.path_clogs.clear(); + self.partition_stability_ticks_remaining = 0; + self.unpartition_stability_ticks_remaining = 0; + } + + /// Return whether any node-level fault blocks inbound or outbound traffic. + fn is_node_clogged(&self, node: NodeId) -> bool { + self.clogged_node_in.contains(&node) + || self.clogged_node_out.contains(&node) + || self.isolated_nodes.contains(&node) + } + + /// Accept one outbound packet into the simulated network. + /// + /// Capacity is checked before loss/replay so a dropped packet still + /// observes the same backpressure as a real send attempt. Accepted packets + /// stay in `in_flight` until `tick` moves them to an inbox. + fn send(&mut self, path: Path, now: Duration, payload: Vec, rng: &Rng) -> Result<(), SendError> { + self.expire_path_clogs(now); + + for node in [path.from, path.to] { + if !self.nodes.contains_key(&node) { + return Err(SendError::UnknownNode { node }); + } + } + + let options = self.options; + + let path_capacity = options.path_maximum_capacity; + if self.path_in_flight(path) >= path_capacity { + return Err(SendError::PathAtCapacity { + from: path.from, + to: path.to, + capacity: path_capacity, + }); + } + + if rng.buggify_ratio(options.packet_loss_probability) { + return Ok(()); + } + + if rng.buggify_ratio(options.path_clog_probability) { + self.clog_path(path, now, rng); + } + + let deliver_at = now.saturating_add(sample_duration_between( + rng, + options.one_way_delay_min, + options.one_way_delay_mean, + )); + + if rng.buggify_ratio(options.packet_replay_probability) + && self.path_in_flight(path).saturating_add(1) < path_capacity + { + self.in_flight.push_back(PacketEvent { + path, + deliver_at, + payload: payload.clone(), + }); + } + self.in_flight.push_back(PacketEvent { + path, + deliver_at, + payload, + }); + Ok(()) + } + + /// Run one network tick at the current virtual time. + /// + /// A tick delivers packets already ready at this instant, advances partition + /// state, then drains again for packets unblocked by a heal. This prevents a + /// newly created partition from retroactively blocking packets whose + /// `deliver_at` is already due. + fn tick(&mut self, now: Duration, rng: &Rng) -> Option> { + self.expire_path_clogs(now); + + let mut wakers = Vec::new(); + let mut delivered = self.drain_deliverable_packets(now, rng, &mut wakers); + self.maybe_update_partition(rng); + delivered |= self.drain_deliverable_packets(now, rng, &mut wakers); + + delivered.then_some(wakers) + } + + /// Move every currently deliverable packet to its destination inbox. + /// + /// Delivery order remains randomized, but choosing each packet uses + /// reservoir sampling over the ready subset instead of allocating a list. + fn drain_deliverable_packets(&mut self, now: Duration, rng: &Rng, wakers: &mut Vec) -> bool { + let mut delivered = false; + while let Some(index) = self.deliverable_packet_index(rng, now) { + delivered = true; + let event = self.in_flight.remove(index).expect("index came from in-flight queue"); + let inbox = self.nodes.entry(event.path.to).or_default(); + inbox.messages.push_back(Packet { + from: event.path.from, + payload: event.payload, + }); + if let Some(waker) = inbox.waker.take() { + wakers.push(waker); + } + } + delivered + } + + /// Count all packets occupying capacity on one directed path. + fn path_in_flight(&self, path: Path) -> usize { + self.in_flight.iter().filter(|event| event.path == path).count() + } + + /// Add a temporary clog whose lifetime is driven by virtual time. + fn clog_path(&mut self, path: Path, now: Duration, rng: &Rng) { + let duration = sample_duration_between(rng, Duration::ZERO, self.options.path_clog_duration_mean); + if duration.is_zero() { + return; + } + let until = now.saturating_add(duration); + self.path_clogs + .entry(path) + .and_modify(|existing| *existing = (*existing).max(until)) + .or_insert(until); + } + + /// Drop expired temporary path clogs before delivery or deadline checks. + fn expire_path_clogs(&mut self, now: Duration) { + self.path_clogs.retain(|_, until| *until > now); + } + + /// Return whether any current network fault blocks a directed path. + fn is_path_blocked(&self, path: Path) -> bool { + self.is_path_blocked_until_external_change(path) || self.path_clogs.contains_key(&path) + } + + /// Return whether a path is blocked by a fault with no packet-local expiry. + fn is_path_blocked_until_external_change(&self, path: Path) -> bool { + self.clogged_node_out.contains(&path.from) + || self.clogged_node_in.contains(&path.to) + || self.isolated_nodes.contains(&path.from) + || self.isolated_nodes.contains(&path.to) + || self.clogged_links.contains(&path) + || self.automatic_partition_links.contains(&path) + } + + /// Choose one ready packet uniformly from the ready subset. + fn deliverable_packet_index(&self, rng: &Rng, now: Duration) -> Option { + let mut selected = None; + let mut ready = 0; + for (index, event) in self.in_flight.iter().enumerate() { + if event.deliver_at > now || self.is_path_blocked(event.path) { + continue; + } + ready += 1; + if rng.index(ready) == 0 { + selected = Some(index); + } + } + selected + } + + /// Return the next packet/path-clog deadline at which delivery may make progress. + fn next_delivery_deadline(&mut self, now: Duration) -> Option { + self.expire_path_clogs(now); + self.in_flight + .iter() + .filter_map(|event| self.packet_delivery_deadline(event, now)) + .min() + } + + /// Compute the earliest time this packet can become deliverable by itself. + fn packet_delivery_deadline(&self, event: &PacketEvent, now: Duration) -> Option { + // Manual node/link clogs and active partitions have no packet-local + // deadline; only an external mutation or a later network tick can unblock them. + if self.is_path_blocked_until_external_change(event.path) { + return None; + } + + let deadline = self + .path_clogs + .get(&event.path) + .copied() + .unwrap_or(event.deliver_at) + .max(event.deliver_at); + (deadline > now).then_some(deadline) + } + + /// Advance the automatic partition state machine during a network tick. + fn maybe_update_partition(&mut self, rng: &Rng) { + if !rng.is_buggify_enabled() || self.nodes.len() < 2 { + return; + } + + if self.partition_stability_ticks_remaining > 0 { + self.partition_stability_ticks_remaining -= 1; + return; + } + + if !self.automatic_partition_links.is_empty() { + if rng.buggify_ratio(self.options.unpartition_probability) { + self.automatic_partition_links.clear(); + self.unpartition_stability_ticks_remaining = self.options.unpartition_stability_ticks; + } + return; + } + + if self.unpartition_stability_ticks_remaining > 0 { + self.unpartition_stability_ticks_remaining -= 1; + return; + } + + if rng.buggify_ratio(self.options.partition_probability) { + self.create_partition(rng); + self.partition_stability_ticks_remaining = self.options.partition_stability_ticks; + } + } + + /// Build a deterministic random partition over the registered nodes. + fn create_partition(&mut self, rng: &Rng) { + let nodes = self.nodes.keys().copied().collect::>(); + if nodes.len() < 2 { + return; + } + let left = random_partition_side(&nodes, rng); + let one_way_left_to_right = self + .options + .one_way_partition_probability + .sample(rng) + .then_some(rng.index(2) == 0); + + for &from in &nodes { + for &to in &nodes { + let path = Path::new(from, to); + if !path.crosses_partition(&left) { + continue; + } + + let blocked = match one_way_left_to_right { + None => true, + Some(left_to_right) => left.contains(&path.from) == left_to_right, + }; + if blocked { + self.automatic_partition_links.insert(path); + } + } + } + } + + /// Pop one queued packet or remember the latest receiver waker. + fn recv(&mut self, node: NodeId, waker: &Waker) -> Poll { + let inbox = self.nodes.entry(node).or_default(); + if let Some(packet) = inbox.messages.pop_front() { + Poll::Ready(packet) + } else { + inbox.waker = Some(waker.clone()); + Poll::Pending + } + } +} + +/// Pick a non-empty, non-total random side in deterministic node order. +fn random_partition_side(nodes: &[NodeId], rng: &Rng) -> BTreeSet { + assert!(nodes.len() >= 2, "partition requires at least two nodes"); + + let mut shuffled = nodes.to_vec(); + let side_len = 1 + rng.index(nodes.len() - 1); + for index in 0..side_len { + let swap_with = index + rng.index(shuffled.len() - index); + shuffled.swap(index, swap_with); + } + shuffled.into_iter().take(side_len).collect() +} + +#[derive(Debug, Default)] +struct Inbox { + messages: VecDeque, + waker: Option, +} + +pub struct Recv { + node: NodeId, + network: Network, +} + +impl Future for Recv { + type Output = Packet; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + self.network.inner.lock().recv(self.node, cx.waker()) + } +} + +#[cfg(test)] +mod tests { + use alloc::vec; + use core::time::Duration; + + use super::super::{Node, Runtime, RuntimeConfig}; + use super::*; + + fn node_net(node: &Node) -> NodeNetwork { + node.net().expect("test runtime should have a simulated network") + } + + #[test] + fn clogged_link_blocks_delivery_until_unclogged() { + let mut runtime = Runtime::with_config(RuntimeConfig::new(0).with_network(Options::default())); + let handle = runtime.handle(); + let a = handle.create_node().build(); + let b = handle.create_node().build(); + let net = handle.network().expect("test runtime should have a simulated network"); + let a_net = net.on_node(a.id()); + let b_net = net.on_node(b.id()); + + runtime.block_on(async { + net.clog_link(a.id(), b.id()); + a_net.send(b.id(), vec![1]).unwrap(); + + net.unclog_link(a.id(), b.id()); + let packet = b_net.recv().await; + assert_eq!(packet.from, a.id()); + assert_eq!(packet.payload, vec![1]); + }); + } + + #[test] + fn path_capacity_refuses_extra_packet() { + let options = Options { + path_maximum_capacity: 1, + ..Options::default() + }; + let mut runtime = Runtime::with_config(RuntimeConfig::new(0).with_network(options)); + let handle = runtime.handle(); + let a = handle.create_node().build(); + let b = handle.create_node().build(); + let a_net = node_net(&a); + + a_net.send(b.id(), vec![1]).unwrap(); + assert!(matches!( + a_net.send(b.id(), vec![2]), + Err(SendError::PathAtCapacity { .. }) + )); + + runtime.block_on(async { + let packet = node_net(&b).recv().await; + assert_eq!(packet.payload, vec![1]); + }); + } + + #[test] + fn path_clog_never_shortens_existing_clog() { + let options = Options { + path_clog_duration_mean: Duration::from_nanos(1), + ..Options::default() + }; + let runtime = Runtime::new(0); + let handle = runtime.handle(); + let a = handle.create_node().build(); + let b = handle.create_node().build(); + let path = Path::new(a.id(), b.id()); + let existing_until = Duration::from_secs(10); + let mut state = NetworkState::new(options); + + state.path_clogs.insert(path, existing_until); + state.clog_path(path, Duration::ZERO, &Rng::new(0)); + + assert_eq!(state.path_clogs.get(&path).copied(), Some(existing_until)); + } + + #[test] + fn manual_node_clog_survives_pause_resume() { + let runtime = Runtime::with_config(RuntimeConfig::new(0).with_network(Options::default())); + let handle = runtime.handle(); + let a = handle.create_node().build(); + let b = handle.create_node().build(); + let net = handle.network().expect("test runtime should have a simulated network"); + + net.clog_node(b.id()); + b.pause(); + b.resume(); + + assert!(net.is_blocked(a.id(), b.id())); + net.unclog_node(b.id()); + assert!(!net.is_blocked(a.id(), b.id())); + } + + #[test] + fn delayed_packet_advances_virtual_time() { + let options = Options { + one_way_delay_min: Duration::from_millis(5), + one_way_delay_mean: Duration::from_millis(5), + ..Options::default() + }; + let mut runtime = Runtime::with_config(RuntimeConfig::new(0).with_network(options)); + let handle = runtime.handle(); + let a = handle.create_node().build(); + let b = handle.create_node().build(); + let a_net = node_net(&a); + let b_net = node_net(&b); + + runtime.block_on(async { + a_net.send(b.id(), vec![1]).unwrap(); + let packet = b_net.recv().await; + assert_eq!(packet.payload, vec![1]); + assert!(handle.now() >= Duration::from_millis(5)); + }); + } + + #[test] + fn same_deadline_packets_are_delivered_before_receiver_runs() { + let options = Options { + one_way_delay_min: Duration::from_millis(5), + one_way_delay_mean: Duration::from_millis(5), + ..Options::default() + }; + let mut runtime = Runtime::with_config(RuntimeConfig::new(0).with_network(options)); + let handle = runtime.handle(); + let a = handle.create_node().build(); + let b = handle.create_node().build(); + let net = handle.network().expect("test runtime should have a simulated network"); + let a_net = net.on_node(a.id()); + let b_net = net.on_node(b.id()); + + runtime.block_on(async { + a_net.send(b.id(), vec![1]).unwrap(); + a_net.send(b.id(), vec![2]).unwrap(); + + let first = b_net.recv().await; + { + let state = net.inner.lock(); + assert!(state.in_flight.is_empty()); + assert_eq!(state.nodes.get(&b.id()).unwrap().messages.len(), 1); + } + + let second = b_net.recv().await; + let mut payloads = vec![first.payload, second.payload]; + payloads.sort(); + assert_eq!(payloads, vec![vec![1], vec![2]]); + }); + } + + #[test] + fn zero_delay_packet_sent_by_woken_task_is_delivered_without_deadlock() { + let options = Options { + one_way_delay_min: Duration::ZERO, + one_way_delay_mean: Duration::ZERO, + ..Options::default() + }; + let mut runtime = Runtime::with_config(RuntimeConfig::new(0).with_network(options)); + let handle = runtime.handle(); + let a = handle.create_node().build(); + let b = handle.create_node().build(); + let a_net = node_net(&a); + let b_net = node_net(&b); + + runtime.block_on(async { + a_net.send(b.id(), vec![1]).unwrap(); + assert_eq!(b_net.recv().await.payload, vec![1]); + + b_net.send(a.id(), vec![2]).unwrap(); + assert_eq!(a_net.recv().await.payload, vec![2]); + }); + } + + #[test] + fn packet_loss_drops_packet() { + let options = Options { + packet_loss_probability: Ratio::new(1, 1), + ..Options::default() + }; + let runtime = Runtime::with_config(RuntimeConfig::new(0).with_network(options)); + runtime.enable_buggify(); + let handle = runtime.handle(); + let a = handle.create_node().build(); + let b = handle.create_node().build(); + let a_net = node_net(&a); + + a_net.send(b.id(), vec![1]).unwrap(); + assert!(handle + .network() + .expect("test runtime should have a simulated network") + .inner + .lock() + .in_flight + .is_empty()); + } + + #[test] + fn buggify_disabled_does_not_drop_packet() { + let options = Options { + packet_loss_probability: Ratio::new(1, 1), + ..Options::default() + }; + let mut runtime = Runtime::with_config(RuntimeConfig::new(0).with_network(options)); + let handle = runtime.handle(); + let a = handle.create_node().build(); + let b = handle.create_node().build(); + let a_net = node_net(&a); + let b_net = node_net(&b); + + a_net.send(b.id(), vec![1]).unwrap(); + + runtime.block_on(async { + let packet = b_net.recv().await; + assert_eq!(packet.payload, vec![1]); + }); + } + + #[test] + fn automatic_partition_heals_blocked_packet_on_later_timer_tick() { + let options = Options { + one_way_delay_min: Duration::ZERO, + one_way_delay_mean: Duration::ZERO, + partition_probability: Ratio::new(1, 1), + unpartition_probability: Ratio::new(1, 1), + ..Options::default() + }; + let mut runtime = Runtime::with_config(RuntimeConfig::new(0).with_network(options)); + runtime.enable_buggify(); + let handle = runtime.handle(); + let a = handle.create_node().build(); + let b = handle.create_node().build(); + let a_net = node_net(&a); + let b_net = node_net(&b); + let timer = handle.clone(); + let _timer = a.spawn(async move { + timer.sleep(Duration::from_millis(1)).await; + }); + + runtime.block_on(async { + a_net.send(b.id(), vec![1]).unwrap(); + assert_eq!(b_net.recv().await.payload, vec![1]); + + a_net.send(b.id(), vec![2]).unwrap(); + let packet = b_net.recv().await; + assert_eq!(packet.payload, vec![2]); + }); + } +} diff --git a/crates/runtime-core/src/sim/probability.rs b/crates/runtime-core/src/sim/probability.rs new file mode 100644 index 00000000000..7f587bdee3b --- /dev/null +++ b/crates/runtime-core/src/sim/probability.rs @@ -0,0 +1,28 @@ +use core::time::Duration; + +use super::Rng; + +/// Sample a deterministic duration from a bounded uniform range. +/// +/// `mean` is the midpoint of the sampled range, not an exponential or normal +/// distribution parameter. When `mean > min`, the range is: +/// +/// `[min, min + 2 * (mean - min)]` +/// +/// This keeps the arithmetic simple and deterministic while preserving the +/// configured mean for a uniform distribution. If `mean <= min`, there is no +/// range to sample from and the function returns `min`. +pub(crate) fn sample_duration_between(rng: &Rng, min: Duration, mean: Duration) -> Duration { + if mean <= min { + return min; + } + + let min_ns = min.as_nanos(); + let spread_ns = mean.as_nanos().saturating_sub(min_ns).saturating_mul(2); + let spread_ns = spread_ns.min(u128::from(u64::MAX)) as u64; + if spread_ns == 0 { + return min; + } + + min.saturating_add(Duration::from_nanos(rng.next_u64() % (spread_ns + 1))) +} diff --git a/crates/runtime-core/src/sim/rng.rs b/crates/runtime-core/src/sim/rng.rs index 3555953e764..e3f649fef93 100644 --- a/crates/runtime-core/src/sim/rng.rs +++ b/crates/runtime-core/src/sim/rng.rs @@ -89,6 +89,10 @@ impl GlobalRng { (self.next_u64() as usize) % len } + /// Sample a Bernoulli event using one deterministic RNG word. + /// + /// Probabilities less than or equal to zero never fire, and probabilities + /// greater than or equal to one always fire. pub fn sample_probability(&self, probability: f64) -> bool { probability_sample(self.next_u64(), probability) } @@ -113,6 +117,11 @@ impl GlobalRng { self.is_buggify_enabled() && self.sample_probability(probability) } + /// Sample `ratio` only when runtime buggify fault injection is enabled. + pub fn buggify_ratio(&self, ratio: Ratio) -> bool { + self.is_buggify_enabled() && ratio.sample(self) + } + #[allow(dead_code)] pub(crate) fn seed(&self) -> u64 { self.inner.lock().seed @@ -189,6 +198,10 @@ impl GlobalRng { #[derive(Debug, Clone, Eq, PartialEq)] pub struct DeterminismLog(Vec); +/// Convert `value` into a unit interval sample and compare it to `probability`. +/// +/// The top 53 bits are used because that is the precision available in an +/// `f64` mantissa, yielding a deterministic value in `[0.0, 1.0)`. fn probability_sample(value: u64, probability: f64) -> bool { if probability <= 0.0 { return false; @@ -204,3 +217,32 @@ fn probability_sample(value: u64, probability: f64) -> bool { fn checksum(value: u64) -> u8 { value.to_ne_bytes().into_iter().fold(0, |acc, byte| acc ^ byte) } + +/// Probability represented as an exact rational number. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct Ratio { + /// Probability numerator. + pub numerator: u64, + /// Probability denominator; must be non-zero when sampled. + pub denominator: u64, +} + +impl Ratio { + pub const ZERO: Self = Self { + numerator: 0, + denominator: 1, + }; + + pub const fn new(numerator: u64, denominator: u64) -> Self { + Self { numerator, denominator } + } + + pub fn is_zero(self) -> bool { + self.numerator == 0 + } + + pub fn sample(self, rng: &Rng) -> bool { + assert!(self.denominator > 0, "ratio denominator must be non-zero"); + self.numerator > 0 && rng.next_u64() % self.denominator < self.numerator.min(self.denominator) + } +} diff --git a/crates/runtime-core/src/sim/time/mod.rs b/crates/runtime-core/src/sim/time/mod.rs index 242ff0ab84a..7af1ab3bb70 100644 --- a/crates/runtime-core/src/sim/time/mod.rs +++ b/crates/runtime-core/src/sim/time/mod.rs @@ -81,6 +81,34 @@ impl TimeHandle { woke } + pub(crate) fn next_timer_deadline(&self) -> Option { + self.inner.lock().timers.values().map(|timer| timer.deadline).min() + } + + /// Wake timers whose deadlines are already at or before the current time. + pub(crate) fn wake_due_timers(&self) -> bool { + let wakers = { + let mut state = self.inner.lock(); + state.take_due_wakers() + }; + let woke = !wakers.is_empty(); + wake_all(wakers); + woke + } + + pub(crate) fn advance_to(&self, deadline: Duration) -> bool { + let wakers = { + let mut state = self.inner.lock(); + if deadline <= state.now { + return false; + } + state.now = deadline; + state.take_due_wakers() + }; + wake_all(wakers); + true + } + /// Register or refresh a timer entry for a sleeping future. /// /// Sleep futures keep a stable `TimerId` across polls. Re-registering with diff --git a/crates/runtime/Cargo.toml b/crates/runtime/Cargo.toml index c8affea0f48..577a59c99e3 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", "spacetimedb-runtime-core/sim", "dep:libc"] +simulation = ["dep:spacetimedb-runtime-core", "spacetimedb-runtime-core/sim", "spacetimedb-runtime-core/std", "dep:libc"] diff --git a/crates/runtime/src/lib.rs b/crates/runtime/src/lib.rs index c6192e1b738..8f876aca6e3 100644 --- a/crates/runtime/src/lib.rs +++ b/crates/runtime/src/lib.rs @@ -70,7 +70,7 @@ enum JoinHandleInner { // // This happens in two cases: // - // 1. After the task output has been yielded — the backend handle no longer + // 1. After the task output has been yielded -- the backend handle no longer // owns `T`, so we swap it out for a neutral placeholder rather than // leave a semantically-invalid variant in place. // 2. In `Drop`, so we can call `detach()` on the simulation handle (which