Make subscribe durable (#3894)

# Description of Changes

With the addition of module-defined views, subscriptions are no longer
read-only as they may invoke view materialization.

The way this works is that a subscription starts off as a mutable
transaction, materializes views if necessary, and then downgrades to a
read-only transaction to evaluate the subscription.

Before this patch, we were calling `commit_downgrade` directly on the
`MutTxId` in order to downgrade the transaction. This would update the
in-memory `CommittedState`, but it wouldn't make the transaction
durable.

This would result in us incrementing the transaction offset of the
in-memory `CommittedState` without writing anything to the commitlog.
This in turn would invalidate snapshots as they would be pointing
further ahead into the commitlog than they should, and so when replaying
from a snapshot we would potentially skip over commits that were not
included in the snapshot.

This patch changes those call sites to use
`RelationalDB::commit_tx_downgrade` which both updates the in-memory
state **and** makes the transaction durable.

**NOTE:** The fact that views are materialized is purely an
implementation detail at this point in time. And technically view tables
are ephemeral meaning they are not persisted to the commitlog. So the
real bug here was that we were updating the tx offset of the in-memory
committed state at all. This is technically fixed by
#3884 and so after
#3884 lands this change
becomes a no-op. However, we still shouldn't be calling `commit` and
`commit_downgrade` directly on a `MutTxId` since in most cases it is
wrong to bypass the durability layer. And without this change, the bug
would still be present were view tables not ephemeral, which they may
not be at some point in the future.

# API and ABI breaking changes

None

# Expected complexity level and risk

1. The change itself is trivial, the bug is not.

# Testing

Adding an automated test for this is not so straightforward. First it's
view related which means we don't have many options apart from a smoke
test, but I don't believe the smoke tests have a mechanism for replaying
the commitlog.

If transaction offsets are supposed to be linear, without any gaps, then
it would be useful to assert that on each append, in which case we could
write a smoke test that would fail as soon as the offsets diverged.

Author: joshua-spacetime

crates/core/src/db/relational_db.rs

@@ -827,16 +827,10 @@ impl RelationalDB {
     }
 
     #[tracing::instrument(level = "trace", skip_all)]
-    pub fn commit_tx_downgrade(
-        &self,
-        tx: MutTx,
-        workload: Workload,
-    ) -> Result<Option<(Arc<TxData>, TxMetrics, Tx)>, DBError> {
+    pub fn commit_tx_downgrade(&self, tx: MutTx, workload: Workload) -> (Arc<TxData>, TxMetrics, Tx) {
         log::trace!("COMMIT MUT TX");
 
-        let Some((tx_data, tx_metrics, tx)) = self.inner.commit_mut_tx_downgrade(tx, workload)? else {
-            return Ok(None);
-        };
+        let (tx_data, tx_metrics, tx) = self.inner.commit_mut_tx_downgrade(tx, workload);
 
         self.maybe_do_snapshot(&tx_data);
 
@@ -845,7 +839,7 @@ impl RelationalDB {
             durability.request_durability(tx.ctx.reducer_context().cloned(), &tx_data);
         }
 
-        Ok(Some((tx_data, tx_metrics, tx)))
+        (tx_data, tx_metrics, tx)
     }
 
     /// Get the [`DurableOffset`] of this database, or `None` if this is an

crates/core/src/sql/execute.rs

@@ -1,4 +1,3 @@
-use std::sync::Arc;
 use std::time::Duration;
 
 use super::ast::SchemaViewer;
@@ -215,20 +214,15 @@ pub async fn run(
                 None => tx,
             };
 
-            let (tx_data, tx_metrics_mut, tx) = tx.commit_downgrade(Workload::Sql);
+            let (tx_data, tx_metrics_mut, tx) = db.commit_tx_downgrade(tx, Workload::Sql);
 
             let (tx_offset_send, tx_offset) = oneshot::channel();
             // Release the tx on drop, so that we record metrics
             // and set the transaction offset.
             let mut tx = scopeguard::guard(tx, |tx| {
                 let (offset, tx_metrics_downgrade, reducer) = db.release_tx(tx);
                 let _ = tx_offset_send.send(offset);
-                db.report_tx_metrics(
-                    reducer,
-                    Some(Arc::new(tx_data)),
-                    Some(tx_metrics_mut),
-                    Some(tx_metrics_downgrade),
-                );
+                db.report_tx_metrics(reducer, Some(tx_data), Some(tx_metrics_mut), Some(tx_metrics_downgrade));
             });
 
             // Compute the header for the result set

crates/core/src/subscription/module_subscription_actor.rs

@@ -1027,9 +1027,7 @@ impl ModuleSubscriptions {
         // We'll later ensure tx is released/cleaned up once out of scope.
         let (read_tx, tx_data, tx_metrics_mut) = match &mut event.status {
             EventStatus::Committed(db_update) => {
-                let Some((tx_data, tx_metrics, read_tx)) = stdb.commit_tx_downgrade(tx, Workload::Update)? else {
-                    return Ok(Err(WriteConflict));
-                };
+                let (tx_data, tx_metrics, read_tx) = stdb.commit_tx_downgrade(tx, Workload::Update);
                 *db_update = DatabaseUpdate::from_writes(&tx_data);
                 (read_tx, tx_data, tx_metrics)
             }
@@ -1102,7 +1100,7 @@ impl ModuleSubscriptions {
         sender: Identity,
     ) -> Result<(TxGuard<impl FnOnce(TxId) + '_>, TransactionOffset), DBError> {
         Self::_unsubscribe_views(&mut tx, view_collector, sender)?;
-        let (tx_data, tx_metrics_mut, tx) = tx.commit_downgrade(Workload::Subscribe);
+        let (tx_data, tx_metrics_mut, tx) = self.relational_db.commit_tx_downgrade(tx, Workload::Subscribe);
         let opts = GuardTxOptions::from_mut(tx_data, tx_metrics_mut);
         Ok(self.guard_tx(tx, opts))
     }
@@ -1136,7 +1134,7 @@ impl ModuleSubscriptions {
                 .materialize_views(tx, view_collector, sender, Workload::Subscribe)
                 .await?
         }
-        let (tx_data, tx_metrics_mut, tx) = tx.commit_downgrade(Workload::Subscribe);
+        let (tx_data, tx_metrics_mut, tx) = self.relational_db.commit_tx_downgrade(tx, Workload::Subscribe);
         let opts = GuardTxOptions::from_mut(tx_data, tx_metrics_mut);
         Ok(self.guard_tx(tx, opts))
     }
@@ -1229,10 +1227,10 @@ impl GuardTxOptions {
         }
     }
 
-    fn from_mut(tx_data: TxData, tx_metrics_mut: TxMetrics) -> Self {
+    fn from_mut(tx_data: Arc<TxData>, tx_metrics_mut: TxMetrics) -> Self {
         Self {
             extra_tx_offset_sender: None,
-            tx_data: Some(Arc::new(tx_data)),
+            tx_data: Some(tx_data),
             tx_metrics_mut: tx_metrics_mut.into(),
         }
     }

crates/core/src/subscription/query.rs

@@ -1087,7 +1087,7 @@ mod tests {
             }
         }
 
-        let (data, _, tx) = tx.commit_downgrade(Workload::ForTests);
+        let (data, _, tx) = db.commit_tx_downgrade(tx, Workload::ForTests);
         let table_id = plan.subscribed_table_id();
         // This awful construction to convert `Arc<str>` into `Box<str>`.
         let table_name = (&**plan.subscribed_table_name()).into();

crates/datastore/src/locking_tx_datastore/datastore.rs

@@ -950,6 +950,8 @@ impl MutTx for Locking {
         tx.rollback()
     }
 
+    /// This method only updates the in-memory `committed_state`.
+    /// For durability, see `RelationalDB::commit_tx`.
     fn commit_mut_tx(&self, tx: Self::MutTx) -> Result<Option<(TxOffset, TxData, TxMetrics, String)>> {
         Ok(Some(tx.commit()))
     }
@@ -960,12 +962,10 @@ impl Locking {
         tx.rollback_downgrade(workload)
     }
 
-    pub fn commit_mut_tx_downgrade(
-        &self,
-        tx: MutTxId,
-        workload: Workload,
-    ) -> Result<Option<(TxData, TxMetrics, TxId)>> {
-        Ok(Some(tx.commit_downgrade(workload)))
+    /// This method only updates the in-memory `committed_state`.
+    /// For durability, see `RelationalDB::commit_tx_downgrade`.
+    pub fn commit_mut_tx_downgrade(&self, tx: MutTxId, workload: Workload) -> (TxData, TxMetrics, TxId) {
+        tx.commit_downgrade(workload)
     }
 }
 

crates/datastore/src/locking_tx_datastore/mut_tx.rs

@@ -1902,6 +1902,11 @@ impl MutTxId {
     /// Commits this transaction in memory, applying its changes to the committed state.
     /// This doesn't handle the persistence layer at all.
     ///
+    /// IMPORTANT: This method updates the in-memory state of the database but does not make it durable.
+    /// That is, the tx will not be persisted to the commitlog.
+    /// Hence you should be careful when calling this method directly.
+    /// In most cases you'll want to use `RelationalDB::commit_tx` which makes the tx durable.
+    ///
     /// Returns:
     /// - [`TxData`], the set of inserts and deletes performed by this transaction.
     /// - [`TxMetrics`], various measurements of the work performed by this transaction.
@@ -1947,11 +1952,16 @@ impl MutTxId {
     /// The lock on the committed state is converted into a read lock,
     /// and returned as a new read-only transaction.
     ///
+    /// IMPORTANT: This method updates the in-memory state of the database but does not make it durable.
+    /// That is, the tx will not be persisted to the commitlog.
+    /// Hence you should be careful when calling this method directly.
+    /// In most cases you'll want to use `RelationalDB::commit_tx_downgrade` which makes the tx durable.
+    ///
     /// Returns:
     /// - [`TxData`], the set of inserts and deletes performed by this transaction.
     /// - [`TxMetrics`], various measurements of the work performed by this transaction.
     /// - [`TxId`], a read-only transaction with a shared lock on the committed state.
-    pub fn commit_downgrade(mut self, workload: Workload) -> (TxData, TxMetrics, TxId) {
+    pub(super) fn commit_downgrade(mut self, workload: Workload) -> (TxData, TxMetrics, TxId) {
         let tx_data = self
             .committed_state_write_lock
             .merge(self.tx_state, self.read_sets, &self.ctx);