Ensure all-ephemeral transactions don't consume a tx offset (#3884)

Views are materialized in mutable transactions, but should not increment
the transaction offset maintained in the committed state.

This fixes storing completely empty transactions in the commitlog, and
maintains that the committed state tx offset is in-sync with the
commitlog's tx offset.

# Expected complexity level and risk

2

# Testing

Added a test.

---------

Signed-off-by: Kim Altintop <kim@eagain.io>
Co-authored-by: Mazdak Farrokhzad <twingoow@gmail.com>

Author: kim

crates/core/src/db/relational_db.rs

@@ -818,18 +818,13 @@ impl RelationalDB {
             Txdata,
         };
 
-        let is_not_ephemeral_table = |table_id: &TableId| -> bool {
-            tx_data
-                .ephemeral_tables()
-                .map(|etables| !etables.contains(table_id))
-                .unwrap_or(true)
-        };
+        let is_persistent_table = |table_id: &TableId| -> bool { !tx_data.is_ephemeral_table(table_id) };
 
         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))
+                .filter(|(table_id, _)| is_persistent_table(table_id))
                 .map(|(table_id, rowdata)| Ops {
                     table_id: *table_id,
                     rowdata: rowdata.clone(),
@@ -840,7 +835,7 @@ impl RelationalDB {
 
             let deletes: Box<_> = tx_data
                 .deletes()
-                .filter(|(table_id, _)| is_not_ephemeral_table(table_id))
+                .filter(|(table_id, _)| is_persistent_table(table_id))
                 .map(|(table_id, rowdata)| Ops {
                     table_id: *table_id,
                     rowdata: rowdata.clone(),
@@ -849,10 +844,15 @@ impl RelationalDB {
                 .filter(|ops| !truncates.contains(&ops.table_id))
                 .collect();
 
-            let truncates = truncates.into_iter().filter(is_not_ephemeral_table).collect();
+            let truncates: Box<_> = truncates.into_iter().filter(is_persistent_table).collect();
 
             let inputs = reducer_context.map(|rcx| rcx.into());
 
+            debug_assert!(
+                !(inserts.is_empty() && truncates.is_empty() && deletes.is_empty() && inputs.is_none()),
+                "empty transaction"
+            );
+
             let txdata = Txdata {
                 inputs,
                 outputs: None,
@@ -2632,6 +2632,36 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_view_materialization_does_not_consume_tx_offset() -> ResultTest<()> {
+        let stdb = TestDB::durable_without_snapshot_repo()?;
+
+        let (tx_offset_1, view_id, table_id) = {
+            let module_def = view_module_def();
+            let view_def = module_def.view("my_view").unwrap();
+
+            let mut tx = begin_mut_tx(&stdb);
+            let (view_id, table_id) = stdb.create_view(&mut tx, &module_def, view_def)?;
+            let (tx_offset, tx_data, ..) = stdb.commit_tx(tx)?.unwrap();
+            assert_eq!(Some(tx_offset), tx_data.tx_offset());
+
+            (tx_offset, view_id, table_id)
+        };
+
+        let mut tx = begin_mut_tx(&stdb);
+        tx.subscribe_view(view_id, ArgId::SENTINEL, Identity::ONE)?;
+        stdb.materialize_view(&mut tx, table_id, Identity::ONE, vec![product![10u8]])?;
+        let (tx_offset_2, tx_data, ..) = stdb.commit_tx(tx)?.unwrap();
+
+        // `tx_data.tx_offset()` should return `None`,
+        // so that it is not considered for durability.
+        // The tx offset reported for confirmed reads should stay the same.
+        assert!(tx_data.tx_offset().is_none());
+        assert_eq!(tx_offset_1, tx_offset_2);
+
+        Ok(())
+    }
+
     #[test]
     fn test_view_tables_are_ephemeral_with_snapshot() -> ResultTest<()> {
         let stdb = TestDB::durable()?;

crates/datastore/src/locking_tx_datastore/committed_state.rs

@@ -747,15 +747,18 @@ impl CommittedState {
         // so that we can pass updated set of table ids.
         self.merge_read_sets(read_sets);
 
+        // Store in `tx_data` which of the updated tables are ephemeral.
+        // NOTE: This must be called before `tx_consumes_offset`, so that
+        // all-ephemeral transactions do not consume a tx offset.
+        tx_data.set_ephemeral_tables(&self.ephemeral_tables);
+
         // If the TX will be logged, record its projected tx offset,
         // then increment the counter.
         if self.tx_consumes_offset(&tx_data, ctx) {
             tx_data.set_tx_offset(self.next_tx_offset);
             self.next_tx_offset += 1;
         }
 
-        tx_data.set_ephemeral_tables(&self.ephemeral_tables);
-
         tx_data
     }
 

crates/datastore/src/traits.rs

@@ -236,7 +236,7 @@ impl TxData {
     /// Determines which ephemeral tables were modified in this transaction.
     ///
     /// Iterates over the tables updated in this transaction and records those that
-    /// also appear in `all_ephemeral_tables`.  
+    /// also appear in `all_ephemeral_tables`.
     /// `self.ephemeral_tables` remains `None` if no ephemeral tables were modified.
     pub fn set_ephemeral_tables(&mut self, all_ephemeral_tables: &EphemeralTables) {
         for tid in self.tables.keys() {
@@ -252,6 +252,16 @@ impl TxData {
         self.ephemeral_tables.as_ref()
     }
 
+    /// Check if `table_id` is in the set of ephemeral tables for this transaction.
+    ///
+    /// Beware that ephemeral tables are known only after [Self::set_ephemeral_tables]
+    /// has been called.
+    pub fn is_ephemeral_table(&self, table_id: &TableId) -> bool {
+        self.ephemeral_tables
+            .as_ref()
+            .is_some_and(|etables| etables.contains(table_id))
+    }
+
     /// Obtain an iterator over the inserted rows per table.
     pub fn inserts(&self) -> impl Iterator<Item = (&TableId, &Arc<[ProductValue]>)> + '_ {
         self.inserts.iter()
@@ -304,12 +314,20 @@ impl TxData {
         self.truncates.iter().copied()
     }
 
-    /// Check if this [`TxData`] contains any `inserted | deleted` rows or `connect/disconnect` operations.
+    /// Check if this [`TxData`] contains any `inserted | deleted` rows
+    /// or `connect/disconnect` operations.
+    ///
+    /// Mutations of ephemeral tables are excluded, i.e. if the transaction
+    /// modifies only ephemeral tables (and is not a connect/disconnect operation),
+    /// the method returns `false`.
     ///
     /// This is used to determine if a transaction should be written to disk.
     pub fn has_rows_or_connect_disconnect(&self, reducer_context: Option<&ReducerContext>) -> bool {
-        self.inserts().any(|(_, inserted_rows)| !inserted_rows.is_empty())
-            || self.deletes().any(|(.., deleted_rows)| !deleted_rows.is_empty())
+        let is_non_ephemeral_mutation =
+            |(table_id, rows): (_, &Arc<[_]>)| !(self.is_ephemeral_table(table_id) || rows.is_empty());
+
+        self.inserts().any(is_non_ephemeral_mutation)
+            || self.deletes().any(is_non_ephemeral_mutation)
             || matches!(
                 reducer_context.map(|rcx| rcx.name.strip_prefix("__identity_")),
                 Some(Some("connected__" | "disconnected__"))