Drop datastore lock after requesting durability

Author: joshua-spacetime

crates/core/src/db/durability.rs

@@ -1,15 +1,14 @@
-use std::{num::NonZeroUsize, sync::Arc, time::Duration};
+use std::{sync::Arc, time::Duration};
 
 use futures::TryFutureExt as _;
 use log::{error, info};
-use prometheus::IntGauge;
 use spacetimedb_commitlog::payload::{
     txdata::{Mutations, Ops},
     Txdata,
 };
 use spacetimedb_datastore::{execution_context::ReducerContext, traits::TxData};
 use spacetimedb_durability::Durability as _;
-use spacetimedb_durability::{DurableOffset, ReorderWindow, Transaction, TxOffset};
+use spacetimedb_durability::{DurableOffset, Transaction, TxOffset};
 use spacetimedb_lib::Identity;
 use spacetimedb_sats::ProductValue;
 use tokio::{
@@ -39,36 +38,8 @@ type ShutdownReply = oneshot::Sender<OwnedNotified>;
 /// Represents a handle to a background task that persists transactions
 /// according to the [`Durability`] policy provided.
 ///
-/// This exists to avoid holding a transaction lock while
-/// preparing the [TxData] for processing by the [Durability] layer.
-///
-/// The durability worker is internal to [RelationalDB], which calls
-/// [DurabilityWorker::request_durability] after committing a transaction.
-///
-/// # Transaction ordering
-///
-/// The backing datastore of [RelationalDB] is responsible for creating a total
-/// ordering of transactions and must uphold that [TxOffset]s are monotonically
-/// increasing without gaps.
-///
-/// However, [RelationalDB::commit_tx] respectively [RelationalDB::commit_tx_downgrade]
-/// may be called from multiple threads. Because those methods are not
-/// synchronized, and release the transaction lock before requesting durability,
-/// it is possible for [DurabilityRequest]s to appear slightly out-of-order on
-/// the worker channel.
-///
-/// To mitigate this, the worker keeps a window of up to `reorder_window_size`
-/// requests if out-of-order requests are detected, and flushes it to the
-/// underlying durability layer once it is able to linearize the offset sequence.
-///
-/// Since we expect out-of-order requests to happen very rarely, this measure
-/// should not negatively impact throughput in the common case, unlike holding
-/// the transaction lock until request submission is complete.
-///
-/// Note that the commitlog rejects out-of-order commits, so if a missing offset
-/// arrives outside `reorder_window_size` (or never), already committed
-/// transactions may be lost (by way of the durability worker crashing).
-/// Those transactions will not be confirmed, however, so this is safe.
+/// The durability worker is internal to [RelationalDB], which uses
+/// [DurabilityWorker::request_durability] while finalizing a transaction.
 ///
 /// [RelationalDB]: crate::db::relational_db::RelationalDB
 pub struct DurabilityWorker {
@@ -92,24 +63,14 @@ impl DurabilityWorker {
     /// Create a new [`DurabilityWorker`] using the given `durability` policy.
     ///
     /// Background tasks will be spawned onto to provided tokio `runtime`.
-    pub fn new(
-        database: Identity,
-        durability: Arc<Durability>,
-        runtime: runtime::Handle,
-        next_tx_offset: TxOffset,
-        reorder_window_size: NonZeroUsize,
-    ) -> Self {
+    pub fn new(database: Identity, durability: Arc<Durability>, runtime: runtime::Handle) -> Self {
         let (request_tx, request_rx) = channel(4 * 4096);
         let (shutdown_tx, shutdown_rx) = channel(1);
 
         let actor = DurabilityWorkerActor {
             request_rx,
             shutdown: shutdown_rx,
             durability: durability.clone(),
-            reorder_window: ReorderWindow::new(next_tx_offset, reorder_window_size),
-            reorder_window_len: WORKER_METRICS
-                .durability_worker_reorder_window_length
-                .with_label_values(&database),
         };
         let _enter = runtime.enter();
         tokio::spawn(
@@ -129,9 +90,7 @@ impl DurabilityWorker {
         }
     }
 
-    /// Create a [`DurabilityWorker`] that uses the local commitlog durability
-    /// actor directly. This removes the extra core durability actor so the
-    /// local path has only one queued background worker.
+    /// Create a [`DurabilityWorker`] that uses the local commitlog durability actor directly.
     pub fn new_local(database: Identity, durability: LocalDurability, runtime: runtime::Handle) -> Self {
         Self {
             database,
@@ -161,6 +120,15 @@ impl DurabilityWorker {
     /// - [Self::shutdown] was called
     ///
     pub fn request_durability(&self, reducer_context: Option<ReducerContext>, tx_data: &Arc<TxData>) {
+        let Some(tx_offset) = tx_data.tx_offset() else {
+            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;
+        };
+
         match &self.inner {
             DurabilityWorkerInner::Generic { request_tx, .. } => {
                 // We first try to send it without blocking.
@@ -199,20 +167,10 @@ impl DurabilityWorker {
                 }
             }
             DurabilityWorkerInner::Local { durability } => {
-                let Some(tx_offset) = tx_data.tx_offset() else {
-                    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 start = std::time::Instant::now();
                 let tx_data = tx_data.clone();
-                let blocked = durability.append_tx_deferred(tx_offset, move || {
-                    prepare_tx_data_for_durability(tx_offset, reducer_context, &tx_data)
-                });
+                let blocked = durability
+                    .append_tx_deferred(move || prepare_tx_data_for_durability(tx_offset, reducer_context, &tx_data));
                 if blocked {
                     WORKER_METRICS
                         .durability_blocking_send_duration
@@ -302,20 +260,15 @@ pub struct DurabilityWorkerActor {
     request_rx: mpsc::Receiver<DurabilityRequest>,
     shutdown: Receiver<ShutdownReply>,
     durability: Arc<Durability>,
-    reorder_window: ReorderWindow<DurabilityRequest>,
-    reorder_window_len: IntGauge,
 }
 
 impl DurabilityWorkerActor {
     /// Processes requests to do durability.
     async fn run(mut self) {
         // When this future completes or is cancelled, ensure that:
         // - shutdown waiters are notified
-        // - metrics are reset
-        let done = scopeguard::guard(Arc::new(Notify::new()), |done| {
-            done.notify_waiters();
-            self.reorder_window_len.set(0);
-        });
+        let done = scopeguard::guard(Arc::new(Notify::new()), |done| done.notify_waiters());
+        let mut request_buf = Vec::with_capacity(4096);
 
         loop {
             tokio::select! {
@@ -328,35 +281,16 @@ impl DurabilityWorkerActor {
                     let _ = reply.send(done.clone().notified_owned());
                 },
 
-                req = self.request_rx.recv() => {
-                    let Some(request) = req else {
+                n = self.request_rx.recv_many(&mut request_buf, usize::MAX) => {
+                    if n == 0 {
                         break;
-                    };
-                    match request.tx_data.tx_offset() {
-                        // Drop the request if it doesn't have a tx offset.
-                        None => {
-                            let name = request.reducer_context.as_ref().map(|rcx| &rcx.name);
-                            debug_assert!(
-                                !request.tx_data.has_rows_or_connect_disconnect(name),
-                                "tx_data has no rows but has connect/disconnect: `{name:?}`"
-                            );
-                        },
-                        // Otherwise, push to the reordering window.
-                        Some(tx_offset) => {
-                            if let Err(e) = self.reorder_window.push(tx_offset, request) {
-                                error!("{e}");
-                                break;
-                            }
-                        },
                     }
                 }
             }
 
-            // Drain all requests that are properly ordered.
-            self.reorder_window
-                .drain()
+            request_buf
+                .drain(..)
                 .for_each(|request| Self::do_durability(&*self.durability, request.reducer_context, &request.tx_data));
-            self.reorder_window_len.set(self.reorder_window.len() as _);
         }
 
         info!("durability worker actor done");
@@ -483,13 +417,7 @@ mod tests {
     #[tokio::test(flavor = "multi_thread")]
     async fn shutdown_waits_until_durable() {
         let durability = Arc::new(CountingDurability::default());
-        let worker = DurabilityWorker::new(
-            Identity::ONE,
-            durability.clone(),
-            runtime::Handle::current(),
-            0,
-            NonZeroUsize::new(1).unwrap(),
-        );
+        let worker = DurabilityWorker::new(Identity::ONE, durability.clone(), runtime::Handle::current());
         for i in 0..=10 {
             let mut txdata = TxData::default();
             txdata.set_tx_offset(i);

crates/core/src/db/persistence.rs

@@ -30,9 +30,10 @@ pub type DiskSizeFn = Arc<dyn Fn() -> io::Result<SizeOnDisk> + Send + Sync>;
 pub struct Persistence {
     /// The [Durability] to use, for persisting transactions.
     pub durability: Arc<Durability>,
-    /// The concrete local durability handle, when the database uses the built-in
-    /// local commitlog path. This allows the core DB layer to bypass helper
-    /// actors on the local hot path without changing the generic trait object.
+    /// TODO: Merge this local durability handle with the generic trait object above.
+    /// This allows us to bypass an actor whose only responsibility is to generate a
+    /// commitlog payload from `TxData`. Ultimately though this should just be a part
+    /// of the [Durability] implementation itself.
     pub local_durability: Option<LocalDurability>,
     /// The [DiskSizeFn].
     ///

crates/core/src/db/relational_db.rs

@@ -58,7 +58,6 @@ use spacetimedb_table::table::{RowRef, TableScanIter};
 use spacetimedb_table::table_index::IndexKey;
 use std::borrow::Cow;
 use std::io;
-use std::num::NonZeroUsize;
 use std::ops::RangeBounds;
 use std::sync::Arc;
 use tokio::sync::watch;
@@ -122,7 +121,6 @@ pub struct RelationalDB {
 // this value, later introduction of dynamic configuration will allow the
 // compiler to find external dependencies.
 pub const SNAPSHOT_FREQUENCY: u64 = 1_000_000;
-const GENERIC_DURABILITY_REORDER_WINDOW_SIZE: NonZeroUsize = NonZeroUsize::new(8).unwrap();
 
 impl std::fmt::Debug for RelationalDB {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
@@ -157,21 +155,7 @@ impl RelationalDB {
             (Some(local_durability), _, Some(rt)) => {
                 Some(DurabilityWorker::new_local(database_identity, local_durability, rt))
             }
-            (None, Some(durability), Some(rt)) => {
-                let next_tx_offset = {
-                    let tx = inner.begin_tx(Workload::Internal);
-                    let next_tx_offset = tx.tx_offset();
-                    let _ = inner.release_tx(tx);
-                    next_tx_offset.into_inner()
-                };
-                Some(DurabilityWorker::new(
-                    database_identity,
-                    durability,
-                    rt,
-                    next_tx_offset,
-                    GENERIC_DURABILITY_REORDER_WINDOW_SIZE,
-                ))
-            }
+            (None, Some(durability), Some(rt)) => Some(DurabilityWorker::new(database_identity, durability, rt)),
             _ => None,
         };
 
@@ -826,33 +810,33 @@ impl RelationalDB {
 
         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 {
+        let Some((tx_offset, tx_data, tx_metrics, reducer)) = self.inner.commit_mut_tx_and_then(tx, |tx_data| {
+            if let Some(durability) = &self.durability {
+                durability.request_durability(reducer_context, tx_data);
+            }
+        })?
+        else {
             return Ok(None);
         };
 
         self.maybe_do_snapshot(&tx_data);
 
-        let tx_data = Arc::new(tx_data);
-        if let Some(durability) = &self.durability {
-            durability.request_durability(reducer_context, &tx_data);
-        }
-
         Ok(Some((tx_offset, tx_data, tx_metrics, reducer)))
     }
 
     #[tracing::instrument(level = "trace", skip_all)]
     pub fn commit_tx_downgrade(&self, tx: MutTx, workload: Workload) -> (Arc<TxData>, TxMetrics, Tx) {
         log::trace!("COMMIT MUT TX");
 
-        let (tx_data, tx_metrics, tx) = self.inner.commit_mut_tx_downgrade(tx, workload);
+        let reducer_context = tx.ctx.reducer_context().cloned();
+        let (tx_data, tx_metrics, tx) = self.inner.commit_mut_tx_downgrade_and_then(tx, workload, |tx_data| {
+            if let Some(durability) = &self.durability {
+                durability.request_durability(reducer_context, tx_data);
+            }
+        });
 
         self.maybe_do_snapshot(&tx_data);
 
-        let tx_data = Arc::new(tx_data);
-        if let Some(durability) = &self.durability {
-            durability.request_durability(tx.ctx.reducer_context().cloned(), &tx_data);
-        }
-
         (tx_data, tx_metrics, tx)
     }
 

crates/core/src/worker_metrics/mod.rs

@@ -491,11 +491,6 @@ metrics_group!(
         #[labels(database_identity: Identity)]
         #[buckets(0.001, 0.01, 0.1, 1.0, 10.0)]
         pub durability_blocking_send_duration: HistogramVec,
-
-        #[name = spacetime_durability_worker_reorder_window_length]
-        #[help = "The number of transactions currently being held in the reorder window"]
-        #[labels(db: Identity)]
-        pub durability_worker_reorder_window_length: IntGaugeVec,
     }
 );
 

crates/datastore/src/locking_tx_datastore/datastore.rs

@@ -968,6 +968,27 @@ impl Locking {
     pub fn commit_mut_tx_downgrade(&self, tx: MutTxId, workload: Workload) -> (TxData, TxMetrics, TxId) {
         tx.commit_downgrade(workload)
     }
+
+    /// Commit `tx` and invoke `before_release` while the write lock is still held.
+    #[allow(clippy::type_complexity)]
+    pub fn commit_mut_tx_and_then(
+        &self,
+        tx: MutTxId,
+        before_release: impl FnOnce(&Arc<TxData>),
+    ) -> Result<Option<(TxOffset, Arc<TxData>, TxMetrics, Option<ReducerName>)>> {
+        Ok(Some(tx.commit_and_then(before_release)))
+    }
+
+    /// Commit `tx`, invoke `before_downgrade` while the write lock is still held,
+    /// then downgrade the lock to a read-only transaction.
+    pub fn commit_mut_tx_downgrade_and_then(
+        &self,
+        tx: MutTxId,
+        workload: Workload,
+        before_downgrade: impl FnOnce(&Arc<TxData>),
+    ) -> (Arc<TxData>, TxMetrics, TxId) {
+        tx.commit_downgrade_and_then(workload, before_downgrade)
+    }
 }
 
 #[derive(Debug, Error)]