Extend apis of runtime crate.

crates/dst/Cargo.toml

@@ -1,9 +1,13 @@
 [package]
-name = "spacetimedb-dst"
+name = "spacetimedb-dst-lib"
 version.workspace = true
 edition.workspace = true
 rust-version.workspace = true
 
+[features]
+default = ["fallocate"]
+fallocate = ["spacetimedb-commitlog/fallocate"]
+
 [dependencies]
 anyhow.workspace = true
 clap.workspace = true

crates/dst/src/engine.rs

@@ -274,10 +274,11 @@ impl EngineTarget {
 impl TargetDriver<Interaction> for EngineTarget {
     type Observation = Observation;
 
-    fn execute(&mut self, interaction: &Interaction) -> Result<Self::Observation, anyhow::Error> {
+    async fn execute<'a>(&'a mut self, interaction: &'a Interaction) -> Result<Self::Observation, anyhow::Error> {
         EngineTarget::execute(self, interaction)
     }
 }
+
 pub struct EngineTest;
 
 impl TestSuite for EngineTest {
@@ -289,7 +290,7 @@ impl TestSuite for EngineTest {
 
     type Properties = EngineProperties;
 
-    fn build(&self, rng: Rng) -> Result<(Self::Interactions, Self::Target, Self::Properties), anyhow::Error> {
+    async fn build(&self, rng: Rng) -> Result<(Self::Interactions, Self::Target, Self::Properties), anyhow::Error> {
         let schema = default_schema(rng.clone());
         let runtime_seed = rng.next_u64();
         let target = EngineTarget::init(schema.clone(), runtime_seed)?;

crates/dst/src/lib.rs

@@ -0,0 +1,6 @@
+pub mod engine;
+pub mod schema;
+pub mod sim;
+pub mod traits;
+
+pub use traits::{Properties, TargetDriver, TestSuite, TestSuiteParts};

crates/dst/src/sim/commitlog.rs

@@ -22,6 +22,12 @@ pub struct InMemoryCommitlog {
     options: Options,
 }
 
+impl Default for InMemoryCommitlog {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 impl InMemoryCommitlog {
     pub fn new() -> Self {
         Self {
@@ -394,6 +400,10 @@ impl FileLike for Segment {
 
         Ok(())
     }
+
+    fn fallocate(&mut self, _size: u64) -> io::Result<()> {
+        Ok(())
+    }
 }
 
 pub struct ReadOnlySegment {

crates/dst/src/traits.rs

@@ -5,7 +5,10 @@ use spacetimedb_runtime::sim::Rng;
 pub trait TargetDriver<I> {
     type Observation;
 
-    fn execute(&mut self, interaction: &I) -> Result<Self::Observation, Error>;
+    fn execute<'a>(
+        &'a mut self,
+        interaction: &'a I,
+    ) -> impl std::future::Future<Output = Result<Self::Observation, Error>> + 'a;
 }
 
 /// Ensures if Output of `TargetDrive` is expected for the input
@@ -20,32 +23,35 @@ pub type TestSuiteParts<S> = (
 );
 
 pub trait TestSuite {
-    type Interaction;
+    type Interaction: std::fmt::Debug;
     type Interactions: Iterator<Item = Self::Interaction> + std::fmt::Debug;
     type Target: TargetDriver<Self::Interaction>;
     type Properties: Properties<Self::Interaction, <Self::Target as TargetDriver<Self::Interaction>>::Observation>;
 
-    fn build(&self, rng: Rng) -> Result<TestSuiteParts<Self>, Error>
+    fn build(&self, rng: Rng) -> impl std::future::Future<Output = Result<TestSuiteParts<Self>, Error>> + '_
     where
         Self: Sized;
 
-    fn run(&self, rng: Rng, max_interactions: Option<usize>) -> Result<(), Error>
+    fn run(&self, rng: Rng, max_interactions: usize) -> impl std::future::Future<Output = Result<(), Error>> + '_
     where
         Self: Sized,
     {
-        let (mut interactions, mut target, mut properties) = self.build(rng)?;
+        async move {
+            let (mut interactions, mut target, mut properties) = self.build(rng).await?;
 
-        let result = (|| {
-            for interaction in interactions.by_ref().take(max_interactions.unwrap_or(usize::MAX)) {
-                let observation = target.execute(&interaction)?;
-                properties.observe(&interaction, &observation)?;
-            }
+            let result = async {
+                for interaction in interactions.by_ref().take(max_interactions) {
+                    let observation = target.execute(&interaction).await?;
+                    properties.observe(&interaction, &observation)?;
+                }
 
-            Ok(())
-        })();
+                Ok(())
+            }
+            .await;
 
-        tracing::info!(interaction_counts = ?interactions, "final interaction counts");
+            tracing::info!(interaction_counts = ?interactions, "final interaction counts");
 
-        result
+            result
+        }
     }
 }

crates/runtime/src/lib.rs

@@ -49,13 +49,32 @@ pub mod sync {
     pub use tokio::sync::watch;
 }
 
-#[derive(Clone)]
 pub enum Handle {
     Tokio(TokioHandle),
     #[cfg(feature = "simulation")]
     Simulation(sim::Handle),
 }
 
+impl Clone for Handle {
+    fn clone(&self) -> Self {
+        match self {
+            Self::Tokio(handle) => Self::Tokio(handle.clone()),
+            #[cfg(feature = "simulation")]
+            Self::Simulation(handle) => Self::Simulation(handle.clone()),
+        }
+    }
+}
+
+impl fmt::Debug for Handle {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Tokio(_) => f.write_str("Handle::Tokio"),
+            #[cfg(feature = "simulation")]
+            Self::Simulation(_) => f.write_str("Handle::Simulation"),
+        }
+    }
+}
+
 pub struct JoinHandle<T> {
     inner: JoinHandleInner<T>,
 }
@@ -135,6 +154,16 @@ impl fmt::Display for JoinError {
     }
 }
 
+impl JoinError {
+    pub fn is_panic(&self) -> bool {
+        match &self.inner {
+            JoinErrorInner::Tokio(err) => err.is_panic(),
+            #[cfg(feature = "simulation")]
+            JoinErrorInner::Simulation(err) => err.is_panic(),
+        }
+    }
+}
+
 impl std::error::Error for JoinError {}
 
 impl<T> JoinHandleInner<T> {
@@ -231,9 +260,37 @@ impl Handle {
 
 #[cfg(feature = "simulation")]
 impl Handle {
+    pub fn now(&self) -> Duration {
+        match self {
+            Self::Tokio(_) => panic!("Handle::now requires a simulation runtime"),
+            Self::Simulation(handle) => handle.now(),
+        }
+    }
+
     pub fn simulation(handle: sim::Handle) -> Self {
         Self::Simulation(handle)
     }
+
+    pub fn on_simulation_node(&self, node: sim::NodeId) -> Self {
+        match self {
+            Self::Tokio(_) => panic!("Handle::on_simulation_node requires a simulation runtime"),
+            Self::Simulation(handle) => Self::Simulation(handle.on_node(node)),
+        }
+    }
+
+    pub fn create_simulation_node(&self) -> Option<sim::NodeBuilder> {
+        match self {
+            Self::Tokio(_) => panic!("Handle::create_simulation_node requires a simulation runtime"),
+            Self::Simulation(handle) => Some(handle.create_node()),
+        }
+    }
+
+    pub fn drain_simulation_task_panics(&self) -> Vec<sim::TaskPanic> {
+        match self {
+            Self::Tokio(_) => panic!("Handle::drain_simulation_task_panics requires a simulation runtime"),
+            Self::Simulation(handle) => handle.drain_task_panics(),
+        }
+    }
 }
 
 impl Handle {
@@ -244,7 +301,7 @@ impl Handle {
             },
             #[cfg(feature = "simulation")]
             Self::Simulation(handle) => JoinHandle {
-                inner: JoinHandleInner::Simulation(handle.spawn_on(sim::NodeId::MAIN, future)),
+                inner: JoinHandleInner::Simulation(handle.spawn(future)),
             },
         }
     }
@@ -268,7 +325,7 @@ impl Handle {
             // the simulation backend.
             #[cfg(feature = "simulation")]
             Self::Simulation(handle) => handle
-                .spawn_on(sim::NodeId::MAIN, async move { f() })
+                .spawn(async move { f() })
                 .await
                 .expect("simulation spawn_blocking task should not be cancelled"),
         }
@@ -295,6 +352,29 @@ impl Handle {
             Self::Simulation(handle) => handle.sleep(duration).await,
         }
     }
+
+    pub async fn yield_now(&self) {
+        match self {
+            Self::Tokio(_) => tokio::task::yield_now().await,
+            #[cfg(feature = "simulation")]
+            Self::Simulation(handle) => handle.yield_now().await,
+        }
+    }
+
+    /// Blocks the current thread until `future` completes.
+    ///
+    /// For Tokio, this uses `block_in_place` before re-entering the runtime with
+    /// `Handle::block_on`, so it is supported from multi-thread runtime workers.
+    /// Do not call this from a current-thread Tokio runtime; Tokio panics on
+    /// that path. For simulation, this runs on the deterministic executor
+    /// thread; blocking inside the future blocks simulated progress.
+    pub fn block_on<F: Future>(&self, future: F) -> F::Output {
+        match self {
+            Self::Tokio(handle) => tokio::task::block_in_place(|| handle.block_on(future)),
+            #[cfg(feature = "simulation")]
+            Self::Simulation(handle) => handle.block_on(future),
+        }
+    }
 }
 
 #[cfg(test)]
@@ -323,10 +403,7 @@ mod tests {
             drop(jh);
 
             // Yield so the spawned task gets polled.
-            handle
-                .timeout(std::time::Duration::from_millis(50), async {})
-                .await
-                .ok();
+            handle.yield_now().await;
         });
 
         assert!(flag.load(Ordering::Acquire));