From 68cc9e1b64a1e323e4ffc33770a436be8731e25a Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Sat, 17 Oct 2026 21:42:13 +0000 Subject: [PATCH] Take the event stream orderings in Rust at rust_event_persistence: full At `full`, `Persister.persist_chunk` takes the chunk's stream orderings from the events or backfill stream tracker, which Python's `_stream_id_gen` and `_backfill_id_gen` share. It runs `nextval` in autocommit after prepare, as late as `get_next_mult` does, and sets each event's stream ordering, received time and instance before the write transaction. Backfilled events take theirs negated. The orderings come back unfinished in `PersistEffects.stream_ids`. `_persist_events_and_state_updates` skips `get_next_mult` when Rust takes them. Its post-transaction block (metrics, MSC4293 `room_ban_redactions`, the extremities prefill) moves to `_after_persisting_events`, which both paths call. At `full` it runs after the effects are applied and before the orderings are finished, as it ran inside `get_next_mult`. So a ban that redacts a user's events is recorded before other workers see it. `finish_rust_persist_stream_ids` finishes the orderings as `get_next_mult` does on exit: it tells replication, and on success moves `stream_positions` on. If the transaction fails, Rust finishes them. If the apply or the follow-ups raise, Python finishes them. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LXMkcZVkcZhi21augBff4a --- rust/src/persist/python.rs | 163 +++++++- rust/src/persist/txn/streams.rs | 66 ++- synapse/config/experimental.py | 4 +- synapse/storage/controllers/persist_events.py | 16 +- synapse/storage/databases/main/events.py | 302 +++++++++----- .../storage/databases/main/rust_persist.py | 42 +- synapse/synapse_rust/persist.pyi | 25 ++ tests/storage/test_rust_persist_prereqs.py | 390 +++++++++++++++++- 8 files changed, 850 insertions(+), 158 deletions(-) diff --git a/rust/src/persist/python.rs b/rust/src/persist/python.rs index f8e33d46df..b565f9d9f3 100644 --- a/rust/src/persist/python.rs +++ b/rust/src/persist/python.rs @@ -36,8 +36,9 @@ use super::{ txn::{self, json, streams::TxnStream, PartialStateConflict, TxnEvent, TxnInput}, }; use crate::{ - database::postgres::write::{ - is_retryable_error, run_write_interaction, WriteError, MAX_ATTEMPTS, + database::postgres::{ + pool::ConnectionPool, + write::{is_retryable_error, run_write_interaction, WriteError, MAX_ATTEMPTS}, }, deferred::{create_deferred_with, OnCancel}, errors::SynapseError, @@ -48,6 +49,7 @@ use crate::{ state::{ConflictCache, StateResStats}, storage::{ native_pool::NativePool, + pool_querier::PoolQuerier, store::Store, stream_ids::{PyStreamIdTracker, StreamIdTracker}, sync_store::{SyncStore, SyncStoreConfig}, @@ -170,30 +172,55 @@ impl Persister { /// Convert the chunk while the GIL is held, and return the future that /// runs the write transaction over it without the GIL. See - /// [`Self::persist_events_txn`]. + /// [`Self::persist_events_txn`] and [`Self::persist_chunk`]. fn start_persist_events_txn( &self, py: Python<'_>, room_id: String, events_and_contexts: Vec<(Py, PersistContext)>, prepared: &PreparedChunk, + orderings: StreamOrderings, inhibit_local_membership_updates: bool, ) -> PyResult>> { let prepared = Arc::clone(&prepared.prepared); if !matches!(&*prepared, Prepared::Chunk(_)) { return Err(PyRuntimeError::new_err("the chunk was prepared in Python")); } - let events = events_and_contexts + let events: Vec<(Event, PersistContext)> = events_and_contexts .into_iter() - .map(|(event, context)| TxnEvent::new(event.bind(py).get().shallow_clone(), context)) - .collect::>>() - .map_err(persist_error_to_py)?; + .map(|(event, context)| (event.bind(py).get().shallow_clone(), context)) + .collect(); let pool = self.inner.pool.get()?; let inner = Arc::clone(&self.inner); let future = async move { let Prepared::Chunk(chunk) = &*prepared else { unreachable!("checked above"); }; + // Taken as late as possible, as Python takes them: until the + // chunk is written, no later event on the stream can be + // announced. + let event_stream = match orderings { + StreamOrderings::Set => None, + StreamOrderings::Allocate { backfilled } => { + let stream = if backfilled { + TxnStream::backfill(Arc::clone(&inner.backfill_stream)) + } else { + TxnStream::events(Arc::clone(&inner.events_stream)) + }; + let ids = allocate_stream_orderings(&pool, &stream, events.len()) + .await + .map_err(persist_error_to_py)?; + set_persisted_positions(&inner, &events, &ids, backfilled) + .map_err(persist_error_to_py)?; + Some((stream, ids, backfilled)) + } + }; + let events = events + .into_iter() + .map(|(event, context)| TxnEvent::new(event, context)) + .collect::>>() + .map_err(persist_error_to_py)?; + let cache_stream = TxnStream::caches(Arc::clone(&inner.cache_stream)); let sticky_stream = TxnStream::sticky_events(Arc::clone(&inner.sticky_stream)); let profile_stream = TxnStream::profile_updates(Arc::clone(&inner.profile_stream)); @@ -213,7 +240,7 @@ impl Persister { sticky_stream: &sticky_stream, profile_stream: &profile_stream, }; - let effects = run_write_interaction(&pool, "persist_events", |txn, _| { + let mut effects = run_write_interaction(&pool, "persist_events", |txn, _| { let input = &input; async move { let mut effects = PersistEffects::default(); @@ -227,6 +254,12 @@ impl Persister { cache_stream.hand_over(effects.cache_stream_ids()); sticky_stream.hand_over(effects.sticky_stream_ids()); profile_stream.hand_over(effects.profile_stream_ids()); + // Finished on drop if the transaction failed, as `get_next_mult` + // finishes them when its block raises. + if let Some((stream, ids, backfilled)) = event_stream { + stream.hand_over(&ids); + effects.add_stream_ids(ids, backfilled); + } Ok(effects) }; Ok(future.boxed()) @@ -372,11 +405,61 @@ impl Persister { room_id, events_and_contexts, prepared, + StreamOrderings::Set, inhibit_local_membership_updates, )?; create_deferred_with(py, &self.inner.runtime, future, OnCancel::RunOn) } + /// As [`Self::persist_events_txn`], but Rust takes the chunk's stream + /// orderings, from the backfill stream if `backfilled`, and sets them on + /// the events with the time and this instance, as + /// `_persist_events_and_state_updates` does. They come back in the + /// effects' `stream_ids`, unfinished. If the persist fails, Rust + /// finishes them. + #[pyo3(signature = (room_id, events_and_contexts, prepared, *, backfilled, inhibit_local_membership_updates))] + fn persist_chunk<'py>( + &self, + py: Python<'py>, + room_id: String, + events_and_contexts: Vec<(Py, PersistContext)>, + prepared: &PreparedChunk, + backfilled: bool, + inhibit_local_membership_updates: bool, + ) -> PyResult> { + let future = self.start_persist_events_txn( + py, + room_id, + events_and_contexts, + prepared, + StreamOrderings::Allocate { backfilled }, + inhibit_local_membership_updates, + )?; + create_deferred_with(py, &self.inner.runtime, future, OnCancel::RunOn) + } + + /// As [`Self::persist_chunk`], but blocks. + #[pyo3(signature = (room_id, events_and_contexts, prepared, *, backfilled, inhibit_local_membership_updates))] + fn persist_chunk_blocking( + &self, + py: Python<'_>, + room_id: String, + events_and_contexts: Vec<(Py, PersistContext)>, + prepared: &PreparedChunk, + backfilled: bool, + inhibit_local_membership_updates: bool, + ) -> PyResult { + let future = self.start_persist_events_txn( + py, + room_id, + events_and_contexts, + prepared, + StreamOrderings::Allocate { backfilled }, + inhibit_local_membership_updates, + )?; + self.block_on(py, future) + } + /// As [`Self::persist_events_txn`], but blocks. #[pyo3(signature = (room_id, events_and_contexts, prepared, *, inhibit_local_membership_updates))] fn persist_events_txn_blocking( @@ -392,6 +475,7 @@ impl Persister { room_id, events_and_contexts, prepared, + StreamOrderings::Set, inhibit_local_membership_updates, )?; self.block_on(py, future) @@ -413,6 +497,69 @@ impl Persister { } } +/// Where the stream orderings of a chunk come from. +#[derive(Clone, Copy)] +enum StreamOrderings { + /// Python set them on the events. + Set, + /// Rust takes them from the events stream, or from the backfill stream + /// if the chunk is backfilled. + Allocate { backfilled: bool }, +} + +/// `n` IDs from `stream`, unfinished, retried as Python's `runInteraction` +/// retries the fetch. +async fn allocate_stream_orderings( + pool: &ConnectionPool, + stream: &TxnStream, + n: usize, +) -> anyhow::Result> { + let mut attempt = 1; + loop { + let result = async { + let querier = PoolQuerier::checkout(pool).await?; + stream.next_ids_outside_txn(&querier, n).await + } + .await; + match result { + Err(e) if attempt < MAX_ATTEMPTS && is_retryable_error(&e) => { + log::warn!("[TXN RETRY] {{_load_next_mult_id}} {e:#} {attempt}/{MAX_ATTEMPTS}"); + attempt += 1; + } + result => return result, + } + } +} + +/// Set each event's stream ordering from `ids`, in order, with the time +/// and this instance, as `_persist_events_and_state_updates` does once it +/// has its stream orderings. Backfilled events take theirs negated. An +/// event that appears twice keeps the later ordering, as in Python. +fn set_persisted_positions( + inner: &PersisterInner, + events: &[(Event, PersistContext)], + ids: &[i64], + backfilled: bool, +) -> anyhow::Result<()> { + if ids.len() != events.len() { + anyhow::bail!( + "took {} stream orderings for {} events", + ids.len(), + events.len() + ); + } + let received_ts = i64::try_from(inner.runtime.clock().now_millis())?; + for ((event, _), &id) in events.iter().zip(ids) { + let stream_ordering = if backfilled { -id } else { id }; + event.internal_metadata().set_persisted_position( + Some(stream_ordering), + Some(received_ts), + Some(inner.instance_name.clone()), + )?; + } + Ok(()) +} + /// What [`Persister::prepare_chunk`] computed, in the shapes Python builds /// its own types from. If `state_res_fallback` is set, the rest is empty /// and Python should compute the chunk itself. diff --git a/rust/src/persist/txn/streams.rs b/rust/src/persist/txn/streams.rs index 4c83c85bde..0cac138562 100644 --- a/rust/src/persist/txn/streams.rs +++ b/rust/src/persist/txn/streams.rs @@ -13,9 +13,11 @@ * */ -//! The streams a persist takes IDs from inside its transaction, as -//! Python's `get_next_txn` and `get_next_mult_txn` do: the cache stream, -//! the sticky events stream and the profile updates stream. +//! The streams a persist takes IDs from: inside its transaction, as +//! Python's `get_next_txn` and `get_next_mult_txn` do, the cache stream, +//! the sticky events stream and the profile updates stream. Before its +//! transaction, as Python's `get_next_mult` does, the events or the +//! backfill stream, for the chunk's stream orderings. //! //! The IDs stay unfinished until the transaction has committed and Python //! has applied its effects, as Python finishes them only after the @@ -24,12 +26,16 @@ use std::sync::{Arc, Mutex}; use anyhow::anyhow; +use tokio_postgres::Row; use crate::{ database::postgres::write::{WriteError, WriteTxn}, - storage::stream_ids::StreamIdTracker, + storage::{events::Querier, stream_ids::StreamIdTracker}, }; +/// `_load_next_mult_id_txn`'s statement: `$2` IDs from the sequence `$1`. +const NEXTVAL_SQL: &str = "SELECT nextval($1::text) FROM generate_series(1, $2::bigint)"; + /// A stream's IDs taken by one persist, over all its attempts. An ID the /// committed attempt took is handed to Python with the effects. The rest /// are marked finished here: when an attempt fails, at the start of the @@ -66,6 +72,17 @@ impl TxnStream { Self::new(tracker, "profile_updates_sequence") } + /// The events stream, `_stream_id_gen`. + pub fn events(tracker: Arc) -> Self { + Self::new(tracker, "events_stream_seq") + } + + /// The backfill stream, `_backfill_id_gen`. Its IDs are positive, as + /// the tracker holds them. The events take them negated. + pub fn backfill(tracker: Arc) -> Self { + Self::new(tracker, "events_backfill_stream_seq") + } + fn lock_pending(&self) -> std::sync::MutexGuard<'_, Vec> { self.pending.lock().unwrap_or_else(|e| e.into_inner()) } @@ -81,21 +98,23 @@ impl TxnStream { /// `_load_next_mult_id_txn`: the next `n` IDs on the stream, /// unfinished. pub async fn next_ids(&self, txn: &WriteTxn, n: usize) -> Result, WriteError> { - let fetch = Fetch::begin(self); let n = i64::try_from(n).map_err(anyhow::Error::new)?; - let rows = txn - .query( - "SELECT nextval($1::text) FROM generate_series(1, $2::bigint)", - &[&self.sequence, &n], - ) - .await?; - let ids = rows - .iter() - .map(|row| row.try_get(0)) - .collect::, _>>() - .map_err(anyhow::Error::new)?; - fetch.allocated(&ids); - Ok(ids) + let fetch = Fetch::begin(self); + let rows = txn.query(NEXTVAL_SQL, &[&self.sequence, &n]).await?; + Ok(fetch.allocated_rows(&rows)?) + } + + /// As [`Self::next_ids`], but outside a transaction, as `get_next_mult` + /// fetches its IDs in autocommit. The sequence ignores transactions. + pub async fn next_ids_outside_txn( + &self, + querier: &impl Querier, + n: usize, + ) -> anyhow::Result> { + let n = i64::try_from(n)?; + let fetch = Fetch::begin(self); + let rows = querier.query(NEXTVAL_SQL, &[&self.sequence, &n]).await?; + fetch.allocated_rows(&rows) } /// Mark finished the IDs of an attempt that failed, as Python's @@ -131,6 +150,17 @@ impl<'a> Fetch<'a> { } } + /// The fetch returned `rows` of IDs, which stay unfinished. Ends the + /// fetch without allocating if a row cannot be read. + fn allocated_rows(self, rows: &[Row]) -> anyhow::Result> { + let ids = rows + .iter() + .map(|row| row.try_get(0)) + .collect::, _>>()?; + self.allocated(&ids); + Ok(ids) + } + /// The fetch allocated `ids`, which stay unfinished. fn allocated(mut self, ids: &[i64]) { if let Some(token) = self.token.take() { diff --git a/synapse/config/experimental.py b/synapse/config/experimental.py index ff760d99fb..e5d97e87fb 100644 --- a/synapse/config/experimental.py +++ b/synapse/config/experimental.py @@ -415,8 +415,8 @@ class ExperimentalConfig(Config): # - "prepare": Rust's results for that are used. # - "transaction": Rust also runs the write transaction. # - "full": Rust persists each queued batch. - # The levels are being built up in that order. So far "transaction" - # and "full" act as "prepare". + # The levels are being built up in that order. So far "full" only + # has Rust take the stream orderings of each chunk as well. rust_event_persistence = experimental.get("rust_event_persistence", "off") # YAML reads an unquoted `off` as a boolean. if rust_event_persistence is False: diff --git a/synapse/storage/controllers/persist_events.py b/synapse/storage/controllers/persist_events.py index 8940f396e2..7d8865bb2e 100644 --- a/synapse/storage/controllers/persist_events.py +++ b/synapse/storage/controllers/persist_events.py @@ -928,16 +928,22 @@ class EventsPersistenceStorageController: def _rust_transaction(self, prepared: _PreparedChunk) -> RustTransaction | None: """Whether to write the chunk in Rust: at `rust_event_persistence: - transaction`, if Rust prepared it.""" + transaction`, if Rust prepared it. At `full`, Rust takes its stream + orderings too.""" + experimental = self.hs.config.experimental if ( prepared.rust is None or self._rust_persister is None - or not self.hs.config.experimental.rust_event_persistence_at_least( - "transaction" - ) + or not experimental.rust_event_persistence_at_least("transaction") ): return None - return RustTransaction(self._rust_persister, prepared.rust) + return RustTransaction( + self._rust_persister, + prepared.rust, + allocate_stream_orderings=experimental.rust_event_persistence_at_least( + "full" + ), + ) async def _prepare_chunk( self, room_id: str, chunk: list[EventPersistencePair], backfilled: bool diff --git a/synapse/storage/databases/main/events.py b/synapse/storage/databases/main/events.py index 9b41259e3b..bc7f8e55aa 100644 --- a/synapse/storage/databases/main/events.py +++ b/synapse/storage/databases/main/events.py @@ -68,7 +68,10 @@ from synapse.storage.database import ( make_tuple_in_list_sql_clause, ) from synapse.storage.databases.main.event_federation import EventFederationStore -from synapse.storage.databases.main.rust_persist import apply_rust_persist_effects +from synapse.storage.databases.main.rust_persist import ( + apply_rust_persist_effects, + finish_rust_persist_stream_ids, +) from synapse.storage.databases.main.search import SearchEntry from synapse.storage.engines import PostgresEngine from synapse.storage.util.id_generators import AbstractStreamIdGenerator @@ -253,6 +256,8 @@ class RustTransaction: persister: "Persister" prepared: "PreparedChunk" + # Rust takes the chunk's stream orderings too (`full`). + allocate_stream_orderings: bool = False @attr.s(slots=True, auto_attribs=True) @@ -405,7 +410,8 @@ class PersistEventsStore: table_changes: The sliding sync table changes and sticky events to un-soft-fail, if the caller computed them. Otherwise they are computed here. - rust_transaction: Write the chunk in Rust, which prepared it. + rust_transaction: Write the chunk in Rust, which prepared it, and + which may take its stream orderings too. Returns: Resolves when the events have been persisted @@ -415,6 +421,18 @@ class PersistEventsStore: a room that has been un-partial stated. """ + if rust_transaction is not None and rust_transaction.allocate_stream_orderings: + # Rust takes the stream orderings itself, as late as it can. + await self._persist_events_in_rust( + rust_transaction, + room_id, + events_and_contexts, + backfilled=use_negative_stream_ordering, + inhibit_local_membership_updates=inhibit_local_membership_updates, + new_forward_extremities=new_forward_extremities, + ) + return + # We want to calculate the stream orderings as late as possible, as # we only notify after all events with a lesser stream ordering have # been persisted. I.e. if we spend 10s inside the with block then @@ -465,6 +483,7 @@ class PersistEventsStore: rust_transaction, room_id, events_and_contexts, + backfilled=use_negative_stream_ordering, inhibit_local_membership_updates=inhibit_local_membership_updates, ) else: @@ -481,116 +500,135 @@ class PersistEventsStore: new_state_dag_forward_extremities=new_state_dag_forward_extremities, sticky_events_to_un_soft_fail=sticky_events_to_un_soft_fail, ) - persist_event_counter.labels(**{SERVER_NAME_LABEL: self.server_name}).inc( - len(events_and_contexts) + await self._after_persisting_events( + room_id, + events_and_contexts, + use_negative_stream_ordering=use_negative_stream_ordering, + new_forward_extremities=new_forward_extremities, ) - if not use_negative_stream_ordering: - # we don't want to set the event_persisted_position to a negative - # stream_ordering. - synapse.metrics.event_persisted_position.labels( - **{SERVER_NAME_LABEL: self.server_name} - ).set(stream) + async def _after_persisting_events( + self, + room_id: str, + events_and_contexts: list[EventPersistencePair], + *, + use_negative_stream_ordering: bool, + new_forward_extremities: set[str] | None, + ) -> None: + """What follows writing a chunk: the metrics, the MSC4293 + `room_ban_redactions` updates and the forward extremities prefill.""" + # The chunk's last stream ordering, which its last event took. + last_stream_ordering = events_and_contexts[-1][ + 0 + ].internal_metadata.stream_ordering + assert last_stream_ordering is not None + persist_event_counter.labels(**{SERVER_NAME_LABEL: self.server_name}).inc( + len(events_and_contexts) + ) - for event, context in events_and_contexts: - if context.app_service: - origin_type = "application_service" - elif self.hs.is_mine_id(event.sender): - origin_type = "local" - else: - origin_type = "remote" + if not use_negative_stream_ordering: + # we don't want to set the event_persisted_position to a negative + # stream_ordering. + synapse.metrics.event_persisted_position.labels( + **{SERVER_NAME_LABEL: self.server_name} + ).set(last_stream_ordering) - # We only track a subset of event types, to avoid high - # cardinality in the metrics. - metrics_event_type = ( - event.type if event.type in TRACKED_EVENT_TYPES else "*other*" - ) + for event, context in events_and_contexts: + if context.app_service: + origin_type = "application_service" + elif self.hs.is_mine_id(event.sender): + origin_type = "local" + else: + origin_type = "remote" - event_counter.labels( - type=metrics_event_type, - origin_type=origin_type, - **{SERVER_NAME_LABEL: self.server_name}, - ).inc() + # We only track a subset of event types, to avoid high + # cardinality in the metrics. + metrics_event_type = ( + event.type if event.type in TRACKED_EVENT_TYPES else "*other*" + ) + event_counter.labels( + type=metrics_event_type, + origin_type=origin_type, + **{SERVER_NAME_LABEL: self.server_name}, + ).inc() + + if ( + not self.hs.config.experimental.msc4293_enabled + or event.type != EventTypes.Member + or event.state_key is None + ): + continue + + # check if this is an unban/join that will undo a ban/kick redaction for + # a user in the room + if event.membership in [Membership.LEAVE, Membership.JOIN]: if ( - not self.hs.config.experimental.msc4293_enabled - or event.type != EventTypes.Member - or event.state_key is None - ): - continue - - # check if this is an unban/join that will undo a ban/kick redaction for - # a user in the room - if event.membership in [Membership.LEAVE, Membership.JOIN]: - if ( - event.membership == Membership.LEAVE - and event.sender == event.state_key - ): - # self-leave, ignore - continue - - # if there is an existing ban/leave causing redactions for - # this user/room combination update the entry with the stream - # ordering when the redactions should stop - in the case of a backfilled - # event where the stream ordering is negative, use the current max stream - # ordering - stream_ordering = event.internal_metadata.stream_ordering - assert stream_ordering is not None - if stream_ordering < 0: - stream_ordering = self._stream_id_gen.get_current_token() - await self.db_pool.simple_update( - "room_ban_redactions", - {"room_id": event.room_id, "user_id": event.state_key}, - {"redact_end_ordering": stream_ordering}, - desc="room_ban_redactions update redact_end_ordering", - ) - - # check for msc4293 redact_events flag and apply if found - if event.membership not in [Membership.LEAVE, Membership.BAN]: - continue - redact = event.content.get("org.matrix.msc4293.redact_events", False) - if not redact or not isinstance(redact, bool): - continue - # self-bans currently are not authorized so we don't check for that - # case - if ( - event.membership == Membership.BAN + event.membership == Membership.LEAVE and event.sender == event.state_key ): + # self-leave, ignore continue - # check that sender can redact - redact_allowed = await self._can_sender_redact(event) - - # Signal that this user's past events in this room - # should be redacted by adding an entry to - # `room_ban_redactions`. - if redact_allowed: - await self.db_pool.simple_upsert( - "room_ban_redactions", - {"room_id": event.room_id, "user_id": event.state_key}, - { - "redacting_event_id": event.event_id, - "redact_end_ordering": None, - }, - { - "room_id": event.room_id, - "user_id": event.state_key, - "redacting_event_id": event.event_id, - "redact_end_ordering": None, - }, - ) - - # normally the cache entry for a redacted event would be invalidated - # by an arriving redaction event, but since we are not creating redaction - # events we invalidate manually - self.store._invalidate_local_get_event_cache_room_id(event.room_id) - - if new_forward_extremities: - self.store.get_latest_event_ids_in_room.prefill( - (room_id,), frozenset(new_forward_extremities) + # if there is an existing ban/leave causing redactions for + # this user/room combination update the entry with the stream + # ordering when the redactions should stop - in the case of a backfilled + # event where the stream ordering is negative, use the current max stream + # ordering + stream_ordering = event.internal_metadata.stream_ordering + assert stream_ordering is not None + if stream_ordering < 0: + stream_ordering = self._stream_id_gen.get_current_token() + await self.db_pool.simple_update( + "room_ban_redactions", + {"room_id": event.room_id, "user_id": event.state_key}, + {"redact_end_ordering": stream_ordering}, + desc="room_ban_redactions update redact_end_ordering", ) + # check for msc4293 redact_events flag and apply if found + if event.membership not in [Membership.LEAVE, Membership.BAN]: + continue + redact = event.content.get("org.matrix.msc4293.redact_events", False) + if not redact or not isinstance(redact, bool): + continue + # self-bans currently are not authorized so we don't check for that + # case + if event.membership == Membership.BAN and event.sender == event.state_key: + continue + + # check that sender can redact + redact_allowed = await self._can_sender_redact(event) + + # Signal that this user's past events in this room + # should be redacted by adding an entry to + # `room_ban_redactions`. + if redact_allowed: + await self.db_pool.simple_upsert( + "room_ban_redactions", + {"room_id": event.room_id, "user_id": event.state_key}, + { + "redacting_event_id": event.event_id, + "redact_end_ordering": None, + }, + { + "room_id": event.room_id, + "user_id": event.state_key, + "redacting_event_id": event.event_id, + "redact_end_ordering": None, + }, + ) + + # normally the cache entry for a redacted event would be invalidated + # by an arriving redaction event, but since we are not creating redaction + # events we invalidate manually + self.store._invalidate_local_get_event_cache_room_id(event.room_id) + + if new_forward_extremities: + self.store.get_latest_event_ids_in_room.prefill( + (room_id,), frozenset(new_forward_extremities) + ) + async def _can_sender_redact(self, event: EventBase) -> bool: state_filter = StateFilter.from_types( [(EventTypes.PowerLevels, ""), (EventTypes.Create, "")] @@ -1137,11 +1175,16 @@ class PersistEventsStore: room_id: str, events_and_contexts: list[EventPersistencePair], *, + backfilled: bool, inhibit_local_membership_updates: bool, + new_forward_extremities: set[str] | None = None, ) -> None: """Write the chunk in Rust, as `_persist_events_txn` does, then apply the cache updates its transaction asks for. Called with the chunk's - stream orderings allocated, which are finished after. + stream orderings allocated, which are finished after, unless + `rust_transaction` takes them itself, from the backfill stream if + `backfilled`. Then this also runs `_after_persisting_events`, with + `new_forward_extremities`, before it finishes them. Raises: PartialStateConflictError: as `_persist_events_txn`. @@ -1152,22 +1195,61 @@ class PersistEventsStore: # caller waits for both, as Python's transaction cannot be cancelled. async def persist_and_apply() -> None: - effects = await make_deferred_yieldable( - rust_transaction.persister.persist_events_txn( - room_id, - events_and_contexts, - rust_transaction.prepared, - inhibit_local_membership_updates=inhibit_local_membership_updates, + persister = rust_transaction.persister + allocate = rust_transaction.allocate_stream_orderings + if allocate: + try: + effects = await make_deferred_yieldable( + persister.persist_chunk( + room_id, + events_and_contexts, + rust_transaction.prepared, + backfilled=backfilled, + inhibit_local_membership_updates=inhibit_local_membership_updates, + ) + ) + except BaseException: + # Rust finished the stream orderings it took. Tell + # replication, as `get_next_mult` does when its block + # raises. + self.hs.get_replication_notifier().notify_replication() + raise + else: + effects = await make_deferred_yieldable( + persister.persist_events_txn( + room_id, + events_and_contexts, + rust_transaction.prepared, + inhibit_local_membership_updates=inhibit_local_membership_updates, + ) ) - ) # The copy of a duplicated event that Rust wrote, as Python's # `_add_to_cache` reads the deduplicated chunk. written = self._filter_events_and_contexts_for_duplicates( events_and_contexts ) - apply_rust_persist_effects( - self.store, effects, {event.event_id: event for event, _ in written} - ) + events = {event.event_id: event for event, _ in written} + if not allocate: + apply_rust_persist_effects(self.store, effects, events) + return + # What follows the write runs before the stream orderings are + # finished, as it runs inside `get_next_mult` in Python: the + # events must not be announced before, say, the MSC4293 + # redactions they start are recorded. + try: + apply_rust_persist_effects( + self.store, effects, events, finish_stream_ids=False + ) + await self._after_persisting_events( + room_id, + events_and_contexts, + use_negative_stream_ordering=backfilled, + new_forward_extremities=new_forward_extremities, + ) + except BaseException: + finish_rust_persist_stream_ids(self.store, effects, succeeded=False) + raise + finish_rust_persist_stream_ids(self.store, effects, succeeded=True) await delay_cancellation(persist_and_apply()) diff --git a/synapse/storage/databases/main/rust_persist.py b/synapse/storage/databases/main/rust_persist.py index e875a02aaa..af257eb5fc 100644 --- a/synapse/storage/databases/main/rust_persist.py +++ b/synapse/storage/databases/main/rust_persist.py @@ -30,6 +30,8 @@ def apply_rust_persist_effects( store: "DataStore", effects: PersistEffects, events: Mapping[str, "EventBase"] | None = None, + *, + finish_stream_ids: bool = True, ) -> None: """Apply what a Rust persist transaction asks for once it has committed: the cache updates the Python transaction queues with @@ -45,13 +47,20 @@ def apply_rust_persist_effects( transaction's earlier callbacks. The events stream IDs are finished last, as Python's `get_next_mult` does once the transaction's callbacks have run, so the current token never passes an event before the stream change - caches know about it. + caches know about it. Then `stream_positions` moves on, as + `get_next_mult` moves it on success. - If a step raises, the IDs of those three streams are still - finished, as `call_on_exception` finishes them in Python, so the + If a step raises, the IDs of all four streams are still finished, as + `call_on_exception` and `get_next_mult` finish them in Python, so the streams do not stall. + + Without `finish_stream_ids`, the events stream IDs are left to the + caller, who must call `finish_rust_persist_stream_ids` whatever + happens. """ 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 try: for ( stream_ordering, @@ -120,12 +129,9 @@ def apply_rust_persist_effects( (room_id,), frozenset(extremities) ) - if effects.stream_ids: - id_gen = ( - store._backfill_id_gen if effects.backfilled else store._stream_id_gen - ) - id_gen._mark_ids_as_finished(effects.stream_ids) - store.hs.get_replication_notifier().notify_replication() + if not event_ids_finished: + finish_rust_persist_stream_ids(store, effects, succeeded=True) + event_ids_finished = True if effects.new_replication_data: store.hs.get_notifier().on_new_replication_data() except BaseException: @@ -138,9 +144,27 @@ def apply_rust_persist_effects( ) if effects.sticky_stream_ids and not sticky_ids_finished: store._sticky_events_id_gen._mark_ids_as_finished(effects.sticky_stream_ids) + if not event_ids_finished: + finish_rust_persist_stream_ids(store, effects, succeeded=False) raise +def finish_rust_persist_stream_ids( + store: "DataStore", effects: PersistEffects, *, succeeded: bool +) -> None: + """Finish the events or backfill stream IDs a Rust persist took, as + `get_next_mult` finishes them when its block exits, and tell + replication. Only on success does `stream_positions` move on.""" + if not effects.stream_ids: + return + id_gen = store._backfill_id_gen if effects.backfilled else store._stream_id_gen + if succeeded: + _finish_txn_stream_ids(store, id_gen, effects.stream_ids) + else: + id_gen._mark_ids_as_finished(effects.stream_ids) + store.hs.get_replication_notifier().notify_replication() + + 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 1db1b474bc..9fe43e347d 100644 --- a/synapse/synapse_rust/persist.pyi +++ b/synapse/synapse_rust/persist.pyi @@ -80,6 +80,31 @@ class Persister: inhibit_local_membership_updates: bool, ) -> PersistEffects: """As `persist_events_txn`, but blocks. For tests.""" + def persist_chunk( + self, + room_id: str, + events_and_contexts: list[tuple[EventBase, EventContext]], + prepared: PreparedChunk, + *, + backfilled: bool, + inhibit_local_membership_updates: bool, + ) -> Deferred[PersistEffects]: + """As `persist_events_txn`, but Rust takes the chunk's stream + orderings, from the backfill stream if `backfilled`, and sets them on + the events with the time and this instance, as + `_persist_events_and_state_updates` does. They come back in the + effects' `stream_ids`, unfinished. If the persist fails, Rust + finishes them.""" + def persist_chunk_blocking( + self, + room_id: str, + events_and_contexts: list[tuple[EventBase, EventContext]], + prepared: PreparedChunk, + *, + backfilled: bool, + inhibit_local_membership_updates: bool, + ) -> PersistEffects: + """As `persist_chunk`, but blocks. For tests.""" class PreparedChunk: """What `Persister.prepare_chunk` computed. If `state_res_fallback` is diff --git a/tests/storage/test_rust_persist_prereqs.py b/tests/storage/test_rust_persist_prereqs.py index e5de14c9e7..bcc3b184a5 100644 --- a/tests/storage/test_rust_persist_prereqs.py +++ b/tests/storage/test_rust_persist_prereqs.py @@ -33,6 +33,7 @@ from twisted.internet.testing import MemoryReactor import synapse.metrics from synapse.api.constants import EventTypes, Membership, RelationTypes +from synapse.api.errors import PartialStateConflictError from synapse.api.room_versions import RoomVersions from synapse.config import ConfigError from synapse.config.experimental import ExperimentalConfig @@ -55,6 +56,7 @@ from synapse.synapse_rust.persist import ( persist_gil_stats, persistence_json_for_test, ) +from synapse.types import create_requester from synapse.util.caches.deferred_cache import DeferredCache from synapse.util.caches.stream_change_cache import StreamChangeCache from synapse.util.clock import Clock @@ -867,6 +869,8 @@ class _CountingPersister: self._persister = persister self.prepares = 0 self.transactions = 0 + # Chunks written with Rust taking their stream orderings. + self.chunks = 0 self.last_effects: PersistEffects | None = None def prepare_chunk(self, *args: Any, **kwargs: Any) -> Any: @@ -882,6 +886,15 @@ class _CountingPersister: return self._persister.persist_events_txn(*args, **kwargs).addCallback(keep) + def persist_chunk(self, *args: Any, **kwargs: Any) -> Any: + self.chunks += 1 + + def keep(result: Any) -> Any: + self.last_effects = result + return result + + return self._persister.persist_chunk(*args, **kwargs).addCallback(keep) + def __getattr__(self, name: str) -> Any: return getattr(self._persister, name) @@ -1074,9 +1087,9 @@ class RustTransactionTestCase(unittest.HomeserverTestCase): ) -class RustTransactionWorkerTestCase(BaseMultiWorkerStreamTestCase): - """`rust_event_persistence: transaction` on a worker event persister: - the worker writes in Rust, to the main database.""" +class RustFullTestCase(unittest.HomeserverTestCase): + """`rust_event_persistence: full`: Rust also takes the chunks' stream + orderings.""" servlets = [ admin.register_servlets, @@ -1087,7 +1100,345 @@ class RustTransactionWorkerTestCase(BaseMultiWorkerStreamTestCase): def default_config(self) -> dict[str, Any]: config = super().default_config() config.setdefault("experimental_features", {})["rust_event_persistence"] = ( - "transaction" + "full" + ) + return config + + def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer: + if RUST_PERSIST_FOR_TESTS not in (None, "full"): + raise SkipTest("SYNAPSE_TEST_RUST_PERSIST overrides the level") + hs = self.setup_test_homeserver() + if not isinstance( + hs.get_datastores().main.db_pool._db_pool, RustConnectionPool + ): + raise SkipTest("needs the native Rust Postgres driver") + return hs + + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + self.store = hs.get_datastores().main + persistence = hs.get_storage_controllers().persistence + assert persistence is not None + self.persistence = persistence + self.persister = _CountingPersister(persistence._rust_persister) + persistence._rust_persister = self.persister # type: ignore[assignment] + self.user = self.register_user("alice", "pass") + self.token = self.login("alice", "pass") + self.room_id = self.helper.create_room_as(self.user, tok=self.token) + + def _build(self, body: str) -> tuple[EventBase, EventContext]: + event, unpersisted_context = self.get_success( + self.hs.get_event_creation_handler().create_event( + create_requester(self.user), + { + "type": EventTypes.Message, + "content": {"msgtype": "m.text", "body": body}, + "room_id": self.room_id, + "sender": self.user, + }, + ) + ) + context = self.get_success(unpersisted_context.persist(event)) + return event, context + + def _stream_position(self, stream_name: str) -> int | None: + rows = self.get_success( + self.store.db_pool.simple_select_onecol( + "stream_positions", + {"stream_name": stream_name, "instance_name": "master"}, + "stream_id", + ) + ) + return rows[0] if rows else None + + def _event_row(self, event_id: str) -> tuple[int, str, int]: + """`(stream_ordering, instance_name, received_ts)` of the event's + `events` row.""" + row = self.get_success( + self.store.db_pool.simple_select_one( + "events", + {"event_id": event_id}, + ("stream_ordering", "instance_name", "received_ts"), + ) + ) + return row[0], row[1], row[2] + + def test_rust_takes_stream_orderings(self) -> None: + """Rust takes the chunk's stream orderings and sets them on the + events, and the applier finishes them and moves `stream_positions` + on, as `get_next_mult` does.""" + id_gen = self.store._stream_id_gen + chunks = self.persister.chunks + with patch.object( + id_gen, "get_next_mult", side_effect=AssertionError("Python allocated") + ): + event_id = self.helper.send(self.room_id, "hello", tok=self.token)[ + "event_id" + ] + self.assertEqual(self.persister.chunks, chunks + 1) + effects = self.persister.last_effects + assert effects is not None + self.assertFalse(effects.backfilled) + (stream_ordering,) = effects.stream_ids + + self.assertEqual( + self._event_row(event_id), + (stream_ordering, "master", self.clock.time_msec()), + ) + self.assertGreaterEqual(id_gen.get_current_token(), stream_ordering) + assert id_gen._writers + self.assertEqual(self._stream_position("events"), stream_ordering) + + def test_chunk_with_duplicate(self) -> None: + """A chunk takes one ordering per entry, in order. An event that + appears twice keeps the later one, as in Python, and every ordering + is finished. What follows the write sees the last ordering.""" + first = self._build("one") + second = self._build("two") + with patch.object( + self.store, + "get_latest_event_ids_in_room", + Mock(wraps=self.store.get_latest_event_ids_in_room), + ) as latest: + self.get_success(self.persistence.persist_events([first, second, first])) + effects = self.persister.last_effects + assert effects is not None + self.assertEqual(len(effects.stream_ids), 3) + self.assertEqual(effects.stream_ids, sorted(effects.stream_ids)) + _, second_id, last_id = effects.stream_ids + + self.assertEqual(self._event_row(first[0].event_id)[0], last_id) + self.assertEqual(self._event_row(second[0].event_id)[0], second_id) + self.assertGreaterEqual(self.store._stream_id_gen.get_current_token(), last_id) + # `_after_persisting_events` ran with the chunk. + self.assertEqual( + REGISTRY.get_sample_value( + "synapse_event_persisted_position", {SERVER_NAME_LABEL: "test"} + ), + last_id, + ) + # Both were built on the same extremity. + latest.prefill.assert_called_with( + (self.room_id,), frozenset([first[0].event_id, second[0].event_id]) + ) + + def test_follow_ups_run_before_orderings_finish(self) -> None: + """What follows the write (the MSC4293 redactions among it) runs + before the token passes the chunk, as inside `get_next_mult`. If it + raises, the orderings are still finished.""" + persist_events = self.hs.get_datastores().persist_events + assert persist_events is not None + id_gen = self.store._stream_id_gen + original = persist_events._after_persisting_events + seen: list[tuple[int, int | None]] = [] + + async def after(room_id: str, events_and_contexts: list, **kwargs: Any) -> None: + seen.append( + ( + id_gen.get_current_token(), + events_and_contexts[-1][0].internal_metadata.stream_ordering, + ) + ) + await original(room_id, events_and_contexts, **kwargs) + + with patch.object( + persist_events, "_after_persisting_events", side_effect=after + ): + self.helper.send(self.room_id, "hello", tok=self.token) + ((token, stream_ordering),) = seen + assert stream_ordering is not None + self.assertLess(token, stream_ordering) + self.assertGreaterEqual(id_gen.get_current_token(), stream_ordering) + + class AfterFailed(Exception): + pass + + event, context = self._build("fails after") + with patch.object( + persist_events, "_after_persisting_events", side_effect=AfterFailed() + ): + self.get_failure( + self.persistence.persist_events([(event, context)]), AfterFailed + ) + failed = event.internal_metadata.stream_ordering + assert failed is not None + self.assertGreaterEqual(id_gen.get_current_token(), failed) + + def test_gil_count(self) -> None: + """Taking the stream orderings and setting them on the events takes + the GIL at no persist site.""" + # The first persist opens the pool, at `persist_db_pool`. + self.helper.send(self.room_id, "warm up", tok=self.token) + before = persist_gil_stats() + chunks = self.persister.chunks + self.helper.send(self.room_id, "hello", tok=self.token) + self.assertEqual(self.persister.chunks, chunks + 1) + after = persist_gil_stats() + changed = { + site: after[site][0] - before.get(site, (0, 0.0))[0] + for site in after + if after[site][0] != before.get(site, (0, 0.0))[0] + } + self.assertEqual(changed, {}) + + def test_backfill_takes_negative_orderings(self) -> None: + """A backfilled chunk takes its stream orderings from the backfill + stream, negated on the events.""" + id_gen = self.store._backfill_id_gen + event, context = self._build("old") + with patch.object( + id_gen, "get_next_mult", side_effect=AssertionError("Python allocated") + ): + self.get_success( + self.persistence.persist_events([(event, context)], backfilled=True) + ) + effects = self.persister.last_effects + assert effects is not None + self.assertTrue(effects.backfilled) + (backfill_id,) = effects.stream_ids + self.assertGreater(backfill_id, 0) + self.assertEqual(event.internal_metadata.stream_ordering, -backfill_id) + self.assertEqual( + self._event_row(event.event_id), + (-backfill_id, "master", self.clock.time_msec()), + ) + # The backfill token is negated too. + self.assertLessEqual(id_gen.get_current_token(), -backfill_id) + assert id_gen._writers + self.assertEqual(self._stream_position("backfill"), -backfill_id) + + def test_failed_persist_finishes_its_orderings(self) -> None: + """A chunk whose transaction fails leaves no stream ordering + unfinished, so the stream does not stall, and replication is told.""" + event, context = self._build("partial") + context.partial_state = True + notifier = self.hs.get_replication_notifier() + with patch.object( + notifier, "notify_replication", wraps=notifier.notify_replication + ) as notify: + self.get_failure( + self.persistence.persist_events([(event, context)]), + PartialStateConflictError, + ) + notify.assert_called() + failed = event.internal_metadata.stream_ordering + assert failed is not None + + event_id = self.helper.send(self.room_id, "after", tok=self.token)["event_id"] + position = self.get_success(self.store.get_position_for_event(event_id)) + self.assertGreater(position.stream, failed) + # A leaked ordering would hold the token below it. + self.assertGreaterEqual( + self.store._stream_id_gen.get_current_token(), position.stream + ) + + def test_applier_failure_finishes_orderings(self) -> None: + """If applying the effects raises before the stream orderings are + finished, they are still finished, and replication is told.""" + + class ApplyFailed(Exception): + pass + + event, context = self._build("hello") + notifier = self.hs.get_replication_notifier() + with ( + patch.object( + self.store._events_stream_cache, + "entity_has_changed", + side_effect=ApplyFailed(), + ), + patch.object( + notifier, "notify_replication", wraps=notifier.notify_replication + ) as notify, + ): + self.get_failure( + self.persistence.persist_events([(event, context)]), ApplyFailed + ) + notify.assert_called() + effects = self.persister.last_effects + assert effects is not None + (stream_ordering,) = effects.stream_ids + self.assertGreaterEqual( + self.store._stream_id_gen.get_current_token(), stream_ordering + ) + + def test_cancelled_persist_applies_its_effects(self) -> None: + """Cancelling the persist while Rust takes the orderings and writes + the chunk waits for both, and the orderings are finished.""" + gate: list[tuple[Any, Any, defer.Deferred]] = [] + blocking = self.persister._persister + + class Gated: + def persist_chunk(self, *args: Any, **kwargs: Any) -> Any: + d: defer.Deferred = defer.Deferred() + gate.append((args, kwargs, d)) + return d + + def __getattr__(self, name: str) -> Any: + return getattr(blocking, name) + + self.persister._persister = Gated() + + persist_events = self.hs.get_datastores().persist_events + assert persist_events is not None + original = persist_events._persist_events_in_rust + cancelled = [] + + async def cancel_then_wait(*args: Any, **kwargs: Any) -> None: + d = defer.ensureDeferred(original(*args, **kwargs)) + d.cancel() + try: + await make_deferred_yieldable(d) + except defer.CancelledError: + cancelled.append(True) + + with patch.object( + persist_events, "_persist_events_in_rust", side_effect=cancel_then_wait + ): + channel = make_request( + self.reactor, + self.site, + "PUT", + f"/rooms/{self.room_id}/send/m.room.message/txn1", + {"msgtype": "m.text", "body": "hello"}, + access_token=self.token, + await_result=False, + ) + self.pump() + self.assertEqual(len(gate), 1) + # Cancelled, but waiting for the write. + self.assertEqual(cancelled, []) + args, kwargs, d = gate[0] + blocking.persist_chunk(*args, **kwargs).chainDeferred(d) + channel.await_result() + self.assertEqual(channel.code, 200, channel.json_body) + event_id = channel.json_body["event_id"] + self.assertEqual(cancelled, [True]) + + effects = self.persister.last_effects + assert effects is not None + (stream_ordering,) = effects.stream_ids + self.assertEqual(self._event_row(event_id)[0], stream_ordering) + self.assertGreaterEqual( + self.store._stream_id_gen.get_current_token(), stream_ordering + ) + + +class RustTransactionWorkerTestCase(BaseMultiWorkerStreamTestCase): + """`rust_event_persistence: transaction` on a worker event persister: + the worker writes in Rust, to the main database.""" + + level = "transaction" + + servlets = [ + admin.register_servlets, + room.register_servlets, + login.register_servlets, + ] + + def default_config(self) -> dict[str, Any]: + config = super().default_config() + config.setdefault("experimental_features", {})["rust_event_persistence"] = ( + self.level ) config["stream_writers"] = {"events": ["worker1"]} config["instance_map"] = { @@ -1097,7 +1448,7 @@ class RustTransactionWorkerTestCase(BaseMultiWorkerStreamTestCase): return config def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: - if RUST_PERSIST_FOR_TESTS not in (None, "transaction"): + if RUST_PERSIST_FOR_TESTS not in (None, self.level): raise SkipTest("SYNAPSE_TEST_RUST_PERSIST overrides the level") if not isinstance( hs.get_datastores().main.db_pool._db_pool, RustConnectionPool @@ -1118,7 +1469,10 @@ class RustTransactionWorkerTestCase(BaseMultiWorkerStreamTestCase): room_id = self.helper.create_room_as(user, tok=token) event_id = self.helper.send(room_id, "hello", tok=token)["event_id"] - self.assertGreater(persister.transactions, 0) + if self.level == "full": + self.assertGreater(persister.chunks, 0) + else: + self.assertGreater(persister.transactions, 0) assert persister.last_effects is not None self.assertIn(event_id, persister.last_effects.event_cache_prefills) # The main process reads what the worker's Rust pool wrote. @@ -1127,6 +1481,30 @@ class RustTransactionWorkerTestCase(BaseMultiWorkerStreamTestCase): self.assertEqual(position.instance_name, "worker1") event = self.get_success(store.get_event(event_id)) self.assertEqual(event.content["body"], "hello") + if self.level == "full": + # The worker took the ordering, finished it and moved its + # `stream_positions` row on. + self.assertEqual(persister.last_effects.stream_ids, [position.stream]) + worker_id_gen = worker.get_datastores().main._stream_id_gen + self.assertGreaterEqual( + worker_id_gen.get_current_token_for_writer("worker1"), + position.stream, + ) + row = self.get_success( + store.db_pool.simple_select_one_onecol( + "stream_positions", + {"stream_name": "events", "instance_name": "worker1"}, + "stream_id", + ) + ) + self.assertEqual(row, position.stream) + + +class RustFullWorkerTestCase(RustTransactionWorkerTestCase): + """`rust_event_persistence: full` on a worker event persister: the + worker takes the stream orderings in Rust.""" + + level = "full" class PersisterOffTestCase(unittest.HomeserverTestCase):