move work of do_durabiliy into a DurabilityWorker

Centril · Centril · commit fe7d98bbd6e7 · 2025-12-11T09:44:32.000+01:00
diff --git a/crates/core/src/db/durability.rs b/crates/core/src/db/durability.rs
@@ -0,0 +1,148 @@
+use crate::db::persistence::Durability;
+use futures::{channel::mpsc, StreamExt};
+use spacetimedb_commitlog::payload::{
+    txdata::{Mutations, Ops},
+    Txdata,
+};
+use spacetimedb_data_structures::map::IntSet;
+use spacetimedb_datastore::{execution_context::ReducerContext, traits::TxData};
+use spacetimedb_durability::DurableOffset;
+use spacetimedb_primitives::TableId;
+use std::sync::Arc;
+
+/// A request to persist a transaction.
+pub struct DurabilityRequest {
+    reducer_context: Option<ReducerContext>,
+    tx_data: Arc<TxData>,
+}
+
+/// Represents a handle to a background task that persists transactions
+/// according to the [`Durability`] policy provided.
+///
+/// This exists to avoid doing some preparatory work
+/// before sending over to the `Durability` layer.
+#[derive(Clone)]
+pub struct DurabilityWorker {
+    request_tx: mpsc::UnboundedSender<DurabilityRequest>,
+    durability: Arc<Durability>,
+}
+
+impl DurabilityWorker {
+    /// Create a new [`DurabilityWorker`] using the given `durability` policy.
+    pub fn new(durability: Arc<Durability>) -> Self {
+        let (request_tx, request_rx) = mpsc::unbounded();
+
+        let actor = DurabilityWorkerActor {
+            request_rx,
+            durability: durability.clone(),
+        };
+        tokio::spawn(actor.run());
+
+        Self { request_tx, durability }
+    }
+
+    /// Request that a transaction be made be made durable.
+    /// That is, if `(tx_data, ctx)` should be appended to the commitlog, do so.
+    ///
+    /// Note that by this stage,
+    /// [`spacetimedb_datastore::locking_tx_datastore::committed_state::tx_consumes_offset`]
+    /// has already decided based on the reducer and operations whether the transaction should be appended;
+    /// this method is responsible only for reading its decision out of the `tx_data`
+    /// and calling `durability.append_tx`.
+    ///
+    /// This method does not block,
+    /// and sends the work to an actor that collects data and calls `durability.append_tx`.
+    ///
+    /// Panics if the durability worker has closed the receive end of its queue(s),
+    /// which is likely due to it having panicked.
+    pub fn request_durability(&self, reducer_context: Option<ReducerContext>, tx_data: &Arc<TxData>) {
+        self.request_tx
+            .unbounded_send(DurabilityRequest {
+                reducer_context,
+                tx_data: tx_data.clone(),
+            })
+            .expect("durability worker panicked");
+    }
+
+    /// Get the [`DurableOffset`] of this database.
+    pub fn durable_tx_offset(&self) -> DurableOffset {
+        self.durability.durable_tx_offset()
+    }
+}
+
+pub struct DurabilityWorkerActor {
+    request_rx: mpsc::UnboundedReceiver<DurabilityRequest>,
+    durability: Arc<Durability>,
+}
+
+impl DurabilityWorkerActor {
+    /// Processes requests to do durability.
+    async fn run(mut self) {
+        while let Some(DurabilityRequest {
+            reducer_context,
+            tx_data,
+        }) = self.request_rx.next().await
+        {
+            Self::do_durability(&*self.durability, reducer_context, &tx_data);
+        }
+    }
+
+    pub fn do_durability(durability: &Durability, reducer_context: Option<ReducerContext>, tx_data: &TxData) {
+        if tx_data.tx_offset().is_none() {
+            let name = reducer_context.as_ref().map(|rcx| &*rcx.name);
+            debug_assert!(
+                !tx_data.has_rows_or_connect_disconnect(name),
+                "tx_data has no rows but has connect/disconnect: `{name:?}`"
+            );
+            return;
+        }
+
+        let is_not_ephemeral_table = |table_id: &TableId| -> bool {
+            tx_data
+                .ephemeral_tables()
+                .map(|etables| !etables.contains(table_id))
+                .unwrap_or(true)
+        };
+
+        let inserts: Box<_> = tx_data
+            .inserts()
+            // Skip ephemeral tables
+            .filter(|(table_id, _)| is_not_ephemeral_table(table_id))
+            .map(|(table_id, rowdata)| Ops {
+                table_id: *table_id,
+                rowdata: rowdata.clone(),
+            })
+            .collect();
+
+        let truncates: IntSet<TableId> = tx_data.truncates().collect();
+
+        let deletes: Box<_> = tx_data
+            .deletes()
+            .filter(|(table_id, _)| is_not_ephemeral_table(table_id))
+            .map(|(table_id, rowdata)| Ops {
+                table_id: *table_id,
+                rowdata: rowdata.clone(),
+            })
+            // filter out deletes for tables that are truncated in the same transaction.
+            .filter(|ops| !truncates.contains(&ops.table_id))
+            .collect();
+
+        let truncates = truncates.into_iter().filter(is_not_ephemeral_table).collect();
+
+        let inputs = reducer_context.map(|rcx| rcx.into());
+
+        let txdata = Txdata {
+            inputs,
+            outputs: None,
+            mutations: Some(Mutations {
+                inserts,
+                deletes,
+                truncates,
+            }),
+        };
+
+        // TODO: Should measure queuing time + actual write
+        // This does not block, as per trait docs.
+        durability.append_tx(txdata);
+    }
+}
diff --git a/crates/core/src/db/mod.rs b/crates/core/src/db/mod.rs
@@ -7,6 +7,7 @@ use crate::subscription::ExecutionCounters;
 use spacetimedb_datastore::execution_context::WorkloadType;
 use spacetimedb_datastore::{locking_tx_datastore::datastore::TxMetrics, traits::TxData};
 
+mod durability;
 pub mod persistence;
 pub mod relational_db;
 pub mod snapshot;
diff --git a/crates/core/src/db/relational_db.rs b/crates/core/src/db/relational_db.rs
@@ -1,3 +1,4 @@
+use crate::db::durability::DurabilityWorker;
 use crate::db::MetricsRecorderQueue;
 use crate::error::{DBError, DatabaseError, RestoreSnapshotError};
 use crate::messages::control_db::HostType;
@@ -10,10 +11,9 @@ use fs2::FileExt;
 use log::info;
 use spacetimedb_commitlog::repo::OnNewSegmentFn;
 use spacetimedb_commitlog::{self as commitlog, SizeOnDisk};
-use spacetimedb_data_structures::map::IntSet;
 use spacetimedb_datastore::db_metrics::DB_METRICS;
 use spacetimedb_datastore::error::{DatastoreError, TableError, ViewError};
-use spacetimedb_datastore::execution_context::{ReducerContext, Workload, WorkloadType};
+use spacetimedb_datastore::execution_context::{Workload, WorkloadType};
 use spacetimedb_datastore::locking_tx_datastore::datastore::TxMetrics;
 use spacetimedb_datastore::locking_tx_datastore::state_view::{
     IterByColEqMutTx, IterByColRangeMutTx, IterMutTx, IterTx, StateView,
@@ -103,7 +103,7 @@ pub struct RelationalDB {
     owner_identity: Identity,
 
     inner: Locking,
-    durability: Option<Arc<Durability>>,
+    durability: Option<DurabilityWorker>,
     snapshot_worker: Option<SnapshotWorker>,
 
     row_count_fn: RowCountFn,
@@ -154,6 +154,7 @@ impl RelationalDB {
             Arc::new(EnumMap::from_fn(|ty| ExecutionCounters::new(&ty, &database_identity)));
 
         let (durability, disk_size_fn, snapshot_worker) = Persistence::unzip(persistence);
+        let durability = durability.map(DurabilityWorker::new);
 
         Self {
             inner,
@@ -766,19 +767,21 @@ impl RelationalDB {
     }
 
     #[tracing::instrument(level = "trace", skip_all)]
-    pub fn commit_tx(&self, tx: MutTx) -> Result<Option<(TxOffset, TxData, TxMetrics, String)>, DBError> {
+    #[allow(clippy::type_complexity)]
+    pub fn commit_tx(&self, tx: MutTx) -> Result<Option<(TxOffset, Arc<TxData>, TxMetrics, String)>, DBError> {
         log::trace!("COMMIT MUT TX");
 
-        // TODO: Never returns `None` -- should it?
         let reducer_context = tx.ctx.reducer_context().cloned();
+        // TODO: Never returns `None` -- should it?
         let Some((tx_offset, tx_data, tx_metrics, reducer)) = self.inner.commit_mut_tx(tx)? else {
             return Ok(None);
         };
 
         self.maybe_do_snapshot(&tx_data);
 
+        let tx_data = Arc::new(tx_data);
         if let Some(durability) = &self.durability {
-            Self::do_durability(&**durability, reducer_context.as_ref(), &tx_data)
+            durability.request_durability(reducer_context, &tx_data);
         }
 
         Ok(Some((tx_offset, tx_data, tx_metrics, reducer)))
@@ -789,7 +792,7 @@ impl RelationalDB {
         &self,
         tx: MutTx,
         workload: Workload,
-    ) -> Result<Option<(TxData, TxMetrics, Tx)>, DBError> {
+    ) -> Result<Option<(Arc<TxData>, TxMetrics, Tx)>, DBError> {
         log::trace!("COMMIT MUT TX");
 
         let Some((tx_data, tx_metrics, tx)) = self.inner.commit_mut_tx_downgrade(tx, workload)? else {
@@ -798,82 +801,14 @@ impl RelationalDB {
 
         self.maybe_do_snapshot(&tx_data);
 
+        let tx_data = Arc::new(tx_data);
         if let Some(durability) = &self.durability {
-            Self::do_durability(&**durability, tx.ctx.reducer_context(), &tx_data)
+            durability.request_durability(tx.ctx.reducer_context().cloned(), &tx_data);
         }
 
         Ok(Some((tx_data, tx_metrics, tx)))
     }
 
-    /// If `(tx_data, ctx)` should be appended to the commitlog, do so.
-    ///
-    /// Note that by this stage,
-    /// [`spacetimedb_datastore::locking_tx_datastore::committed_state::tx_consumes_offset`]
-    /// has already decided based on the reducer and operations whether the transaction should be appended;
-    /// this method is responsible only for reading its decision out of the `tx_data`
-    /// and calling `durability.append_tx`.
-    fn do_durability(durability: &Durability, reducer_context: Option<&ReducerContext>, tx_data: &TxData) {
-        use commitlog::payload::{
-            txdata::{Mutations, Ops},
-            Txdata,
-        };
-
-        let is_not_ephemeral_table = |table_id: &TableId| -> bool {
-            tx_data
-                .ephemeral_tables()
-                .map(|etables| !etables.contains(table_id))
-                .unwrap_or(true)
-        };
-
-        if tx_data.tx_offset().is_some() {
-            let inserts: Box<_> = tx_data
-                .inserts()
-                // Skip ephemeral tables
-                .filter(|(table_id, _)| is_not_ephemeral_table(table_id))
-                .map(|(table_id, rowdata)| Ops {
-                    table_id: *table_id,
-                    rowdata: rowdata.clone(),
-                })
-                .collect();
-
-            let truncates: IntSet<TableId> = tx_data.truncates().collect();
-
-            let deletes: Box<_> = tx_data
-                .deletes()
-                .filter(|(table_id, _)| is_not_ephemeral_table(table_id))
-                .map(|(table_id, rowdata)| Ops {
-                    table_id: *table_id,
-                    rowdata: rowdata.clone(),
-                })
-                // filter out deletes for tables that are truncated in the same transaction.
-                .filter(|ops| !truncates.contains(&ops.table_id))
-                .collect();
-
-            let truncates = truncates.into_iter().filter(is_not_ephemeral_table).collect();
-
-            let inputs = reducer_context.map(|rcx| rcx.into());
-
-            let txdata = Txdata {
-                inputs,
-                outputs: None,
-                mutations: Some(Mutations {
-                    inserts,
-                    deletes,
-                    truncates,
-                }),
-            };
-
-            // TODO: Should measure queuing time + actual write
-            durability.append_tx(txdata);
-        } else {
-            debug_assert!(
-                !tx_data.has_rows_or_connect_disconnect(reducer_context),
-                "tx_data has no rows but has connect/disconnect: `{:?}`",
-                reducer_context.map(|rcx| &rcx.name),
-            );
-        }
-    }
-
     /// Get the [`DurableOffset`] of this database, or `None` if this is an
     /// in-memory instance.
     pub fn durable_tx_offset(&self) -> Option<DurableOffset> {
@@ -1511,8 +1446,8 @@ impl RelationalDB {
     }
 
     /// Reports the metrics for `reducer`, using counters provided by `db`.
-    pub fn report_mut_tx_metrics(&self, reducer: String, metrics: TxMetrics, tx_data: Option<TxData>) {
-        self.report_tx_metrics(reducer, tx_data.map(Arc::new), Some(metrics), None);
+    pub fn report_mut_tx_metrics(&self, reducer: String, metrics: TxMetrics, tx_data: Option<Arc<TxData>>) {
+        self.report_tx_metrics(reducer, tx_data, Some(metrics), None);
     }
 
     /// Reports subscription metrics for `reducer`, using counters provided by `db`.
diff --git a/crates/core/src/subscription/module_subscription_actor.rs b/crates/core/src/subscription/module_subscription_actor.rs
@@ -1031,7 +1031,7 @@ impl ModuleSubscriptions {
                     return Ok(Err(WriteConflict));
                 };
                 *db_update = DatabaseUpdate::from_writes(&tx_data);
-                (read_tx, Arc::new(tx_data), tx_metrics)
+                (read_tx, tx_data, tx_metrics)
             }
             EventStatus::Failed(_) | EventStatus::OutOfEnergy => {
                 // If the transaction failed, we need to rollback the mutable tx.
@@ -1198,7 +1198,7 @@ impl ModuleSubscriptions {
                     let _ = extra.send(tx_offset);
                 }
                 self.relational_db
-                    .report_tx_metrics(reducer, Some(Arc::new(tx_data)), Some(tx_metrics_mut), None);
+                    .report_tx_metrics(reducer, Some(tx_data), Some(tx_metrics_mut), None);
             }
         });
         (guard, offset_rx)
diff --git a/crates/datastore/src/execution_context.rs b/crates/datastore/src/execution_context.rs
@@ -39,17 +39,17 @@ pub struct ReducerContext {
     pub arg_bsatn: Bytes,
 }
 
-impl From<&ReducerContext> for txdata::Inputs {
+impl From<ReducerContext> for txdata::Inputs {
     fn from(
         ReducerContext {
             name,
             caller_identity,
             caller_connection_id,
             timestamp,
             arg_bsatn,
-        }: &ReducerContext,
+        }: ReducerContext,
     ) -> Self {
-        let reducer_name = Arc::new(Varchar::from_str_truncate(name));
+        let reducer_name = Arc::new(Varchar::from_string_truncate(name));
         let cap = arg_bsatn.len()
         /* caller_identity */
         + 32
@@ -58,10 +58,10 @@ impl From<&ReducerContext> for txdata::Inputs {
         /* timestamp */
         + 8;
         let mut buf = Vec::with_capacity(cap);
-        bsatn::to_writer(&mut buf, caller_identity).unwrap();
-        bsatn::to_writer(&mut buf, caller_connection_id).unwrap();
-        bsatn::to_writer(&mut buf, timestamp).unwrap();
-        buf.extend_from_slice(arg_bsatn);
+        bsatn::to_writer(&mut buf, &caller_identity).unwrap();
+        bsatn::to_writer(&mut buf, &caller_connection_id).unwrap();
+        bsatn::to_writer(&mut buf, &timestamp).unwrap();
+        buf.extend_from_slice(&arg_bsatn);
 
         txdata::Inputs {
             reducer_name,
diff --git a/crates/datastore/src/locking_tx_datastore/committed_state.rs b/crates/datastore/src/locking_tx_datastore/committed_state.rs
@@ -675,7 +675,7 @@ impl CommittedState {
         // Note that this may change in the future: some analytics and/or
         // timetravel queries may benefit from seeing all inputs, even if
         // the database state did not change.
-        tx_data.has_rows_or_connect_disconnect(ctx.reducer_context())
+        tx_data.has_rows_or_connect_disconnect(ctx.reducer_context().map(|rcx| &*rcx.name))
     }
 
     pub(super) fn drop_view_from_read_sets(&mut self, view_id: ViewId, sender: Option<Identity>) {
diff --git a/crates/datastore/src/traits.rs b/crates/datastore/src/traits.rs

Original file line number	Diff line number	Diff line change
`@@ -1031,7 +1031,7 @@ impl ModuleSubscriptions {`
`1031`	`1031`	`return Ok(Err(WriteConflict));`
`1032`	`1032`	`};`
`1033`	`1033`	`*db_update = DatabaseUpdate::from_writes(&tx_data);`
`1034`		`- (read_tx, Arc::new(tx_data), tx_metrics)`
	`1034`	`+ (read_tx, tx_data, tx_metrics)`
`1035`	`1035`	`}`
`1036`	`1036`	`EventStatus::Failed(_) \| EventStatus::OutOfEnergy => {`
`1037`	`1037`	`// If the transaction failed, we need to rollback the mutable tx.`
`@@ -1198,7 +1198,7 @@ impl ModuleSubscriptions {`
`1198`	`1198`	`let _ = extra.send(tx_offset);`
`1199`	`1199`	`}`
`1200`	`1200`	`self.relational_db`
`1201`		`- .report_tx_metrics(reducer, Some(Arc::new(tx_data)), Some(tx_metrics_mut), None);`
	`1201`	`+ .report_tx_metrics(reducer, Some(tx_data), Some(tx_metrics_mut), None);`
`1202`	`1202`	`}`
`1203`	`1203`	`});`
`1204`	`1204`	`(guard, offset_rx)`
Original file line number	Diff line number	Diff line change
`@@ -675,7 +675,7 @@ impl CommittedState {`
`675`	`675`	`// Note that this may change in the future: some analytics and/or`
`676`	`676`	`// timetravel queries may benefit from seeing all inputs, even if`
`677`	`677`	`// the database state did not change.`
`678`		`- tx_data.has_rows_or_connect_disconnect(ctx.reducer_context())`
	`678`	`+ tx_data.has_rows_or_connect_disconnect(ctx.reducer_context().map(\|rcx\| &*rcx.name))`
`679`	`679`	`}`
`680`	`680`
`681`	`681`	`pub(super) fn drop_view_from_read_sets(&mut self, view_id: ViewId, sender: Option<Identity>) {`