Deterministic runtime crate (#5016)

# Description of Changes.

Introduces deterministic runtime crate.
Integrate it with RelationalDB.

I think best steps to review:
- Read the
[README](https://github.com/clockworklabs/SpacetimeDB/blob/shub/sim/crates/runtime/README.md)
of runtime crate.
- Look at the integration with existing crates - `durability`, `core`,
`snapshot`, etc.
- Read runtime crate's code.

Draft branch to Test code -
#5019

# API and ABI breaking changes
NA

# Expected complexity level and risk
Does not intend to change any production functionality, but it's big
code.

# Testing

- new crate contains unit and integration tests.
- Existing tests should work for production.

---------

Signed-off-by: Shubham Mishra <shivam828787@gmail.com>
Co-authored-by: Zeke Foppa <196249+bfops@users.noreply.github.com>

Author: Shubham8287

Cargo.lock

@@ -25,6 +25,7 @@ members = [
   "crates/physical-plan",
   "crates/primitives",
   "crates/query",
+  "crates/runtime",
   "crates/sats",
   "crates/schema",
   "crates/smoketests",
@@ -149,6 +150,7 @@ spacetimedb-fs-utils = { path = "crates/fs-utils", version = "=2.4.1" }
 spacetimedb-snapshot = { path = "crates/snapshot", version = "=2.4.1" }
 spacetimedb-subscription = { path = "crates/subscription", version = "=2.4.1" }
 spacetimedb-query-builder = { path = "crates/query-builder", version = "=2.4.1" }
+spacetimedb-runtime = { path = "crates/runtime", version = "=2.4.1" }
 
 # Prevent `ahash` from pulling in `getrandom` by disabling default features.
 # Modules use `getrandom02` and we need to prevent an incompatible version

Cargo.toml

@@ -28,6 +28,7 @@ spacetimedb-primitives.workspace = true
 spacetimedb-paths.workspace = true
 spacetimedb-physical-plan.workspace = true
 spacetimedb-query.workspace = true
+spacetimedb-runtime = { workspace = true, features = ["tokio"] }
 spacetimedb-sats = { workspace = true, features = ["serde"] }
 spacetimedb-schema.workspace = true
 spacetimedb-table.workspace = true

crates/core/Cargo.toml

@@ -9,9 +9,9 @@ use spacetimedb_datastore::{execution_context::ReducerContext, traits::TxData};
 use spacetimedb_durability::Transaction;
 use spacetimedb_lib::Identity;
 use spacetimedb_sats::ProductValue;
-use tokio::{runtime, time::timeout};
 
 use crate::db::persistence::Durability;
+use spacetimedb_runtime::Handle;
 
 pub(super) fn request_durability(
     durability: &Durability,
@@ -32,11 +32,11 @@ pub(super) fn request_durability(
     }));
 }
 
-pub(super) fn spawn_close(durability: Arc<Durability>, runtime: &runtime::Handle, database_identity: Identity) {
-    let rt = runtime.clone();
-    rt.spawn(async move {
-        let label = format!("[{database_identity}]");
-        match timeout(Duration::from_secs(10), durability.close()).await {
+pub(super) fn spawn_close(durability: Arc<Durability>, runtime: &Handle, database_identity: Identity) {
+    let label = format!("[{database_identity}]");
+    let runtime = runtime.clone();
+    runtime.clone().spawn(async move {
+        match runtime.timeout(Duration::from_secs(10), durability.close()).await {
             Err(_elapsed) => {
                 error!("{label} timeout waiting for durability shutdown");
             }

crates/core/src/db/durability.rs

@@ -11,6 +11,7 @@ use spacetimedb_paths::server::ServerDataDir;
 use spacetimedb_snapshot::DynSnapshotRepo;
 
 use crate::{messages::control_db::Database, util::asyncify};
+use spacetimedb_runtime::Handle;
 
 use super::{
     relational_db::{self, Txdata},
@@ -98,8 +99,8 @@ pub struct Persistence {
     /// persistent (as opposed to in-memory) databases. This is enforced by
     /// this type.
     pub snapshots: Option<SnapshotWorker>,
-    /// The tokio runtime onto which durability-related tasks shall be spawned.
-    pub runtime: tokio::runtime::Handle,
+    /// Runtime onto which durability-related tasks shall be spawned.
+    pub runtime: Handle,
 }
 
 impl Persistence {
@@ -109,6 +110,15 @@ impl Persistence {
         disk_size: impl Fn() -> io::Result<SizeOnDisk> + Send + Sync + 'static,
         snapshots: Option<SnapshotWorker>,
         runtime: tokio::runtime::Handle,
+    ) -> Self {
+        Self::new_with_runtime(durability, disk_size, snapshots, Handle::tokio(runtime))
+    }
+
+    pub fn new_with_runtime(
+        durability: impl spacetimedb_durability::Durability<TxData = Txdata> + 'static,
+        disk_size: impl Fn() -> io::Result<SizeOnDisk> + Send + Sync + 'static,
+        snapshots: Option<SnapshotWorker>,
+        runtime: Handle,
     ) -> Self {
         Self {
             durability: Arc::new(durability),
@@ -148,7 +158,7 @@ impl Persistence {
         Option<Arc<Durability>>,
         Option<DiskSizeFn>,
         Option<SnapshotWorker>,
-        Option<tokio::runtime::Handle>,
+        Option<Handle>,
     ) {
         this.map(
             |Self {
@@ -207,14 +217,16 @@ impl PersistenceProvider for LocalPersistenceProvider {
     async fn persistence(&self, database: &Database, replica_id: u64) -> anyhow::Result<Persistence> {
         let replica_dir = self.data_dir.replica(replica_id);
         let snapshot_dir = replica_dir.snapshots();
+        let runtime = Handle::tokio_current();
 
         let database_identity = database.database_identity;
         let snapshot_worker =
             asyncify(move || relational_db::open_snapshot_repo(snapshot_dir, database_identity, replica_id))
                 .await
-                .map(|repo| SnapshotWorker::new(repo, snapshot::Compression::Enabled))?;
+                .map(|repo| SnapshotWorker::new(repo, snapshot::Compression::Enabled, runtime.clone()))?;
         let (durability, disk_size) = relational_db::local_durability_with_options(
             replica_dir,
+            runtime.clone(),
             Some(&snapshot_worker),
             self.durability.into_options(),
         )
@@ -231,7 +243,7 @@ impl PersistenceProvider for LocalPersistenceProvider {
             durability,
             disk_size,
             snapshots: Some(snapshot_worker),
-            runtime: tokio::runtime::Handle::current(),
+            runtime,
         })
     }
 }

crates/core/src/db/persistence.rs

@@ -43,6 +43,7 @@ use spacetimedb_lib::ConnectionId;
 use spacetimedb_lib::Identity;
 use spacetimedb_paths::server::{ReplicaDir, SnapshotsPath};
 use spacetimedb_primitives::*;
+use spacetimedb_runtime::Handle;
 use spacetimedb_sats::memory_usage::MemoryUsage;
 use spacetimedb_sats::raw_identifier::RawIdentifier;
 use spacetimedb_sats::{AlgebraicType, AlgebraicValue, ProductType, ProductValue};
@@ -100,7 +101,7 @@ pub struct RelationalDB {
 
     inner: Locking,
     durability: Option<Arc<Durability>>,
-    durability_runtime: Option<tokio::runtime::Handle>,
+    durability_runtime: Option<Handle>,
     snapshot_worker: Option<SnapshotWorker>,
 
     row_count_fn: RowCountFn,
@@ -1688,9 +1689,10 @@ pub type LocalDurability = Arc<durability::Local<ProductValue>>;
 /// of the commitlog.
 pub async fn local_durability(
     replica_dir: ReplicaDir,
+    runtime: Handle,
     snapshot_worker: Option<&SnapshotWorker>,
 ) -> Result<(LocalDurability, DiskSizeFn), DBError> {
-    local_durability_with_options(replica_dir, snapshot_worker, <_>::default()).await
+    local_durability_with_options(replica_dir, runtime, snapshot_worker, <_>::default()).await
 }
 
 /// Initialize local durability with explicit parameters.
@@ -1701,10 +1703,10 @@ pub async fn local_durability(
 /// of the commitlog.
 pub async fn local_durability_with_options(
     replica_dir: ReplicaDir,
+    runtime: Handle,
     snapshot_worker: Option<&SnapshotWorker>,
     opts: durability::local::Options,
 ) -> Result<(LocalDurability, DiskSizeFn), DBError> {
-    let rt = tokio::runtime::Handle::current();
     let on_new_segment = snapshot_worker.map(|snapshot_worker| {
         let snapshot_worker = snapshot_worker.clone();
         Arc::new(move || {
@@ -1716,7 +1718,7 @@ pub async fn local_durability_with_options(
     let local = asyncify(move || {
         durability::Local::open(
             replica_dir.clone(),
-            rt,
+            runtime,
             opts,
             // Give the durability a handle to request a new snapshot run,
             // which it will send down whenever we rotate commitlog segments.
@@ -1982,19 +1984,22 @@ pub mod tests_utils {
         ) -> Result<(RelationalDB, Arc<durability::Local<ProductValue>>), DBError> {
             let snapshots = want_snapshot_repo
                 .then(|| {
-                    open_snapshot_repo(root.snapshots(), db_identity, replica_id)
-                        .map(|repo| SnapshotWorker::new(repo, snapshot::Compression::Disabled))
+                    open_snapshot_repo(root.snapshots(), db_identity, replica_id).map(|repo| {
+                        SnapshotWorker::new(repo, snapshot::Compression::Disabled, Handle::tokio(rt.clone()))
+                    })
                 })
                 .transpose()?;
 
-            let (local, disk_size_fn) = rt.block_on(local_durability(root.clone(), snapshots.as_ref()))?;
+            let runtime = Handle::tokio(rt.clone());
+            let (local, disk_size_fn) =
+                rt.block_on(local_durability(root.clone(), runtime.clone(), snapshots.as_ref()))?;
             let history = local.as_history();
 
             let persistence = Persistence {
                 durability: local.clone(),
                 disk_size: disk_size_fn,
                 snapshots,
-                runtime: rt,
+                runtime,
             };
 
             let (db, _) = RelationalDB::open(
@@ -2107,17 +2112,20 @@ pub mod tests_utils {
         ) -> Result<(RelationalDB, Arc<durability::Local<ProductValue>>), DBError> {
             let snapshots = want_snapshot_repo
                 .then(|| {
-                    open_snapshot_repo(root.snapshots(), Identity::ZERO, 0)
-                        .map(|repo| SnapshotWorker::new(repo, snapshot::Compression::Disabled))
+                    open_snapshot_repo(root.snapshots(), Identity::ZERO, 0).map(|repo| {
+                        SnapshotWorker::new(repo, snapshot::Compression::Disabled, Handle::tokio(rt.clone()))
+                    })
                 })
                 .transpose()?;
-            let (local, disk_size_fn) = rt.block_on(local_durability(root.clone(), snapshots.as_ref()))?;
+            let runtime = Handle::tokio(rt.clone());
+            let (local, disk_size_fn) =
+                rt.block_on(local_durability(root.clone(), runtime.clone(), snapshots.as_ref()))?;
             let history = local.as_history();
             let persistence = Persistence {
                 durability: local.clone(),
                 disk_size: disk_size_fn,
                 snapshots,
-                runtime: rt,
+                runtime,
             };
             let db = Self::open_db(history, Some(persistence), None, 0)?;