From fd8c19ed2cc03fd6699cf531c6d046c45b42f0e9 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Sat, 17 Oct 2026 23:25:58 +0000 Subject: [PATCH] Run what follows each chunk's write in Rust at rust_event_persistence: full `_after_persisting_events` moves into the batch step, after the write and before Python applies the chunk and finishes its stream orderings, as it ran inside `get_next_mult` in Python: - The persisted events, by type and origin, and the persisted position go to the Rust metrics that `flush_rust_persist_metrics` reports. - With MSC4293, a join, or a leave by someone else, ends the redaction of the target's events. A ban or kick with `redact_events` set, by a sender who may redact (`_can_sender_redact`, read from the current state), starts it. Python then drops the room's events from its event cache, through a new `event_cache_room_invalidations` effect, even if a later event of the chunk fails, as Python drops them right after each start. - The forward extremities prefill moves into the effects. The MSC4293 flag goes with each batch, as Python reads it per chunk and tests change it at runtime. If what follows the write fails, the chunk is still written, so Python applies its effects and finishes its orderings as a failure, as `get_next_mult` does when its block raises, then raises the error and releases the batch's state groups as a failed persist does. Where Python would raise on odd power levels or memberships, Rust is more lenient: it reads a level that is a string of an integer as that number (Python compares two strings, or raises), a null default as 0 and a `users` or `events` that is not an object as missing, and skips a member event without a membership. Numbers, floats and bools compare as in Python. The chunk path, for chunks Rust hands back, keeps the Python block. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LXMkcZVkcZhi21augBff4a --- rust/src/persist/effects.rs | 17 + rust/src/persist/follow_ups.rs | 383 ++++++++++++++++++ rust/src/persist/mod.rs | 3 + rust/src/persist/python.rs | 91 ++++- synapse/storage/controllers/persist_events.py | 35 +- .../storage/databases/main/rust_persist.py | 5 + synapse/synapse_rust/persist.pyi | 13 +- tests/storage/test_rust_persist_prereqs.py | 352 ++++++++++++++-- 8 files changed, 840 insertions(+), 59 deletions(-) create mode 100644 rust/src/persist/follow_ups.rs diff --git a/rust/src/persist/effects.rs b/rust/src/persist/effects.rs index ebcd2f6590..3c917d2cbb 100644 --- a/rust/src/persist/effects.rs +++ b/rust/src/persist/effects.rs @@ -137,6 +137,9 @@ pub struct PersistEffects { /// The events to put in `_get_event_cache`, by `_add_to_cache`'s /// `local_prefill`. Python holds the events. event_cache_prefills: Vec, + /// `_invalidate_local_get_event_cache_room_id(room_id)`, after the + /// prefills: the room's events, once MSC4293 starts redacting a user's. + event_cache_room_invalidations: Vec, /// `_get_state_group_for_event.prefill((event_id,), state_group)`. state_group_prefills: Vec<(String, i64)>, /// `get_latest_event_ids_in_room.prefill((room_id,), extremities)`. @@ -199,6 +202,12 @@ impl PersistEffects { self.event_cache_prefills.push(event_id); } + pub fn add_event_cache_room_invalidation(&mut self, room_id: String) { + if !self.event_cache_room_invalidations.contains(&room_id) { + self.event_cache_room_invalidations.push(room_id); + } + } + pub fn add_state_group_prefill(&mut self, event_id: String, state_group: i64) { self.state_group_prefills.push((event_id, state_group)); } @@ -240,6 +249,7 @@ impl PersistEffects { sticky_stream_ids = Vec::new(), profile_stream_ids = Vec::new(), event_cache_prefills = Vec::new(), + event_cache_room_invalidations = Vec::new(), state_group_prefills = Vec::new(), latest_event_ids = None, stream_changes = Vec::new(), @@ -257,6 +267,7 @@ impl PersistEffects { sticky_stream_ids: Vec, profile_stream_ids: Vec, event_cache_prefills: Vec, + event_cache_room_invalidations: Vec, state_group_prefills: Vec<(String, i64)>, latest_event_ids: Option<(String, Vec)>, stream_changes: Vec<(String, String, i64)>, @@ -273,6 +284,7 @@ impl PersistEffects { sticky_stream_ids, profile_stream_ids, event_cache_prefills, + event_cache_room_invalidations, state_group_prefills, latest_event_ids, stream_changes: stream_changes @@ -344,6 +356,11 @@ impl PersistEffects { self.event_cache_prefills.clone() } + #[getter] + fn event_cache_room_invalidations(&self) -> Vec { + self.event_cache_room_invalidations.clone() + } + #[getter] fn state_group_prefills(&self) -> Vec<(String, i64)> { self.state_group_prefills.clone() diff --git a/rust/src/persist/follow_ups.rs b/rust/src/persist/follow_ups.rs new file mode 100644 index 0000000000..004b5ab66f --- /dev/null +++ b/rust/src/persist/follow_ups.rs @@ -0,0 +1,383 @@ +/* + * This file is licensed under the Affero General Public License (AGPL) version 3. + * + * Copyright (C) 2026 Element Creations Ltd + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * See the GNU Affero General Public License for more details: + * . + * + */ + +//! What follows writing a chunk, before its stream orderings are +//! finished: `_after_persisting_events`. The metrics it counts, and the +//! MSC4293 `room_ban_redactions` rows of the bans and kicks that redact a +//! user's events, with the joins and leaves that end them. + +use std::sync::Arc; + +use anyhow::{anyhow, bail}; +use futures::FutureExt; +use serde_json::Value; + +use super::{context::PersistContext, extremities::is_mine_id, metrics::PersistCounts}; +use crate::{ + database::postgres::{ + pool::ConnectionPool, + write::{run_write_interaction, WriteError}, + }, + event_auth::is_creator, + events::Event, + state::StateKey, + storage::{events::RedactBehaviour, sync_store::SyncReader}, + types::state_filter::StateFilter, +}; + +/// `TRACKED_EVENT_TYPES`: the event types counted by name in +/// `synapse_storage_events_persisted_events_sep`. Others count as +/// `*other*`. +const TRACKED_EVENT_TYPES: [&str; 7] = [ + "m.room.message", + "m.room.encrypted", + "m.room.member", + "m.room.third_party_invite", + "m.room.redaction", + "m.room.create", + "m.room.tombstone", +]; + +/// MSC4293's flag on a ban or kick: redact the user's events. +const REDACT_EVENTS: &str = "org.matrix.msc4293.redact_events"; + +/// What `_after_persisting_events` counts for the chunk `events`, every +/// entry of it, as Python counts them. +pub fn count_chunk( + events: &[(Event, PersistContext)], + server_name: &str, + backfilled: bool, +) -> PersistCounts { + let mut counts = PersistCounts { + persisted_events: events.len() as u64, + ..Default::default() + }; + for (event, context) in events { + let origin_type = if context.app_service { + "application_service" + } else if is_mine_id(event.sender(), server_name) { + "local" + } else { + "remote" + }; + let event_type = if TRACKED_EVENT_TYPES.contains(&event.type_()) { + event.type_() + } else { + "*other*" + }; + *counts + .persisted_events_by_type + .entry((event_type.to_owned(), origin_type)) + .or_default() += 1; + } + // Not to a negative position. + if !backfilled { + counts.persisted_position = events + .last() + .and_then(|(event, _)| event.internal_metadata().stream_ordering()) + .map(|ordering| ordering.get()); + } + counts +} + +/// The MSC4293 part of `_after_persisting_events`, for each event of the +/// chunk in order: a join, or a leave by someone else, ends the +/// redactions of the target's events, at its stream ordering (the events +/// stream's current token for a backfilled one). A ban, or a kick, with +/// `redact_events` set, by a sender who may redact, starts them. +/// +/// Sets `started` once any starts, so that Python drops the room's events +/// from its event cache, even if a later event fails: Python drops them +/// right after each start. +pub async fn record_room_ban_redactions( + pool: &ConnectionPool, + reader: &SyncReader<'_>, + room_id: &str, + events: &[(Event, PersistContext)], + events_token: impl Fn() -> i64, + started: &mut bool, +) -> anyhow::Result<()> { + for (event, _) in events { + let Some(target) = event.state_key() else { + continue; + }; + if event.type_() != "m.room.member" { + continue; + } + let membership = event + .content() + .get_field("membership") + .and_then(Value::as_str); + if matches!(membership, Some("leave" | "join")) { + // A user leaving of their own accord changes nothing. + if membership == Some("leave") && event.sender() == target { + continue; + } + let mut ordering = event + .internal_metadata() + .stream_ordering() + .ok_or_else(|| anyhow!("{} has no stream ordering", event.event_id()))? + .get(); + if ordering < 0 { + ordering = events_token(); + } + end_redactions(pool, room_id, target, ordering).await?; + } + + if !matches!(membership, Some("leave" | "ban")) { + continue; + } + // Only a literal `true`. + if event.content().get_field(REDACT_EVENTS) != Some(&Value::Bool(true)) { + continue; + } + // Self-bans are not allowed. + if membership == Some("ban") && event.sender() == target { + continue; + } + if can_sender_redact(reader, room_id, event.sender()).await? { + start_redactions(pool, room_id, target, event.event_id()).await?; + *started = true; + } + } + Ok(()) +} + +/// The `simple_update` that ends the target's redactions. +async fn end_redactions( + pool: &ConnectionPool, + room_id: &str, + user_id: &str, + ordering: i64, +) -> Result<(), WriteError> { + run_write_interaction( + pool, + "room_ban_redactions update redact_end_ordering", + |txn, _| { + async move { + txn.execute( + "UPDATE room_ban_redactions SET redact_end_ordering = $3 + WHERE room_id = $1 AND user_id = $2", + &[&room_id, &user_id, &ordering], + ) + .await?; + Ok(()) + } + .boxed() + }, + ) + .await +} + +/// The `simple_upsert` that starts redacting the target's events. +async fn start_redactions( + pool: &ConnectionPool, + room_id: &str, + user_id: &str, + redacting_event_id: &str, +) -> Result<(), WriteError> { + run_write_interaction(pool, "room_ban_redactions upsert", |txn, _| { + async move { + txn.execute( + "INSERT INTO room_ban_redactions + (room_id, user_id, redacting_event_id, redact_end_ordering) + VALUES ($1, $2, $3, NULL) + ON CONFLICT (room_id, user_id) DO UPDATE SET + redacting_event_id = EXCLUDED.redacting_event_id, + redact_end_ordering = NULL", + &[&room_id, &user_id, &redacting_event_id], + ) + .await?; + Ok(()) + } + .boxed() + }) + .await +} + +/// `_can_sender_redact`: whether `sender` may redact others' events in the +/// room, by its current power levels and create event. +async fn can_sender_redact( + reader: &SyncReader<'_>, + room_id: &str, + sender: &str, +) -> anyhow::Result { + let filter = StateFilter::from_types([ + ("m.room.power_levels", Some("")), + ("m.room.create", Some("")), + ]); + let state = reader.current_state_ids(room_id, &filter).await?; + // Python indexes the state, so a room without either fails. + let event_id = |event_type: &str| -> anyhow::Result> { + state + .get(&StateKey::new(event_type, "")) + .cloned() + .ok_or_else(|| anyhow!("{room_id} has no {event_type} in its current state")) + }; + let ids = [event_id("m.room.power_levels")?, event_id("m.room.create")?]; + let events = reader + .events(&ids, RedactBehaviour::Redact, false, false) + .await?; + let find = |id: &Arc| events.iter().find(|e| e.event_id() == &**id); + let power_levels = find(&ids[0]); + let Some(create) = find(&ids[1]) else { + log::warn!("No create event found for room {room_id}"); + return Ok(false); + }; + + let power_levels = if create.room_version().msc4289_creator_power_enabled { + // The creators have infinite power, everyone else 0 without power + // levels, which is less than the default to redact (50). + if is_creator(create, sender).map_err(|e| anyhow!("{e}"))? { + return Ok(true); + } + match power_levels { + Some(power_levels) => power_levels, + None => return Ok(false), + } + } else { + // Without power levels the creator has 100, everyone else 0. + match power_levels { + Some(power_levels) => power_levels, + None => return Ok(create.sender() == sender), + } + }; + + let content = power_levels.content(); + let sender_level = match content + .get_field("users") + .and_then(|users| users.get(sender)) + .filter(|level| !level.is_null()) + { + Some(level) => level_of(level)?, + None => field_level(content.get_field("users_default"))?, + }; + let redact_level = match content.get_field("redact").filter(|l| !l.is_null()) { + Some(level) => level_of(level)?, + None => field_level(content.get_field("events_default"))?, + }; + if let Some(level) = content + .get_field("events") + .and_then(|events| events.get("m.room.redaction")) + .filter(|level| !level.is_null()) + { + if sender_level < level_of(level)? { + return Ok(false); + } + } + Ok(sender_level >= redact_level) +} + +/// A power level field Python reads with a default of 0. +fn field_level(value: Option<&Value>) -> anyhow::Result { + match value.filter(|v| !v.is_null()) { + Some(level) => level_of(level), + None => Ok(0.0), + } +} + +/// A power level as Python compares it: any number, a bool as 0 or 1. +/// Python compares a string level only with another string, and raises +/// otherwise. Here a string of an integer is read as a number, and any +/// other string fails. +fn level_of(value: &Value) -> anyhow::Result { + match value { + Value::Number(n) => n + .as_f64() + .ok_or_else(|| anyhow!("power level {n} is not a number")), + Value::Bool(b) => Ok(if *b { 1.0 } else { 0.0 }), + Value::String(s) => s + .trim() + .parse::() + .map(|level| level as f64) + .map_err(|_| anyhow!("power level {s:?} is not an integer")), + other => bail!("power level {other} is not a number"), + } +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + use crate::{events::internal_metadata::EventInternalMetadata, room_versions::RoomVersion}; + + #[test] + fn levels_read_as_python_compares_them() { + assert_eq!(level_of(&json!(50)).unwrap(), 50.0); + assert_eq!(level_of(&json!(5.5)).unwrap(), 5.5); + assert_eq!(level_of(&json!(true)).unwrap(), 1.0); + assert_eq!(level_of(&json!("50")).unwrap(), 50.0); + assert!(level_of(&json!("high")).is_err()); + assert!(level_of(&json!([])).is_err()); + assert_eq!(field_level(None).unwrap(), 0.0); + assert_eq!(field_level(Some(&Value::Null)).unwrap(), 0.0); + } + + fn event(event_type: &str, sender: &str, stream_ordering: i64) -> Event { + let json = json!({ + "auth_events": [], "prev_events": [], "type": event_type, + "sender": sender, "content": {}, "depth": 3, + "room_id": "!r:here", "origin_server_ts": 5, + "hashes": {"sha256": event_type}, "signatures": {}, "unsigned": {}, + }); + let metadata = EventInternalMetadata::from_json_str("{}").unwrap(); + metadata + .set_persisted_position(Some(stream_ordering), None, None) + .unwrap(); + Event::from_json_str(&json.to_string(), &RoomVersion::V10, metadata, None).unwrap() + } + + #[test] + fn chunks_are_counted_as_python_counts_them() { + let app_service = PersistContext { + app_service: true, + ..Default::default() + }; + let events = [ + ( + event("m.room.message", "@a:here", 4), + PersistContext::default(), + ), + ( + event("m.room.message", "@b:there", 5), + PersistContext::default(), + ), + ( + event("m.room.topic", "@a:here", 6), + PersistContext::default(), + ), + (event("m.room.member", "@bot:here", 7), app_service), + ]; + let counts = count_chunk(&events, "here", false); + assert_eq!(counts.persisted_events, 4); + let by_type = |event_type: &str, origin: &'static str| { + counts + .persisted_events_by_type + .get(&(event_type.to_owned(), origin)) + .copied() + }; + assert_eq!(by_type("m.room.message", "local"), Some(1)); + assert_eq!(by_type("m.room.message", "remote"), Some(1)); + assert_eq!(by_type("*other*", "local"), Some(1)); + assert_eq!(by_type("m.room.member", "application_service"), Some(1)); + assert_eq!(counts.persisted_events_by_type.len(), 4); + assert_eq!(counts.persisted_position, Some(7)); + + // No position for a backfilled chunk. + let counts = count_chunk(&events, "here", true); + assert_eq!(counts.persisted_position, None); + } +} diff --git a/rust/src/persist/mod.rs b/rust/src/persist/mod.rs index 953ee16a3e..89f6e03916 100644 --- a/rust/src/persist/mod.rs +++ b/rust/src/persist/mod.rs @@ -24,6 +24,8 @@ //! - [`effects`]: the Python cache updates a Rust transaction asks for once //! it commits. //! - [`extremities`]: the room's forward extremities after a chunk. +//! - [`follow_ups`]: what follows writing a chunk: metrics and the MSC4293 +//! redaction rows. //! - [`metrics`]: counts for the Python metrics, flushed when scraped. //! - [`prepare`]: everything a chunk needs before its write transaction, //! built from the steps above. @@ -44,6 +46,7 @@ pub mod chain_cover; pub mod context; pub mod effects; pub mod extremities; +pub mod follow_ups; pub mod metrics; pub mod prepare; pub mod python; diff --git a/rust/src/persist/python.rs b/rust/src/persist/python.rs index 7e1e59e99b..b06db15e77 100644 --- a/rust/src/persist/python.rs +++ b/rust/src/persist/python.rs @@ -30,6 +30,7 @@ use super::{ context::PersistContext, effects::PersistEffects, extremities::MissingPrevStateEvents, + follow_ups, metrics::{PersistCounts, PersistMetrics}, prepare::{self, PrepareInput, PrepareOptions, Prepared}, reader::{self, PersistReader, PyDelta, PySlidingSync}, @@ -410,19 +411,21 @@ impl Persister { /// [`WrittenChunk`] before the next step, and calls /// [`Self::batch_abort`] if it cannot. Converts the events, without /// I/O. - #[pyo3(signature = (room_id, events_and_contexts, *, backfilled))] + #[pyo3(signature = (room_id, events_and_contexts, *, backfilled, msc4293_enabled))] fn persist_batch( &self, py: Python<'_>, room_id: String, events_and_contexts: Vec<(Py, PersistContext)>, backfilled: bool, + msc4293_enabled: bool, ) -> PyResult { let state = BatchState { inner: Arc::clone(&self.inner), pool: self.inner.pool.get()?, room_id, backfilled, + msc4293_enabled, events: shallow_events(py, events_and_contexts), deduplicated: None, next: 0, @@ -624,6 +627,9 @@ struct BatchState { pool: ConnectionPool, room_id: String, backfilled: bool, + /// `experimental_features.msc4293_enabled`, read per batch as Python + /// reads it per chunk. + msc4293_enabled: bool, events: Vec<(Event, PersistContext)>, /// Set by the first step. deduplicated: Option, @@ -705,7 +711,7 @@ impl Drop for BatchState { /// What one [`batch_next`] step did. enum BatchStep { /// A chunk was written, and its effects are Python's to apply. - Written(WrittenChunk), + Written(Box), /// The batch is done. Done(PersistedBatch), } @@ -717,7 +723,7 @@ impl<'py> IntoPyObject<'py> for BatchStep { fn into_pyobject(self, py: Python<'py>) -> PyResult> { match self { - BatchStep::Written(chunk) => Ok(Bound::new(py, chunk)?.into_any()), + BatchStep::Written(chunk) => Ok(Bound::new(py, *chunk)?.into_any()), BatchStep::Done(batch) => Ok(Bound::new(py, batch)?.into_any()), } } @@ -836,6 +842,11 @@ async fn batch_step(state: &mut BatchState) -> PyResult { .map_err(write_error_to_py)?; state.guard_held = true; } + // The follow-ups read the events after the write sets their orderings. + let chunk_copy: Vec<(Event, PersistContext)> = chunk + .iter() + .map(|(event, context)| (event.shallow_clone(), context.clone())) + .collect(); let written = write_chunk( inner, pool, @@ -848,7 +859,7 @@ async fn batch_step(state: &mut BatchState) -> PyResult { state.backfilled, ) .await; - let effects = match written { + let mut effects = match written { Ok(effects) => effects, Err(e) => { // As Python's `finally`, whose error replaces the persist's. @@ -856,12 +867,62 @@ async fn batch_step(state: &mut BatchState) -> PyResult { return Err(e); } }; + // The chunk is written, so its effects go to Python whatever follows. + let follow_up_error = follow_ups(state, &chunk_copy, prepared_chunk, &mut effects) + .await + .err(); state.next = end; - Ok(BatchStep::Written(WrittenChunk { + Ok(BatchStep::Written(Box::new(WrittenChunk { effects, prepared, indices, - })) + follow_up_error: Mutex::new(follow_up_error), + }))) +} + +/// `_after_persisting_events` for a chunk Rust wrote: count it, record the +/// MSC4293 redactions it starts or ends, and prefill the room's forward +/// extremities. Runs before the chunk's stream orderings are finished, as +/// in Python, so the events are not announced before their redactions are +/// recorded. +async fn follow_ups( + state: &BatchState, + chunk: &[(Event, PersistContext)], + prepared: &prepare::PreparedChunk, + effects: &mut PersistEffects, +) -> PyResult<()> { + let (inner, pool, room_id) = (&state.inner, &state.pool, state.room_id.as_str()); + let counts = follow_ups::count_chunk(chunk, &inner.server_name, state.backfilled); + inner.metrics.record(|total| total.add(counts)); + if state.msc4293_enabled { + let reader = inner.sync.reader(pool.clone()); + let mut started = false; + let recorded = follow_ups::record_room_ban_redactions( + pool, + &reader, + room_id, + chunk, + || inner.events_stream.get_persisted_upto_position(), + &mut started, + ) + .await; + // Even if a later event failed, for those that started. + if started { + effects.add_event_cache_room_invalidation(room_id.to_owned()); + } + recorded.map_err(persist_error_to_py)?; + } + if let Some(extremities) = prepared + .state + .new_forward_extremities + .as_ref() + .filter(|e| !e.is_empty()) + { + let mut extremities: Vec = extremities.iter().map(|e| e.to_string()).collect(); + extremities.sort(); + effects.set_latest_event_ids(room_id.to_owned(), extremities); + } + Ok(()) } /// [`batch_next`]'s chunk, written, whose effects Python must apply. @@ -870,6 +931,9 @@ pub struct WrittenChunk { effects: PersistEffects, prepared: Arc, indices: Vec, + /// Why what follows the write failed, for Python to raise once it has + /// applied the effects. + follow_up_error: Mutex>, } #[pymethods] @@ -896,6 +960,21 @@ impl WrittenChunk { fn indices(&self) -> Vec { self.indices.clone() } + + /// Raise the error of what followed the write, if it failed. Python + /// calls this once it has applied the effects, as the chunk is written + /// either way. + fn check_follow_ups(&self) -> PyResult<()> { + match self + .follow_up_error + .lock() + .unwrap_or_else(|e| e.into_inner()) + .take() + { + Some(err) => Err(err), + None => Ok(()), + } + } } /// Where the stream orderings of a chunk come from. diff --git a/synapse/storage/controllers/persist_events.py b/synapse/storage/controllers/persist_events.py index ce287e8a2b..acffdc562a 100644 --- a/synapse/storage/controllers/persist_events.py +++ b/synapse/storage/controllers/persist_events.py @@ -82,7 +82,10 @@ from synapse.storage.databases.main.events import ( persist_event_counter, ) from synapse.storage.databases.main.events_worker import EventRedactBehaviour -from synapse.storage.databases.main.rust_persist import finish_rust_persist_stream_ids +from synapse.storage.databases.main.rust_persist import ( + apply_rust_persist_effects, + finish_rust_persist_stream_ids, +) from synapse.synapse_rust.persist import ( PersistBatch, Persister, @@ -966,7 +969,10 @@ class EventsPersistenceStorageController: assert self._rust_persister is not None persister = self._rust_persister handle = persister.persist_batch( - room_id, events_and_contexts, backfilled=backfilled + room_id, + events_and_contexts, + backfilled=backfilled, + msc4293_enabled=self.hs.config.experimental.msc4293_enabled, ) async def persist() -> dict[str, str]: @@ -1015,8 +1021,11 @@ class EventsPersistenceStorageController: handle: PersistBatch, ) -> None: """Apply what a chunk of the batch Rust wrote asks for, and finish - its stream orderings. If that fails, release the batch.""" + its stream orderings, as a failure if what Rust ran after writing + it failed, as `get_next_mult` exits with an exception. Then raise + that error. If anything fails, release the batch.""" assert self._rust_persister is not None + store = self.persist_events_store.store effects = step.effects try: try: @@ -1028,19 +1037,17 @@ class EventsPersistenceStorageController: self._state_resolution_handler.record_rust_state_res_metrics( room_id, prepared.state_res_stats ) - except BaseException: - # Only the apply finishes the orderings otherwise. - finish_rust_persist_stream_ids( - self.persist_events_store.store, effects, succeeded=False + apply_rust_persist_effects( + store, + effects, + self.persist_events_store._events_written(chunk), + finish_stream_ids=False, ) + step.check_follow_ups() + except BaseException: + finish_rust_persist_stream_ids(store, effects, succeeded=False) raise - await self.persist_events_store._apply_rust_chunk( - room_id, - chunk, - effects, - backfilled=backfilled, - new_forward_extremities=prepared.new_forward_extremities, - ) + finish_rust_persist_stream_ids(store, effects, succeeded=True) except BaseException: await self._rust_persister.batch_abort(handle) raise diff --git a/synapse/storage/databases/main/rust_persist.py b/synapse/storage/databases/main/rust_persist.py index af257eb5fc..f8019af000 100644 --- a/synapse/storage/databases/main/rust_persist.py +++ b/synapse/storage/databases/main/rust_persist.py @@ -120,6 +120,11 @@ def apply_rust_persist_effects( assert events is not None store._get_event_cache.set_local((event_id,), events[event_id], None) + # After the prefills, as Python's `_after_persisting_events` drops + # the room's events once its transaction's callbacks have run. + for room_id in effects.event_cache_room_invalidations: + store._invalidate_local_get_event_cache_room_id(room_id) + for event_id, state_group in effects.state_group_prefills: store._get_state_group_for_event.prefill((event_id,), state_group) diff --git a/synapse/synapse_rust/persist.pyi b/synapse/synapse_rust/persist.pyi index e2d94e38e1..07e47416e4 100644 --- a/synapse/synapse_rust/persist.pyi +++ b/synapse/synapse_rust/persist.pyi @@ -112,12 +112,14 @@ class Persister: events_and_contexts: list[tuple[EventBase, EventContext]], *, backfilled: bool, + msc4293_enabled: bool, ) -> PersistBatch: """`_persist_event_batch` for a queued batch of `room_id`'s events, one chunk at a time: await `batch_next` until it resolves with a `PersistedBatch`, applying the effects of each `WrittenChunk` before the next step, and call `batch_abort` if they cannot be applied. - Converts the events, without I/O.""" + With `msc4293_enabled`, each step records the MSC4293 redactions + its chunk starts or ends. Converts the events, without I/O.""" def batch_next( self, batch: PersistBatch ) -> Deferred[WrittenChunk | PersistedBatch]: @@ -154,6 +156,10 @@ class WrittenChunk: @property def indices(self) -> list[int]: """The chunk's positions in the batch.""" + def check_follow_ups(self) -> None: + """Raise the error of what followed the write (the MSC4293 + redaction rows), if it failed. Call once the effects are applied, + as the chunk is written either way.""" class PersistedBatch: """What `Persister.persist_batch` did, once the batch is done.""" @@ -229,6 +235,7 @@ class PersistEffects: sticky_stream_ids: list[int] = ..., profile_stream_ids: list[int] = ..., event_cache_prefills: list[str] = ..., + event_cache_room_invalidations: list[str] = ..., state_group_prefills: list[tuple[str, int]] = ..., latest_event_ids: tuple[str, list[str]] | None = None, stream_changes: list[tuple[str, str, int]] = ..., @@ -283,6 +290,10 @@ class PersistEffects: those neither rejected nor redacted. The caller holds the events.""" @property + def event_cache_room_invalidations(self) -> list[str]: + """`_invalidate_local_get_event_cache_room_id(room_id)`, after the + prefills.""" + @property def state_group_prefills(self) -> list[tuple[str, int]]: """`(event ID, state group)` for `_get_state_group_for_event`.""" diff --git a/tests/storage/test_rust_persist_prereqs.py b/tests/storage/test_rust_persist_prereqs.py index 8ca9681588..cddb4d5201 100644 --- a/tests/storage/test_rust_persist_prereqs.py +++ b/tests/storage/test_rust_persist_prereqs.py @@ -1246,47 +1246,103 @@ class RustFullTestCase(unittest.HomeserverTestCase): (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]] = [] + def _join_bob(self) -> str: + bob = self.register_user("bob", "pass") + bob_token = self.login("bob", "pass") + self.helper.join(self.room_id, bob, tok=bob_token) + return bob - 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, + def _redaction_rows(self) -> list[tuple[str, str]]: + return self.get_success( + self.store.db_pool.simple_select_list( + "room_ban_redactions", None, ("user_id", "redacting_event_id") + ) + ) + + def test_follow_ups_run_before_orderings_finish(self) -> None: + """What follows the write, the MSC4293 redactions among it, runs in + Rust before Python applies the chunk and the token passes it, as + inside `get_next_mult`.""" + self.hs.config.experimental.msc4293_enabled = True + bob = self._join_bob() + id_gen = self.store._stream_id_gen + seen: list[tuple[int, list[int], list[tuple[str, str]]]] = [] + + def applying( + store: Any, effects: PersistEffects, *args: Any, **kwargs: Any + ) -> None: + rows = self.get_success( + self.store.db_pool.simple_select_list( + "room_ban_redactions", None, ("user_id", "redacting_event_id") ) ) - await original(room_id, events_and_contexts, **kwargs) + seen.append((id_gen.get_current_token(), effects.stream_ids, rows)) + apply_rust_persist_effects(store, effects, *args, **kwargs) - with patch.object( - persist_events, "_after_persisting_events", side_effect=after + with patch( + "synapse.storage.controllers.persist_events.apply_rust_persist_effects", + side_effect=applying, ): - self.helper.send(self.room_id, "hello", tok=self.token) - ((token, stream_ordering),) = seen - assert stream_ordering is not None + ban_id = self.helper.change_membership( + self.room_id, + self.user, + bob, + "ban", + {"org.matrix.msc4293.redact_events": True}, + tok=self.token, + )["event_id"] + ((token, (stream_ordering,), rows),) = seen self.assertLess(token, stream_ordering) + self.assertEqual(rows, [(bob, ban_id)]) 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 + def test_follow_up_failure(self) -> None: + """If what follows the write fails, the chunk is written and + applied, its orderings finished and its state groups released, and + the persist fails, as Python's would.""" + self.hs.config.experimental.msc4293_enabled = True + bob = self._join_bob() + # Python indexes the current power levels, so fails without them. + self.get_success( + self.store.db_pool.simple_delete( + "current_state_events", + {"room_id": self.room_id, "type": EventTypes.PowerLevels}, + desc="test", ) - failed = event.internal_metadata.stream_ordering - assert failed is not None - self.assertGreaterEqual(id_gen.get_current_token(), failed) + ) + event, context = self._build("hello") + self.get_success(self.persistence.persist_events([(event, context)])) + ban, ban_context = self.get_success( + self.hs.get_event_creation_handler().create_event( + create_requester(self.user), + { + "type": EventTypes.Member, + "state_key": bob, + "content": { + "membership": "ban", + "org.matrix.msc4293.redact_events": True, + }, + "room_id": self.room_id, + "sender": self.user, + }, + ) + ) + context = self.get_success(ban_context.persist(ban)) + failure = self.get_failure( + self.persistence.persist_events([(ban, context)]), RuntimeError + ) + self.assertIn("power_levels", str(failure.value)) + stream_ordering = self._event_row(ban.event_id)[0] + self.assertGreaterEqual( + self.store._stream_id_gen.get_current_token(), stream_ordering + ) + self.assertEqual(self._redaction_rows(), []) + rows = self.get_success( + self.store.db_pool.simple_select_list( + "state_groups_persisting", None, ("state_group",) + ) + ) + self.assertEqual(rows, []) def test_batch_is_one_rust_call(self) -> None: """A batch of 150 events is one Rust call, written in two chunks, @@ -1336,11 +1392,9 @@ class RustFullTestCase(unittest.HomeserverTestCase): """The batch's state groups are marked as being persisted while each chunk is written, and no longer once it is.""" held: list[list[tuple[int, str]]] = [] - persist_events = self.hs.get_datastores().persist_events - assert persist_events is not None - apply_chunk = persist_events._apply_rust_chunk + apply_chunk = self.persistence._apply_rust_batch_chunk - async def checking(*args: Any, **kwargs: Any) -> None: + async def checking(*args: Any) -> None: held.append( await self.store.db_pool.simple_select_list( "state_groups_persisting", @@ -1348,10 +1402,12 @@ class RustFullTestCase(unittest.HomeserverTestCase): ("state_group", "instance_name"), ) ) - await apply_chunk(*args, **kwargs) + await apply_chunk(*args) event, context = self._build("hello") - with patch.object(persist_events, "_apply_rust_chunk", side_effect=checking): + with patch.object( + self.persistence, "_apply_rust_batch_chunk", side_effect=checking + ): self.get_success(self.persistence.persist_events([(event, context)])) assert context.state_group is not None assert context.state_group_before_event is not None @@ -1655,6 +1711,226 @@ class RustFullTestCase(unittest.HomeserverTestCase): ) +class RustFullRedactionsTestCase(unittest.HomeserverTestCase): + """`rust_event_persistence: full` with MSC4293: Rust records the + redactions a ban or kick starts, and the joins and leaves that end + them, as `_after_persisting_events` does.""" + + servlets = [ + admin.register_servlets, + room.register_servlets, + login.register_servlets, + ] + + def default_config(self) -> dict[str, Any]: + config = super().default_config() + experimental = config.setdefault("experimental_features", {}) + experimental["rust_event_persistence"] = "full" + experimental["msc4293_enabled"] = True + 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.alice = self.register_user("alice", "pass") + self.alice_token = self.login("alice", "pass") + self.carol = self.register_user("carol", "pass") + self.carol_token = self.login("carol", "pass") + self.dave = self.register_user("dave", "pass") + self.dave_token = self.login("dave", "pass") + + def _room(self, room_version: str | None = None) -> str: + """A room Alice created, which Carol and Dave joined.""" + room_id = self.helper.create_room_as( + self.alice, room_version=room_version, tok=self.alice_token + ) + for user, token in ( + (self.carol, self.carol_token), + (self.dave, self.dave_token), + ): + self.helper.invite(room_id, self.alice, user, tok=self.alice_token) + self.helper.join(room_id, user, tok=token) + return room_id + + def _rows(self, room_id: str) -> list[tuple[str, str, int | None]]: + return self.get_success( + self.store.db_pool.simple_select_list( + "room_ban_redactions", + {"room_id": room_id}, + ("user_id", "redacting_event_id", "redact_end_ordering"), + ) + ) + + def _membership( + self, + room_id: str, + sender: str, + token: str, + target: str, + membership: str, + *, + redact: bool = False, + ) -> str: + return self.helper.change_membership( + room_id, + sender, + target, + membership, + {"org.matrix.msc4293.redact_events": True} if redact else {}, + tok=token, + )["event_id"] + + def _stream_ordering(self, event_id: str) -> int: + return self.get_success(self.store.get_position_for_event(event_id)).stream + + def test_redaction_power(self) -> None: + """Whether Carol, not a creator, may start redacting Dave's events + follows the power levels as `_can_sender_redact` reads them.""" + for levels, allowed in ( + ({"users": {self.carol: 75}, "redact": 50}, True), + ( + { + "users": {self.carol: 75}, + "redact": 50, + "events": {"m.room.redaction": 100}, + }, + False, + ), + ({"users": {self.carol: 75}, "events_default": 80}, False), + ({"users": {self.dave: 0}, "users_default": 75}, True), + ): + room_id = self._room() + self.helper.send_state( + room_id, EventTypes.PowerLevels, levels, tok=self.alice_token + ) + ban = self._membership( + room_id, self.carol, self.carol_token, self.dave, "ban", redact=True + ) + self.assertEqual( + self._rows(room_id), [(self.dave, ban, None)] if allowed else [], levels + ) + + def test_redaction_power_before_msc4289(self) -> None: + """In a room version without creator power, the power levels name + the creator too.""" + room_id = self._room(room_version="11") + self.helper.send_state( + room_id, + EventTypes.PowerLevels, + {"users": {self.alice: 100, self.carol: 75}, "redact": 50}, + tok=self.alice_token, + ) + ban = self._membership( + room_id, self.carol, self.carol_token, self.dave, "ban", redact=True + ) + self.assertEqual(self._rows(room_id), [(self.dave, ban, None)]) + + def test_unban_ends_and_ban_starts_again(self) -> None: + """An unban ends the redactions at its stream ordering. A ban after + starts them again, without an end.""" + room_id = self._room() + self._membership( + room_id, self.alice, self.alice_token, self.dave, "ban", redact=True + ) + unban = self._membership( + room_id, self.alice, self.alice_token, self.dave, "leave" + ) + ((user, _, end),) = self._rows(room_id) + self.assertEqual(end, self._stream_ordering(unban)) + + self.helper.invite(room_id, self.alice, self.dave, tok=self.alice_token) + self.helper.join(room_id, self.dave, tok=self.dave_token) + ban = self._membership( + room_id, self.alice, self.alice_token, self.dave, "ban", redact=True + ) + self.assertEqual(self._rows(room_id), [(self.dave, ban, None)]) + + def test_self_leave_keeps_redactions(self) -> None: + """A user leaving of their own accord, by rejecting an invite after + a kick, ends nothing.""" + room_id = self._room() + kick = self._membership( + room_id, self.alice, self.alice_token, self.dave, "leave", redact=True + ) + self.helper.invite(room_id, self.alice, self.dave, tok=self.alice_token) + self.helper.leave(room_id, self.dave, tok=self.dave_token) + self.assertEqual(self._rows(room_id), [(self.dave, kick, None)]) + + def test_backfilled_join_ends_at_current_token(self) -> None: + """A backfilled join ends the redactions at the events stream's + current token, not its negative ordering.""" + room_id = self._room() + self._membership( + room_id, self.alice, self.alice_token, self.dave, "ban", redact=True + ) + self._membership(room_id, self.alice, self.alice_token, self.dave, "leave") + self.helper.invite(room_id, self.alice, self.dave, tok=self.alice_token) + event, unpersisted = self.get_success( + self.hs.get_event_creation_handler().create_event( + create_requester(self.dave), + { + "type": EventTypes.Member, + "state_key": self.dave, + "content": {"membership": "join"}, + "room_id": room_id, + "sender": self.dave, + }, + ) + ) + context = self.get_success(unpersisted.persist(event)) + self.get_success( + self.persistence.persist_events([(event, context)], backfilled=True) + ) + token = self.store._stream_id_gen.get_current_token() + ((_, _, end),) = self._rows(room_id) + self.assertEqual(end, token) + + def test_redacted_in_the_same_chunk(self) -> None: + """A message and the ban that redacts it, persisted together: the + message is dropped from the event cache after the chunk's prefills, + so reading it again shows it redacted.""" + room_id = self._room() + pairs = [] + for event_dict in ( + { + "type": EventTypes.Message, + "content": {"msgtype": "m.text", "body": "noise"}, + "sender": self.dave, + }, + { + "type": EventTypes.Member, + "state_key": self.dave, + "content": { + "membership": "ban", + "org.matrix.msc4293.redact_events": True, + }, + "sender": self.alice, + }, + ): + event, unpersisted = self.get_success( + self.hs.get_event_creation_handler().create_event( + create_requester(str(event_dict["sender"])), + {**event_dict, "room_id": room_id}, + ) + ) + pairs.append((event, self.get_success(unpersisted.persist(event)))) + self.get_success(self.persistence.persist_events(pairs)) + message = self.get_success(self.store.get_event(pairs[0][0].event_id)) + self.assertEqual(message.content, {}) + + class RustTransactionWorkerTestCase(BaseMultiWorkerStreamTestCase): """`rust_event_persistence: transaction` on a worker event persister: the worker writes in Rust, to the main database."""