commit_and_broadcast_event: read_tx.merge(update_metrics)

crates/core/src/db/datastore/locking_tx_datastore/datastore.rs

@@ -858,8 +858,8 @@ pub fn report_tx_metricses(
     reducer: &str,
     db: &RelationalDB,
     tx_data: Option<&TxData>,
-    metrics_mut: Option<TxMetrics>,
-    metrics_read: TxMetrics,
+    metrics_mut: Option<&TxMetrics>,
+    metrics_read: &TxMetrics,
 ) {
     if let Some(metrics_mut) = metrics_mut {
         metrics_mut.report_with_db(reducer, db, tx_data);

crates/core/src/db/relational_db.rs

@@ -972,7 +972,7 @@ impl RelationalDB {
         let mut tx = self.begin_tx(workload);
         let res = f(&mut tx);
         let (tx_metics, reducer) = self.release_tx(tx);
-        report_tx_metricses(&reducer, self, None, None, tx_metics);
+        report_tx_metricses(&reducer, self, None, None, &tx_metics);
         res
     }
 

crates/core/src/sql/execute.rs

@@ -174,6 +174,8 @@ pub fn execute_sql_tx<'a>(
 
 pub struct SqlResult {
     pub rows: Vec<ProductValue>,
+    /// These metrics will be reported via `report_tx_metrics`.
+    /// They should not be reported separately to avoid double counting.
     pub metrics: ExecutionMetrics,
 }
 
@@ -199,10 +201,16 @@ pub fn run(
             // and hence there are no deltas to process.
             let (tx_data, tx_metrics_mut, tx) = tx.commit_downgrade(Workload::Sql);
 
-            // Release the tx on drop, so that we record metrics
+            // Release the tx on drop, so that we record metrics.
             let mut tx = scopeguard::guard(tx, |tx| {
                 let (tx_metrics_downgrade, reducer) = db.release_tx(tx);
-                report_tx_metricses(&reducer, db, Some(&tx_data), Some(tx_metrics_mut), tx_metrics_downgrade);
+                report_tx_metricses(
+                    &reducer,
+                    db,
+                    Some(&tx_data),
+                    Some(&tx_metrics_mut),
+                    &tx_metrics_downgrade,
+                );
             });
 
             // Compute the header for the result set

crates/core/src/subscription/module_subscription_actor.rs

@@ -644,6 +644,9 @@ impl ModuleSubscriptions {
     }
 
     /// Commit a transaction and broadcast its ModuleEvent to all interested subscribers.
+    ///
+    /// The returned [`ExecutionMetrics`] are reported in this method via `report_tx_metrics`.
+    /// They are returned for testing purposes but should not be reported separately.
     pub fn commit_and_broadcast_event(
         &self,
         caller: Option<&ClientConnectionSender>,
@@ -655,7 +658,7 @@ impl ModuleSubscriptions {
         let subscriptions = self.subscriptions.read();
         let stdb = &self.relational_db;
         // Downgrade mutable tx.
-        // Ensure tx is released/cleaned up once out of scope.
+        // 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 {
@@ -670,22 +673,29 @@ impl ModuleSubscriptions {
             }
         };
 
-        let read_tx = scopeguard::guard(read_tx, |tx| {
+        // When we're done with this method, release the tx and report metrics.
+        let mut read_tx = scopeguard::guard(read_tx, |tx| {
             let (tx_metrics_read, reducer) = self.relational_db.release_tx(tx);
-            report_tx_metricses(&reducer, stdb, tx_data.as_ref(), Some(tx_metrics_mut), tx_metrics_read);
+            report_tx_metricses(
+                &reducer,
+                stdb,
+                tx_data.as_ref(),
+                Some(&tx_metrics_mut),
+                &tx_metrics_read,
+            );
         });
-
-        let read_tx = tx_data
+        // Create the delta transaction we'll use to eval updates against.
+        let delta_read_tx = tx_data
             .as_ref()
             .map(|tx_data| DeltaTx::new(&read_tx, tx_data))
             .unwrap_or_else(|| DeltaTx::from(&*read_tx));
 
         let event = Arc::new(event);
-        let mut metrics = ExecutionMetrics::default();
+        let mut update_metrics: ExecutionMetrics = ExecutionMetrics::default();
 
         match &event.status {
             EventStatus::Committed(_) => {
-                metrics.merge(subscriptions.eval_updates(&read_tx, event.clone(), caller));
+                update_metrics = subscriptions.eval_updates(&delta_read_tx, event.clone(), caller);
             }
             EventStatus::Failed(_) => {
                 if let Some(client) = caller {
@@ -701,7 +711,10 @@ impl ModuleSubscriptions {
             EventStatus::OutOfEnergy => {} // ?
         }
 
-        Ok(Ok((event, metrics)))
+        // Merge in the subscription evaluation metrics.
+        read_tx.metrics.merge(update_metrics);
+
+        Ok(Ok((event, update_metrics)))
     }
 }