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."""