diff --git a/rust/src/database/postgres/write.rs b/rust/src/database/postgres/write.rs index 8be001c41b..1e0b19f638 100644 --- a/rust/src/database/postgres/write.rs +++ b/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>, { + // 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; +} + +/// 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(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 where F: for<'t> FnMut(&'t WriteTxn, &'t &'env ()) -> BoxFuture<'t, Result>, @@ -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 +where + F: for<'t> FnMut(&'t WriteTxn, &'t &'env ()) -> BoxFuture<'t, Result>, +{ let txn = WriteTxn::begin(conn).await?; match f(&txn, &&()).await { Ok(value) if commit => { diff --git a/rust/src/persist/effects.rs b/rust/src/persist/effects.rs index 3c917d2cbb..4e760574b9 100644 --- a/rust/src/persist/effects.rs +++ b/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:?}") } diff --git a/rust/src/persist/metrics.rs b/rust/src/persist/metrics.rs index 65b3602b84..a2d86ef00d 100644 --- a/rust/src/persist/metrics.rs +++ b/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, + /// What the write interactions of failed persists took, which no + /// result carries back to Python. + failed_txn_timings: Mutex, } 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 { diff --git a/rust/src/persist/python.rs b/rust/src/persist/python.rs index 9b5385d3e0..5b7b995b47 100644 --- a/rust/src/persist/python.rs +++ b/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> { 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>> { 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> { - 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(inner: &PersisterInner, result: PyResult, timings: &TxnTimings) -> PyResult { + 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 { 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 { 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, state_res_stats: Option, fallback_reason: Option, + /// What the update's write interactions took. + txn_timings: TxnTimings, } #[pymethods] @@ -1164,6 +1220,12 @@ impl CurrentStateUpdate { fn fallback_reason(&self) -> Option { 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> { + 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, remaining: Vec, fallback_reason: Option, + /// What the last step's write interactions took. + txn_timings: TxnTimings, } #[pymethods] @@ -1265,6 +1334,12 @@ impl PersistedBatch { fn fallback_reason(&self) -> Option { 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, /// What preparing counted, until it is recorded. counts: Mutex>, + /// 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> { diff --git a/synapse/storage/controllers/persist_events.py b/synapse/storage/controllers/persist_events.py index eadb572b16..02c2158a3e 100644 --- a/synapse/storage/controllers/persist_events.py +++ b/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", diff --git a/synapse/storage/databases/main/rust_persist.py b/synapse/storage/databases/main/rust_persist.py index f8019af000..1c386bc125 100644 --- a/synapse/storage/databases/main/rust_persist.py +++ b/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: diff --git a/synapse/synapse_rust/persist.pyi b/synapse/synapse_rust/persist.pyi index 7dbb0f810c..05980cab61 100644 --- a/synapse/synapse_rust/persist.pyi +++ b/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.""" diff --git a/tests/storage/test_rust_persist_prepare.py b/tests/storage/test_rust_persist_prepare.py index d8fecf034d..78343d124b 100644 --- a/tests/storage/test_rust_persist_prepare.py +++ b/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") diff --git a/tests/storage/test_rust_persist_prereqs.py b/tests/storage/test_rust_persist_prereqs.py index 850151c9e7..dd9bfe7b1e 100644 --- a/tests/storage/test_rust_persist_prereqs.py +++ b/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."""