Record the Rust persister's transactions as Python records its own

The native driver's write path recorded no `sql_txn_count` or
`sql_txn_duration`, nor the log context's transaction counts, so the
`persist_events` panels read near zero at `rust_event_persistence:
transaction` and `full`.

`run_write_interaction` now times each interaction as `new_transaction`
does: every attempt, from when it has its connection, so the wait for
one is left out. The times go into a task-local list that
`with_txn_timings` collects. The persister's entry points carry the list
back on what they already return (`PersistEffects`, `PreparedChunk`,
`PersistedBatch`, `CurrentStateUpdate`), and Python records it where
`new_transaction` would: the two counters, the pool's perf counters and
the current log context, which is the caller's when Python applies what
Rust returned. That costs no GIL hop.

When a persist fails, or Python gives up a batch, no result carries the
times back, so the persister keeps them and `flush_rust_persist_metrics`
counts them at the next scrape, without a log context. In shadow mode
Python's own interactions are the ones used, so Rust's dry run is not
recorded. The stream orderings' `nextval`, which runs outside a write
interaction, is recorded as `_load_next_mult_id`, as Python's is. The
`db.<desc>` tracing span is not recorded.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LXMkcZVkcZhi21augBff4a

rust/src/database/postgres/write.rs

@@ -30,8 +30,11 @@
 //! pipelines them on the connection.
 
 use std::{
+    cell::RefCell,
     fmt::Write as _,
+    future::Future,
     sync::atomic::{AtomicU8, Ordering},
+    time::{Duration, Instant},
 };
 
 use async_trait::async_trait;
@@ -493,23 +496,66 @@ async fn run_interaction<'env, T, F>(
 where
     F: for<'t> FnMut(&'t WriteTxn, &'t &'env ()) -> BoxFuture<'t, Result<T, WriteError>>,
 {
+    // As `new_transaction`, from when each attempt has its connection: the
+    // wait for one is not the transaction's.
+    let mut elapsed = Duration::ZERO;
     let mut attempt = 1;
-    loop {
-        let result = run_once(pool, commit, &mut f).await;
+    let result = loop {
+        let result = run_once(pool, commit, &mut f, &mut elapsed).await;
         match result {
             Err(error) if error.is_retryable() && attempt < MAX_ATTEMPTS => {
                 warn!("[TXN RETRY] {{{name}}} {error} {attempt}/{MAX_ATTEMPTS}");
                 attempt += 1;
             }
-            result => return result,
+            result => break result,
         }
-    }
+    };
+    record_txn_timing(name, elapsed);
+    result
 }
 
+/// What each write interaction run in a [`with_txn_timings`] scope took,
+/// as `(name, seconds)`, in the order they finished.
+pub type TxnTimings = Vec<(String, f64)>;
+
+tokio::task_local! {
+    static TXN_TIMINGS: RefCell<TxnTimings>;
+}
+
+/// Run `future`, and return with its output what each write interaction
+/// it ran took, whether it succeeded or not, retries included and the wait
+/// for a connection left out, as `DatabasePool.new_transaction` times
+/// them. A caller whose future fails must still report them. Python records them where it
+/// records its own (`sql_txn_count`, `sql_txn_duration` and the log
+/// context's transaction counts), which Rust cannot reach without the GIL.
+/// Interactions run on other tasks are not seen.
+pub async fn with_txn_timings<F: Future>(future: F) -> (F::Output, TxnTimings) {
+    TXN_TIMINGS
+        .scope(RefCell::new(Vec::new()), async {
+            let output = future.await;
+            let timings = TXN_TIMINGS.with(|timings| timings.take());
+            (output, timings)
+        })
+        .await
+}
+
+/// Record that an interaction run outside [`run_write_interaction`] took
+/// `elapsed`, if a [`with_txn_timings`] scope is listening.
+pub fn record_txn_timing(name: &str, elapsed: Duration) {
+    let _ = TXN_TIMINGS.try_with(|timings| {
+        timings
+            .borrow_mut()
+            .push((name.to_owned(), elapsed.as_secs_f64()))
+    });
+}
+
+/// One attempt, adding its time from when it has a connection to
+/// `elapsed`.
 async fn run_once<'env, T, F>(
     pool: &ConnectionPool,
     commit: bool,
     f: &mut F,
+    elapsed: &mut Duration,
 ) -> Result<T, WriteError>
 where
     F: for<'t> FnMut(&'t WriteTxn, &'t &'env ()) -> BoxFuture<'t, Result<T, WriteError>>,
@@ -520,6 +566,20 @@ where
         }
         e => WriteError::Other(anyhow::anyhow!("checking out a database connection: {e}")),
     })?;
+    let start = Instant::now();
+    let result = attempt(conn, commit, f).await;
+    *elapsed += start.elapsed();
+    result
+}
+
+async fn attempt<'env, T, F>(
+    conn: PooledConnection,
+    commit: bool,
+    f: &mut F,
+) -> Result<T, WriteError>
+where
+    F: for<'t> FnMut(&'t WriteTxn, &'t &'env ()) -> BoxFuture<'t, Result<T, WriteError>>,
+{
     let txn = WriteTxn::begin(conn).await?;
     match f(&txn, &&()).await {
         Ok(value) if commit => {

rust/src/persist/effects.rs

@@ -35,6 +35,8 @@
 
 use pyo3::prelude::*;
 
+use crate::database::postgres::write::TxnTimings;
+
 /// A stream change cache that a persist moves an entity forward in.
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub enum StreamChangeCache {
@@ -155,6 +157,10 @@ pub struct PersistEffects {
     /// Rows were written to the cache stream:
     /// `Notifier.on_new_replication_data()`.
     new_replication_data: bool,
+    /// What the write interactions behind these effects took, as `(name,
+    /// nanoseconds)`, for Python to record as it records its own. Whole
+    /// nanoseconds, as a float would cost the struct its `Eq`.
+    txn_timings: Vec<(String, u64)>,
 }
 
 impl PersistEffects {
@@ -229,6 +235,16 @@ impl PersistEffects {
         self.new_replication_data = true;
     }
 
+    /// Add what write interactions took, from
+    /// [`crate::database::postgres::write::with_txn_timings`].
+    pub fn add_txn_timings(&mut self, timings: TxnTimings) {
+        self.txn_timings.extend(
+            timings
+                .into_iter()
+                .map(|(name, secs)| (name, (secs * 1e9) as u64)),
+        );
+    }
+
     /// Start again, for a transaction that is retried.
     pub fn clear(&mut self) {
         *self = PersistEffects::default();
@@ -296,6 +312,7 @@ impl PersistEffects {
             stream_ids,
             backfilled,
             new_replication_data,
+            txn_timings: Vec::new(),
         })
     }
 
@@ -394,6 +411,15 @@ impl PersistEffects {
         self.new_replication_data
     }
 
+    /// `(name, seconds)` of each write interaction behind these effects.
+    #[getter]
+    fn txn_timings(&self) -> TxnTimings {
+        self.txn_timings
+            .iter()
+            .map(|(name, nanos)| (name.clone(), *nanos as f64 / 1e9))
+            .collect()
+    }
+
     fn __repr__(&self) -> String {
         format!("{self:?}")
     }

rust/src/persist/metrics.rs

@@ -27,6 +27,8 @@ use std::{
 
 use pyo3::{prelude::*, types::PyDict};
 
+use crate::database::postgres::write::TxnTimings;
+
 /// The counts since the last [`PersistMetrics::take`].
 #[derive(Clone, Debug, Default, PartialEq, Eq)]
 pub struct PersistCounts {
@@ -64,6 +66,9 @@ pub struct PersistCounts {
 #[derive(Default)]
 pub struct PersistMetrics {
     counts: Mutex<PersistCounts>,
+    /// What the write interactions of failed persists took, which no
+    /// result carries back to Python.
+    failed_txn_timings: Mutex<TxnTimings>,
 }
 
 impl PersistMetrics {
@@ -76,6 +81,26 @@ impl PersistMetrics {
     pub fn take(&self) -> PersistCounts {
         std::mem::take(&mut *self.counts.lock().unwrap_or_else(|e| e.into_inner()))
     }
+
+    /// Keep the timings of a failed persist's write interactions until
+    /// the metrics are next scraped.
+    pub fn record_failed_txn_timings(&self, timings: TxnTimings) {
+        self.failed_txn_timings
+            .lock()
+            .unwrap_or_else(|e| e.into_inner())
+            .extend(timings);
+    }
+
+    /// The timings [`Self::record_failed_txn_timings`] kept since the last
+    /// call.
+    pub fn take_failed_txn_timings(&self) -> TxnTimings {
+        std::mem::take(
+            &mut *self
+                .failed_txn_timings
+                .lock()
+                .unwrap_or_else(|e| e.into_inner()),
+        )
+    }
 }
 
 impl PersistCounts {

rust/src/persist/python.rs

@@ -41,7 +41,10 @@ use super::{
 use crate::{
     database::postgres::{
         pool::ConnectionPool,
-        write::{is_retryable_error, run_write_interaction, WriteError, MAX_ATTEMPTS},
+        write::{
+            is_retryable_error, record_txn_timing, run_write_interaction, with_txn_timings,
+            TxnTimings, WriteError, MAX_ATTEMPTS,
+        },
     },
     deferred::{create_deferred_with, OnCancel},
     errors::SynapseError,
@@ -196,7 +199,7 @@ impl Persister {
             let Prepared::Chunk(chunk) = &*prepared else {
                 unreachable!("checked above");
             };
-            write_chunk(
+            let (effects, timings) = with_txn_timings(write_chunk(
                 &inner,
                 &pool,
                 &room_id,
@@ -204,8 +207,11 @@ impl Persister {
                 chunk,
                 orderings,
                 inhibit_local_membership_updates,
-            )
-            .await
+            ))
+            .await;
+            let mut effects = reported(&inner, effects, &timings)?;
+            effects.add_txn_timings(timings);
+            Ok(effects)
         }
         .boxed())
     }
@@ -225,12 +231,14 @@ impl Persister {
         let pool = self.inner.pool.get()?;
         let inner = Arc::clone(&self.inner);
         Ok(async move {
-            let (prepared, counts) = prepare_chunk(&inner, &pool, &input, shadow)
-                .await
-                .map_err(persist_error_to_py)?;
+            let (prepared, txn_timings) =
+                with_txn_timings(prepare_chunk(&inner, &pool, &input, shadow)).await;
+            let (prepared, counts) =
+                reported(&inner, prepared.map_err(persist_error_to_py), &txn_timings)?;
             Ok(PreparedChunk {
                 prepared: Arc::new(prepared),
                 counts: Mutex::new(Some(counts)),
+                txn_timings,
             })
         }
         .boxed())
@@ -239,16 +247,37 @@ impl Persister {
     /// The future of the batch's next step. See [`Self::batch_next`].
     fn start_batch_next(&self, batch: &PersistBatch) -> BoxFuture<'static, PyResult<BatchStep>> {
         let state = Arc::clone(&batch.state);
-        async move { batch_next(&mut *state.lock().await).await }.boxed()
+        let inner = Arc::clone(&self.inner);
+        async move {
+            let (step, timings) = with_txn_timings(batch_next(&mut *state.lock().await)).await;
+            Ok(match reported(&inner, step, &timings)? {
+                BatchStep::Written(mut chunk) => {
+                    chunk.effects.add_txn_timings(timings);
+                    BatchStep::Written(chunk)
+                }
+                BatchStep::Done(mut batch) => {
+                    batch.txn_timings = timings;
+                    BatchStep::Done(batch)
+                }
+            })
+        }
+        .boxed()
     }
 
     /// The future that releases the batch. See [`Self::batch_abort`].
     fn start_batch_abort(&self, batch: &PersistBatch) -> BoxFuture<'static, PyResult<()>> {
         let state = Arc::clone(&batch.state);
+        let inner = Arc::clone(&self.inner);
         async move {
-            let mut state = state.lock().await;
-            state.done = true;
-            state.release_guard(true).await
+            let (released, timings) = with_txn_timings(async {
+                let mut state = state.lock().await;
+                state.done = true;
+                state.release_guard(true).await
+            })
+            .await;
+            // Nothing carries them back to Python, which gives up.
+            inner.metrics.record_failed_txn_timings(timings);
+            released
         }
         .boxed()
     }
@@ -260,7 +289,14 @@ impl Persister {
     ) -> PyResult<BoxFuture<'static, PyResult<CurrentStateUpdate>>> {
         let pool = self.inner.pool.get()?;
         let inner = Arc::clone(&self.inner);
-        Ok(async move { update_current_state(&inner, &pool, &room_id).await }.boxed())
+        Ok(async move {
+            let (update, timings) =
+                with_txn_timings(update_current_state(&inner, &pool, &room_id)).await;
+            let mut update = reported(&inner, update, &timings)?;
+            update.txn_timings = timings;
+            Ok(update)
+        }
+        .boxed())
     }
 
     /// Run `future` to completion on the runtime, releasing the GIL. For the
@@ -280,7 +316,9 @@ impl Persister {
     /// What persisting has counted since the last call, for
     /// `flush_rust_persist_metrics`.
     fn take_metrics<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyDict>> {
-        self.inner.metrics.take().to_py(py)
+        let dict = self.inner.metrics.take().to_py(py)?;
+        dict.set_item("txn_timings", self.inner.metrics.take_failed_txn_timings())?;
+        Ok(dict)
     }
 
     /// What `_persist_event_batch` computes for a chunk of one room's events
@@ -526,6 +564,15 @@ impl Persister {
     }
 }
 
+/// `result`, keeping `timings` for the next metrics scrape if it is an
+/// error, as no result carries them back to Python then.
+fn reported<T>(inner: &PersisterInner, result: PyResult<T>, timings: &TxnTimings) -> PyResult<T> {
+    if result.is_err() {
+        inner.metrics.record_failed_txn_timings(timings.clone());
+    }
+    result
+}
+
 /// The events of `events_and_contexts`, sharing their data, internal
 /// metadata included, with the Python events, so that Rust can read them
 /// without the GIL and set their stream orderings.
@@ -826,6 +873,7 @@ async fn batch_step(state: &mut BatchState) -> PyResult<BatchStep> {
             replaced_events: deduplicated.replaced_events.clone(),
             remaining: Vec::new(),
             fallback_reason: None,
+            txn_timings: Vec::new(),
         }));
     }
     let end = (start + batch::CHUNK_SIZE).min(deduplicated.kept.len());
@@ -836,6 +884,7 @@ async fn batch_step(state: &mut BatchState) -> PyResult<BatchStep> {
             replaced_events: deduplicated.replaced_events.clone(),
             remaining: deduplicated.kept[start..].to_vec(),
             fallback_reason: Some(reason),
+            txn_timings: Vec::new(),
         })
     };
 
@@ -987,6 +1036,8 @@ impl WrittenChunk {
             prepared: Arc::clone(&self.prepared),
             // Counted when it was prepared.
             counts: Mutex::new(None),
+            // Recorded with the chunk's effects.
+            txn_timings: Vec::new(),
         }
     }
 
@@ -1060,6 +1111,7 @@ async fn update_current_state(
                     effects: None,
                     state_res_stats: None,
                     fallback_reason: Some(fallback.0),
+                    txn_timings: Vec::new(),
                 }),
                 Err(e) => Err(persist_error_to_py(e)),
             };
@@ -1071,6 +1123,7 @@ async fn update_current_state(
             effects: None,
             state_res_stats: stats,
             fallback_reason: None,
+            txn_timings: Vec::new(),
         });
     }
 
@@ -1131,6 +1184,7 @@ async fn update_current_state(
         effects: Some(effects),
         state_res_stats: stats,
         fallback_reason: None,
+        txn_timings: Vec::new(),
     })
 }
 
@@ -1140,6 +1194,8 @@ pub struct CurrentStateUpdate {
     effects: Option<PersistEffects>,
     state_res_stats: Option<StateResStats>,
     fallback_reason: Option<String>,
+    /// What the update's write interactions took.
+    txn_timings: TxnTimings,
 }
 
 #[pymethods]
@@ -1164,6 +1220,12 @@ impl CurrentStateUpdate {
     fn fallback_reason(&self) -> Option<String> {
         self.fallback_reason.clone()
     }
+
+    /// `(name, seconds)` of each write interaction the update ran.
+    #[getter]
+    fn txn_timings(&self) -> TxnTimings {
+        self.txn_timings.clone()
+    }
 }
 
 /// Where the stream orderings of a chunk come from.
@@ -1183,6 +1245,7 @@ async fn allocate_stream_orderings(
     stream: &TxnStream,
     n: usize,
 ) -> anyhow::Result<Vec<i64>> {
+    let start = std::time::Instant::now();
     let mut attempt = 1;
     loop {
         let result = async {
@@ -1190,6 +1253,10 @@ async fn allocate_stream_orderings(
             stream.next_ids_outside_txn(&querier, n).await
         }
         .await;
+        // As Python's `runInteraction`, retries included.
+        if !matches!(&result, Err(e) if attempt < MAX_ATTEMPTS && is_retryable_error(e)) {
+            record_txn_timing("_load_next_mult_id", start.elapsed());
+        }
         match result {
             Err(e) if attempt < MAX_ATTEMPTS && is_retryable_error(&e) => {
                 log::warn!("[TXN RETRY] {{_load_next_mult_id}} {e:#} {attempt}/{MAX_ATTEMPTS}");
@@ -1241,6 +1308,8 @@ pub struct PersistedBatch {
     replaced_events: HashMap<String, String>,
     remaining: Vec<usize>,
     fallback_reason: Option<String>,
+    /// What the last step's write interactions took.
+    txn_timings: TxnTimings,
 }
 
 #[pymethods]
@@ -1265,6 +1334,12 @@ impl PersistedBatch {
     fn fallback_reason(&self) -> Option<String> {
         self.fallback_reason.clone()
     }
+
+    /// `(name, seconds)` of each write interaction the last step ran.
+    #[getter]
+    fn txn_timings(&self) -> TxnTimings {
+        self.txn_timings.clone()
+    }
 }
 
 /// What [`Persister::prepare_chunk`] computed, in the shapes Python builds
@@ -1276,6 +1351,8 @@ pub struct PreparedChunk {
     prepared: Arc<Prepared>,
     /// What preparing counted, until it is recorded.
     counts: Mutex<Option<PersistCounts>>,
+    /// What preparing's write interactions took.
+    txn_timings: TxnTimings,
 }
 
 impl PreparedChunk {
@@ -1289,6 +1366,12 @@ impl PreparedChunk {
 
 #[pymethods]
 impl PreparedChunk {
+    /// `(name, seconds)` of each write interaction preparing ran.
+    #[getter]
+    fn txn_timings(&self) -> TxnTimings {
+        self.txn_timings.clone()
+    }
+
     /// Why the Rust state resolution could not resolve the room.
     #[getter]
     fn state_res_fallback(&self) -> Option<&str> {

synapse/storage/controllers/persist_events.py

@@ -68,6 +68,7 @@ from synapse.logging.opentracing import (
 )
 from synapse.metrics import SERVER_NAME_LABEL, pre_collect_hooks
 from synapse.storage.controllers.state import StateStorageController
+from synapse.storage.database import sql_txn_count, sql_txn_duration
 from synapse.storage.databases import Databases
 from synapse.storage.databases.main.events import (
     DeltaState,
@@ -85,6 +86,7 @@ from synapse.storage.databases.main.events_worker import EventRedactBehaviour
 from synapse.storage.databases.main.rust_persist import (
     apply_rust_persist_effects,
     finish_rust_persist_stream_ids,
+    record_rust_txn_timings,
 )
 from synapse.synapse_rust.persist import (
     PersistBatch,
@@ -426,6 +428,11 @@ def flush_rust_persist_metrics(persister: Persister, server_name: str) -> None:
             for _ in range(count):
                 child.observe(value)
 
+    # Those of failed persists, which no result carried back.
+    for desc, duration in counts["txn_timings"]:
+        sql_txn_count.labels(desc=desc, **labels).inc(1)
+        sql_txn_duration.labels(desc=desc, **labels).inc(duration)
+
     if counts["persisted_position"] is not None:
         synapse.metrics.event_persisted_position.labels(**labels).set(
             counts["persisted_position"]
@@ -831,6 +838,7 @@ class EventsPersistenceStorageController:
                 # replication, as `get_next` does when its block raises.
                 self.hs.get_replication_notifier().notify_replication()
                 raise
+            record_rust_txn_timings(self.persist_events_store.store, update.txn_timings)
             # Recorded first, as Python records them before it writes.
             if update.state_res_stats is not None:
                 self._state_resolution_handler.record_rust_state_res_metrics(
@@ -1038,6 +1046,9 @@ class EventsPersistenceStorageController:
                     self.hs.get_replication_notifier().notify_replication()
                     raise
                 if not isinstance(step, WrittenChunk):
+                    record_rust_txn_timings(
+                        self.persist_events_store.store, step.txn_timings
+                    )
                     break
                 await self._apply_rust_batch_chunk(
                     room_id, events_and_contexts, backfilled, step, handle
@@ -1252,6 +1263,9 @@ class EventsPersistenceStorageController:
             rust = await self._rust_persister.prepare_chunk(
                 room_id, chunk, backfilled, shadow=shadow
             )
+        if not shadow:
+            # In shadow mode, Python's own interactions are the ones used.
+            record_rust_txn_timings(self.persist_events_store.store, rust.txn_timings)
         if rust.state_res_fallback is not None:
             logger.info(
                 "Preparing events for %s in Python: %s",

synapse/storage/databases/main/rust_persist.py

@@ -14,9 +14,12 @@
 
 """Applying what a Rust persist transaction asks for once it commits."""
 
-from collections.abc import Mapping
+from collections.abc import Iterable, Mapping
 from typing import TYPE_CHECKING
 
+from synapse.logging.context import current_context
+from synapse.metrics import SERVER_NAME_LABEL
+from synapse.storage.database import sql_txn_count, sql_txn_duration
 from synapse.storage.util.id_generators import MultiWriterIdGenerator
 from synapse.synapse_rust.persist import PersistEffects
 from synapse.util.caches.stream_change_cache import StreamChangeCache
@@ -35,7 +38,8 @@ def apply_rust_persist_effects(
 ) -> None:
     """Apply what a Rust persist transaction asks for once it has
     committed: the cache updates the Python transaction queues with
-    `txn.call_after`. See `synapse_rust.persist.PersistEffects`.
+    `txn.call_after`. See `synapse_rust.persist.PersistEffects`. First,
+    record what its write interactions took.
     `store` is the main store of the event persister. On a worker that is a
     `GenericWorkerStore`, which has every attribute used here.
     `events` must hold the events `effects.event_cache_prefills` names.
@@ -58,6 +62,7 @@ def apply_rust_persist_effects(
     caller, who must call `finish_rust_persist_stream_ids` whatever
     happens.
     """
+    record_rust_txn_timings(store, effects.txn_timings)
     cache_ids_finished = profile_ids_finished = sticky_ids_finished = False
     # Left unfinished on purpose, without `finish_stream_ids`.
     event_ids_finished = not finish_stream_ids
@@ -170,6 +175,24 @@ def finish_rust_persist_stream_ids(
         store.hs.get_replication_notifier().notify_replication()
 
 
+def record_rust_txn_timings(
+    store: "DataStore", timings: Iterable[tuple[str, float]]
+) -> None:
+    """Record what Rust's write interactions took as
+    `DatabasePool.new_transaction` records its own: in the transaction
+    metrics, the pool's perf counters and the current log context, which
+    is the caller's when Python applies what Rust returned."""
+    db_pool = store.db_pool
+    labels = {SERVER_NAME_LABEL: db_pool.server_name}
+    context = current_context()
+    for desc, duration in timings:
+        context.add_database_transaction(duration)
+        db_pool._current_txn_total_time += duration
+        db_pool._txn_perf_counters.update(desc, duration)
+        sql_txn_count.labels(desc=desc, **labels).inc(1)
+        sql_txn_duration.labels(desc=desc, **labels).inc(duration)
+
+
 def _finish_txn_stream_ids(
     store: "DataStore", id_gen: MultiWriterIdGenerator, stream_ids: list[int]
 ) -> None:

synapse/synapse_rust/persist.pyi

@@ -163,6 +163,9 @@ class CurrentStateUpdate:
         """Why the Rust state resolution could not resolve the room, in
         which case nothing was written and Python should recalculate the
         state."""
+    @property
+    def txn_timings(self) -> list[tuple[str, float]]:
+        """`(name, seconds)` of each write interaction the update ran."""
 
 class PersistBatch:
     """A batch being persisted. See `Persister.persist_batch`."""
@@ -200,11 +203,17 @@ class PersistedBatch:
     @property
     def fallback_reason(self) -> str | None:
         """Why Rust left the `remaining` events."""
+    @property
+    def txn_timings(self) -> list[tuple[str, float]]:
+        """`(name, seconds)` of each write interaction the last step ran."""
 
 class PreparedChunk:
     """What `Persister.prepare_chunk` computed. If `state_res_fallback` is
     set, the rest is empty and Python should compute the chunk itself."""
 
+    @property
+    def txn_timings(self) -> list[tuple[str, float]]:
+        """`(name, seconds)` of each write interaction preparing ran."""
     @property
     def state_res_fallback(self) -> str | None:
         """Why the Rust state resolution could not resolve the room."""
@@ -343,6 +352,10 @@ class PersistEffects:
         """Whether `stream_ids` are on the backfill stream."""
 
     @property
+    def txn_timings(self) -> list[tuple[str, float]]:
+        """`(name, seconds)` of each write interaction behind these effects,
+        to record as `DatabasePool.new_transaction` records its own."""
+    @property
     def new_replication_data(self) -> bool:
         """Whether rows were written to the cache invalidation stream."""
 

tests/storage/test_rust_persist_prepare.py

@@ -133,6 +133,7 @@ class _PrepareTestBase(unittest.HomeserverTestCase):
                     replaced_events={},
                     remaining=list(range(handle.events)),
                     fallback_reason="mocked",
+                    txn_timings=[],
                 )
             )
 
@@ -348,6 +349,17 @@ class ShadowTestCase(_PrepareTestBase):
             self._send("hello")
         self.assertEqual(self._counted(), {"error": 1})
 
+    def test_txn_timings_not_recorded(self) -> None:
+        """Python's own interactions are the ones used in shadow mode, so
+        Rust's dry run is not recorded beside them."""
+        with patch(
+            "synapse.storage.controllers.persist_events.record_rust_txn_timings"
+        ) as record:
+            self.helper.send_state(
+                self.room_id, EventTypes.Name, {"name": "x"}, tok=self.alice_tok
+            )
+        record.assert_not_called()
+
 
 class PrepareTestCase(_PrepareTestBase):
     level: RustEventPersistenceLevel = "prepare"
@@ -460,7 +472,9 @@ class PrepareTestCase(_PrepareTestBase):
     def test_fallback(self) -> None:
         self.persistence._rust_persister = self._mock_persister(
             prepare_chunk=Mock(
-                return_value=defer.succeed(Mock(state_res_fallback="no index"))
+                return_value=defer.succeed(
+                    Mock(state_res_fallback="no index", txn_timings=[])
+                )
             )
         )
         event_id = self._send("hello")

tests/storage/test_rust_persist_prereqs.py

@@ -765,6 +765,7 @@ class FlushRustPersistMetricsTestCase(unittest.TestCase):
             "stale_forward_extremities": {},
             "state_dag_forward_extremities": {},
             "persisted_position": 42,
+            "txn_timings": [("persist_events", 0.5)],
         }
         persister = Mock(spec=["take_metrics"])
         persister.take_metrics.return_value = counts
@@ -797,6 +798,19 @@ class FlushRustPersistMetricsTestCase(unittest.TestCase):
             5,
         )
         self.assertEqual(self._sample("synapse_event_persisted_position"), 42)
+        # The interactions of failed persists.
+        self.assertEqual(
+            self._sample(
+                "synapse_storage_transaction_time_count_total", desc="persist_events"
+            ),
+            1,
+        )
+        self.assertEqual(
+            self._sample(
+                "synapse_storage_transaction_time_sum_total", desc="persist_events"
+            ),
+            0.5,
+        )
 
         # A position that did not move is not set again, so the Python path
         # can move the gauge on without being pulled back.
@@ -806,6 +820,7 @@ class FlushRustPersistMetricsTestCase(unittest.TestCase):
         persister.take_metrics.return_value = {
             **counts,
             "persisted_position": None,
+            "txn_timings": [],
         }
         flush_rust_persist_metrics(persister, self.server_name)
         self.assertEqual(self._sample("synapse_event_persisted_position"), 50)
@@ -919,6 +934,28 @@ class _CountingPersister:
         return getattr(self._persister, name)
 
 
+def _txn_count(desc: str) -> float:
+    """`synapse_storage_transaction_time_count` for `desc`."""
+    return (
+        REGISTRY.get_sample_value(
+            "synapse_storage_transaction_time_count_total",
+            {"desc": desc, SERVER_NAME_LABEL: "test"},
+        )
+        or 0.0
+    )
+
+
+def _txn_secs(desc: str) -> float:
+    """`synapse_storage_transaction_time_sum` for `desc`."""
+    return (
+        REGISTRY.get_sample_value(
+            "synapse_storage_transaction_time_sum_total",
+            {"desc": desc, SERVER_NAME_LABEL: "test"},
+        )
+        or 0.0
+    )
+
+
 class RustTransactionTestCase(unittest.HomeserverTestCase):
     """`rust_event_persistence: transaction`: Rust writes the chunks it
     prepared."""
@@ -1019,6 +1056,13 @@ class RustTransactionTestCase(unittest.HomeserverTestCase):
             max(effects.profile_stream_ids),
         )
 
+    def test_txn_metrics(self) -> None:
+        """The Rust transaction is counted as Python's would be."""
+        room_id = self.helper.create_room_as(self.user, tok=self.token)
+        before = _txn_count("persist_events")
+        self.helper.send(room_id, "hello", tok=self.token)
+        self.assertEqual(_txn_count("persist_events"), before + 1)
+
     def test_gil_count(self) -> None:
         """Writing a chunk takes the GIL from tokio at no persist site: the
         events cross at the call and the effects come back in its result."""
@@ -1762,6 +1806,68 @@ class RustFullTestCase(unittest.HomeserverTestCase):
             self.store._stream_id_gen.get_current_token(), stream_id
         )
 
+    def test_txn_metrics(self) -> None:
+        """Each Rust interaction of a batch is counted as Python's would
+        be: the orderings, the state group guard and the write."""
+        names = (
+            "_load_next_mult_id",
+            "mark_state_groups_as_persisting",
+            "persist_events",
+            "finish_persisting",
+        )
+        before = {name: _txn_count(name) for name in names}
+        secs = _txn_secs("persist_events")
+        self.helper.send(self.room_id, "hello", tok=self.token)
+        self.assertEqual(
+            {name: _txn_count(name) - before[name] for name in names},
+            dict.fromkeys(names, 1.0),
+        )
+        self.assertGreater(_txn_secs("persist_events"), secs)
+
+        # A batch of two chunks: two writes, and so on.
+        before = {name: _txn_count(name) for name in names}
+        pairs = [self._build(f"message {i}") for i in range(150)]
+        self.get_success(self.persistence.persist_events(pairs))
+        self.assertEqual(
+            {name: _txn_count(name) - before[name] for name in names},
+            dict.fromkeys(names, 2.0),
+        )
+
+        # The update of a room's current state.
+        before_update = _txn_count("update_current_state")
+        self.get_success(
+            self.store.db_pool.simple_delete(
+                "current_state_events",
+                {"room_id": self.room_id, "type": EventTypes.Name},
+                desc="test",
+            )
+        )
+        self.helper.send_state(
+            self.room_id, EventTypes.Name, {"name": "n"}, tok=self.token
+        )
+        self.get_success(
+            self.store.db_pool.simple_delete(
+                "current_state_events",
+                {"room_id": self.room_id, "type": EventTypes.Name},
+                desc="test",
+            )
+        )
+        self.get_success(self.persistence.update_current_state(self.room_id))
+        self.assertEqual(_txn_count("update_current_state"), before_update + 1)
+
+    def test_failed_txn_metrics(self) -> None:
+        """The interactions of a persist that fails are counted when the
+        metrics are next collected."""
+        event, context = self._build("partial")
+        context.partial_state = True
+        before = _txn_count("persist_events")
+        self.get_failure(
+            self.persistence.persist_events([(event, context)]),
+            PartialStateConflictError,
+        )
+        self.persistence._flush_rust_persist_metrics()
+        self.assertEqual(_txn_count("persist_events"), before + 1)
+
     def test_gil_count(self) -> None:
         """Taking the stream orderings and setting them on the events takes
         the GIL at no persist site."""