diff --git a/rust/src/config/mod.rs b/rust/src/config/mod.rs index 14dfba3212..dcd6fcd8d2 100644 --- a/rust/src/config/mod.rs +++ b/rust/src/config/mod.rs @@ -50,6 +50,8 @@ pub struct AuthConfig { } #[derive(FromPyObject, Clone)] pub struct ServerConfig { + pub enable_search: bool, + pub enable_ephemeral_messages: bool, pub msc4140_enabled: bool, pub public_baseurl: String, pub include_profile_updates_in_sync: bool, diff --git a/rust/src/events/constants.rs b/rust/src/events/constants.rs index 596f10c1b6..6db8131a39 100644 --- a/rust/src/events/constants.rs +++ b/rust/src/events/constants.rs @@ -21,6 +21,18 @@ pub mod event_type { pub const M_ROOM_HISTORY_VISIBILITY: &str = "m.room.history_visibility"; /// Event type: m.room.redaction pub const M_ROOM_REDACTION: &str = "m.room.redaction"; + /// Event type: m.room.name + pub const M_ROOM_NAME: &str = "m.room.name"; + /// Event type: m.room.topic + pub const M_ROOM_TOPIC: &str = "m.room.topic"; + /// Event type: m.room.message + pub const M_ROOM_MESSAGE: &str = "m.room.message"; + /// Event type: m.room.retention + pub const M_ROOM_RETENTION: &str = "m.room.retention"; + /// Event type: m.room.encryption + pub const M_ROOM_ENCRYPTION: &str = "m.room.encryption"; + /// Event type: m.room.tombstone + pub const M_ROOM_TOMBSTONE: &str = "m.room.tombstone"; } /// Event Fields diff --git a/rust/src/events/internal_metadata.rs b/rust/src/events/internal_metadata.rs index 698caacc1a..8c4209bd50 100644 --- a/rust/src/events/internal_metadata.rs +++ b/rust/src/events/internal_metadata.rs @@ -720,7 +720,7 @@ impl EventInternalMetadata { Ok(self.read_inner()?.is_redacted()) } - fn is_notifiable(&self) -> PyResult { + pub fn is_notifiable(&self) -> PyResult { Ok(self.read_inner()?.is_notifiable()) } diff --git a/rust/src/events/unsigned.rs b/rust/src/events/unsigned.rs index c2e7cd3773..34ab555323 100644 --- a/rust/src/events/unsigned.rs +++ b/rust/src/events/unsigned.rs @@ -309,6 +309,18 @@ impl Unsigned { Ok(self.py_read()?.persisted_fields.age_ts.clone()) } + /// The `invite_room_state` field: the stripped state of the room an + /// invite is for. + pub fn invite_room_state(&self) -> PyResult>> { + Ok(self.py_read()?.persisted_fields.invite_room_state.clone()) + } + + /// The `knock_room_state` field: the stripped state of the room a knock + /// is for. + pub fn knock_room_state(&self) -> PyResult>> { + Ok(self.py_read()?.persisted_fields.knock_room_state.clone()) + } + /// Get the `prev_content` field: the content of the state this event /// replaced, if it is a state event and the content is known. pub fn prev_content(&self) -> PyResult>> { diff --git a/rust/src/persist/effects.rs b/rust/src/persist/effects.rs index ff8928aac3..6bf4e99da9 100644 --- a/rust/src/persist/effects.rs +++ b/rust/src/persist/effects.rs @@ -123,6 +123,10 @@ pub struct PersistEffects { /// `cache.invalidate(keys)` for the named cache: the local half of /// `_invalidate_cache_and_stream`. cache_invalidations: Vec<(String, Vec)>, + /// The IDs of the cache stream rows the transaction wrote, to mark + /// finished once the caches are invalidated, then + /// `ReplicationNotifier.notify_replication()`. + cache_stream_ids: 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)`. @@ -157,6 +161,14 @@ impl PersistEffects { self.cache_invalidations.push((cache_name.to_owned(), keys)); } + pub fn add_cache_stream_id(&mut self, stream_id: i64) { + self.cache_stream_ids.push(stream_id); + } + + pub fn cache_stream_ids(&self) -> &[i64] { + &self.cache_stream_ids + } + pub fn add_state_group_prefill(&mut self, event_id: String, state_group: i64) { self.state_group_prefills.push((event_id, state_group)); } @@ -194,6 +206,7 @@ impl PersistEffects { event_cache_invalidations = Vec::new(), state_caches = Vec::new(), cache_invalidations = Vec::new(), + cache_stream_ids = Vec::new(), state_group_prefills = Vec::new(), latest_event_ids = None, stream_changes = Vec::new(), @@ -207,6 +220,7 @@ impl PersistEffects { event_cache_invalidations: Vec, state_caches: Vec<(String, Vec)>, cache_invalidations: Vec<(String, Vec)>, + cache_stream_ids: Vec, state_group_prefills: Vec<(String, i64)>, latest_event_ids: Option<(String, Vec)>, stream_changes: Vec<(String, String, i64)>, @@ -219,6 +233,7 @@ impl PersistEffects { event_cache_invalidations, state_caches, cache_invalidations, + cache_stream_ids, state_group_prefills, latest_event_ids, stream_changes: stream_changes @@ -267,6 +282,12 @@ impl PersistEffects { self.cache_invalidations.clone() } + #[getter] + #[pyo3(name = "cache_stream_ids")] + fn py_cache_stream_ids(&self) -> Vec { + self.cache_stream_ids.clone() + } + #[getter] fn state_group_prefills(&self) -> Vec<(String, i64)> { self.state_group_prefills.clone() diff --git a/rust/src/persist/python.rs b/rust/src/persist/python.rs index 5ea6f1083f..45365261f4 100644 --- a/rust/src/persist/python.rs +++ b/rust/src/persist/python.rs @@ -34,7 +34,7 @@ use super::{ prepare::{self, PrepareInput, PrepareOptions, Prepared}, reader::{self, PersistReader, PyDelta, PySlidingSync}, state_delta::NativeStateResolver, - txn::{self, json, PartialStateConflict, TxnEvent, TxnInput}, + txn::{self, caches::CacheStream, json, PartialStateConflict, TxnEvent, TxnInput}, }; use crate::{ database::postgres::write::{ @@ -87,6 +87,10 @@ pub(super) struct PersisterInner { pub conflict_cache: Arc, /// `experimental_features.msc4354_enabled`. pub msc4354_enabled: bool, + /// `server.enable_search`. + pub enable_search: bool, + /// `server.enable_ephemeral_messages`. + pub enable_ephemeral_messages: bool, } impl Persister { @@ -151,6 +155,8 @@ impl Persister { metrics: PersistMetrics::default(), conflict_cache, msc4354_enabled: config.experimental.msc4354_enabled, + enable_search: config.server.enable_search, + enable_ephemeral_messages: config.server.enable_ephemeral_messages, }), })) } @@ -276,17 +282,21 @@ impl Persister { /// `prepared` is what Rust prepared for it. For /// `tests/storage/test_rust_persist_txn.py`. /// + /// The cache stream IDs it wrote come back unfinished in the effects. + /// /// Returns what it collected for Python to apply, and for the Python /// steps after it, by position in the chunk: the events left, the /// events stored redacted with the event that redacts each, and the /// events that stopped being outliers. #[allow(clippy::type_complexity)] + #[pyo3(signature = (room_id, events_and_contexts, prepared, *, inhibit_local_membership_updates))] fn persist_events_txn_for_test( &self, py: Python<'_>, room_id: String, events_and_contexts: Vec<(Py, PersistContext)>, prepared: &PreparedChunk, + inhibit_local_membership_updates: bool, ) -> PyResult<( PersistEffects, Vec, @@ -308,12 +318,18 @@ impl Persister { let Prepared::Chunk(chunk) = &*prepared else { unreachable!("checked above"); }; + let cache_stream = CacheStream::new(Arc::clone(&inner.cache_stream)); let input = TxnInput { room_id: &room_id, events: &events, prepared: chunk, instance_name: &inner.instance_name, + server_name: &inner.server_name, clock: inner.runtime.clock(), + inhibit_local_membership_updates, + enable_search: inner.enable_search, + enable_ephemeral_messages: inner.enable_ephemeral_messages, + cache_stream: &cache_stream, }; let position = |event: &TxnEvent| { input @@ -322,7 +338,7 @@ impl Persister { .position(|x| std::ptr::eq(x, event)) .expect("an event of the chunk") }; - run_write_interaction(&pool, "persist_events", |txn, _| { + let result = run_write_interaction(&pool, "persist_events", |txn, _| { let (input, position) = (&input, &position); async move { let mut effects = PersistEffects::default(); @@ -345,7 +361,9 @@ impl Persister { .boxed() }) .await - .map_err(write_error_to_py) + .map_err(write_error_to_py)?; + cache_stream.hand_over(&result.0); + Ok(result) }; self.block_on(py, future.boxed()) } diff --git a/rust/src/persist/sliding_sync.rs b/rust/src/persist/sliding_sync.rs index 75a3ccf77d..26e82998d1 100644 --- a/rust/src/persist/sliding_sync.rs +++ b/rust/src/persist/sliding_sync.rs @@ -30,8 +30,8 @@ //! The write transaction treats both as sets. //! //! `_get_sliding_sync_insert_values_from_stripped_state` belongs to the -//! write transaction, which calls it for out-of-band memberships, and is -//! ported with it. +//! write transaction, which calls it for out-of-band memberships. It is in +//! [`super::txn::metadata`]. use std::{ collections::{HashMap, HashSet}, diff --git a/rust/src/persist/txn/caches.rs b/rust/src/persist/txn/caches.rs new file mode 100644 index 0000000000..fa90092e6b --- /dev/null +++ b/rust/src/persist/txn/caches.rs @@ -0,0 +1,215 @@ +/* + * 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: + * . + * + */ + +//! The cache stream: `_invalidate_cache_and_stream`, for the caches that +//! other workers cannot tell they must invalidate from the other streams. +//! +//! Each invalidation writes a `cache_invalidation_stream_by_instance` row +//! inside the transaction, with an ID from the cache stream's +//! [`StreamIdTracker`], and asks Python to invalidate the cache locally. +//! The IDs stay unfinished until Python has done so, as Python's +//! `get_next_txn` finishes them only after the transaction's earlier +//! `call_after` callbacks. + +use std::sync::{Arc, Mutex}; + +use anyhow::anyhow; + +use super::TxnInput; +use crate::{ + database::postgres::write::{WriteError, WriteTxn}, + persist::effects::PersistEffects, + storage::stream_ids::StreamIdTracker, +}; + +/// The cache stream IDs of one persist, over all its attempts. An ID the +/// committed attempt wrote is handed to Python with the effects. The rest +/// are marked finished here: at the start of the next attempt, or when +/// this is dropped. +pub struct CacheStream { + tracker: Arc, + /// The IDs allocated and neither finished nor handed to Python. + pending: Mutex>, +} + +impl CacheStream { + pub fn new(tracker: Arc) -> Self { + CacheStream { + tracker, + pending: Mutex::new(Vec::new()), + } + } + + fn pending(&self) -> std::sync::MutexGuard<'_, Vec> { + self.pending.lock().unwrap_or_else(|e| e.into_inner()) + } + + /// `_load_next_id_txn`: the next ID on the stream, unfinished. + async fn next_id(&self, txn: &WriteTxn) -> Result { + let fetch = Fetch::begin(self); + let rows = txn + .query("SELECT nextval('cache_invalidation_stream_seq')", &[]) + .await?; + let row = rows + .first() + .ok_or_else(|| anyhow!("nextval returned no rows"))?; + let id = row.try_get(0).map_err(anyhow::Error::new)?; + fetch.allocated(id); + Ok(id) + } + + /// Mark finished the IDs of an attempt that failed, as Python's + /// `call_on_exception` does. + pub fn finish_pending(&self) { + let ids = std::mem::take(&mut *self.pending()); + if !ids.is_empty() { + self.tracker.mark_finished(&ids); + } + } + + /// The transaction that wrote `effects` committed: Python marks its IDs + /// finished once it has applied them. + pub fn hand_over(&self, effects: &PersistEffects) { + let handed = effects.cache_stream_ids(); + self.pending().retain(|id| !handed.contains(id)); + } +} + +/// A fetch of an ID from the sequence, between the tracker's `begin_fetch` +/// and `end_fetch`. Python's `try`/`finally` cannot be interrupted, but a +/// future can be dropped at any await, so the fetch ends on drop if it did +/// not allocate. Otherwise the stream's position would stay behind it. +struct Fetch<'a> { + stream: &'a CacheStream, + token: Option, +} + +impl<'a> Fetch<'a> { + fn begin(stream: &'a CacheStream) -> Self { + Fetch { + stream, + token: Some(stream.tracker.begin_fetch()), + } + } + + /// The fetch allocated `id`, which stays unfinished. + fn allocated(mut self, id: i64) { + if let Some(token) = self.token.take() { + self.stream.tracker.end_fetch(token, Some(&[id])); + self.stream.pending().push(id); + } + } +} + +impl Drop for Fetch<'_> { + fn drop(&mut self) { + if let Some(token) = self.token.take() { + self.stream.tracker.end_fetch(token, None); + } + } +} + +impl Drop for CacheStream { + fn drop(&mut self) { + self.finish_pending(); + } +} + +/// `_invalidate_cache_and_stream`: invalidate `keys` in the named cache +/// here once the transaction commits, and on the other workers through the +/// cache stream. +pub async fn invalidate_cache_and_stream( + txn: &WriteTxn, + input: &TxnInput<'_>, + effects: &mut PersistEffects, + cache_name: &str, + keys: Vec, +) -> Result<(), WriteError> { + let stream_id = input.cache_stream.next_id(txn).await?; + txn.execute( + "INSERT INTO cache_invalidation_stream_by_instance + (stream_id, instance_name, cache_func, keys, invalidation_ts) + VALUES ($1, $2, $3, $4, $5)", + &[ + &stream_id, + &input.instance_name, + &cache_name, + &keys, + &input.now_ms()?, + ], + ) + .await?; + effects.add_cache_invalidation(cache_name, keys); + effects.add_cache_stream_id(stream_id); + effects.set_new_replication_data(); + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Allocate the next ID, as `next_id` does without the sequence. + fn allocate(stream: &CacheStream, id: i64) { + Fetch::begin(stream).allocated(id); + } + + #[test] + fn ids_are_finished_unless_handed_over() { + let tracker = Arc::new(StreamIdTracker::new( + "caches".to_owned(), + "master".to_owned(), + true, + )); + let stream = CacheStream::new(Arc::clone(&tracker)); + + // An attempt that failed: the next one finishes its IDs. + allocate(&stream, 2); + allocate(&stream, 3); + assert_eq!(tracker.get_persisted_upto_position(), 1); + stream.finish_pending(); + assert_eq!(tracker.get_persisted_upto_position(), 3); + + // The attempt that committed hands its IDs to Python. + allocate(&stream, 4); + let mut effects = PersistEffects::default(); + effects.add_cache_stream_id(4); + stream.hand_over(&effects); + // Allocated after the hand-over, so not Python's: finished on drop. + allocate(&stream, 5); + drop(stream); + assert_eq!(tracker.get_persisted_upto_position(), 3); + + // Python finishes the IDs it was handed. + tracker.mark_finished(&[4]); + assert_eq!(tracker.get_persisted_upto_position(), 5); + } + + #[test] + fn a_dropped_fetch_ends() { + let tracker = Arc::new(StreamIdTracker::new( + "caches".to_owned(), + "master".to_owned(), + true, + )); + let stream = CacheStream::new(Arc::clone(&tracker)); + allocate(&stream, 2); + // Dropped before the sequence answered, as a cancelled future is. + drop(Fetch::begin(&stream)); + stream.finish_pending(); + // A fetch still in flight would hold the position at 1. + assert_eq!(tracker.get_persisted_upto_position(), 2); + } +} diff --git a/rust/src/persist/txn/edges.rs b/rust/src/persist/txn/edges.rs index 15dc2d4d85..54d7454730 100644 --- a/rust/src/persist/txn/edges.rs +++ b/rust/src/persist/txn/edges.rs @@ -13,14 +13,18 @@ * */ -//! The room's extremities: `_update_forward_extremities_txn`, -//! `_set_state_dag_extremities_txn` and `_update_backward_extremeties`. +//! The room's DAG: the extremities (`_update_forward_extremities_txn`, +//! `_set_state_dag_extremities_txn` and `_update_backward_extremeties`) and +//! the edges (`_handle_mult_prev_events` and `_store_state_dag_edges`). use std::{collections::HashSet, sync::Arc}; +use anyhow::anyhow; + +use super::Persisting; use crate::{ database::postgres::write::{Column, WriteError, WriteTxn}, - events::Event, + events::{constants::event_type, Event}, }; /// `_update_forward_extremities_txn`: replace the room's forward @@ -193,3 +197,89 @@ pub async fn update_backward_extremities( } Ok(()) } + +/// `_handle_mult_prev_events`: the `event_edges` of `events`, then their +/// backward extremities. `events` must be of one room. +pub async fn handle_mult_prev_events( + txn: &WriteTxn, + instance_name: &str, + events: &[Persisting<'_>], +) -> Result<(), WriteError> { + let mut event_ids = Vec::new(); + let mut prev_event_ids = Vec::new(); + let mut backward = Vec::with_capacity(events.len()); + for persisting in events { + let event = persisting.event(); + for prev in event.prev_event_ids() { + event_ids.push(event.event_id()); + prev_event_ids.push(prev); + } + backward.push(BackwardEvent { + event, + outlier: persisting.source.is_outlier()?, + stream_ordering: persisting.stream_ordering(), + }); + } + txn.insert_many( + "event_edges", + &[ + Column::new("event_id", "text", &event_ids), + Column::new("prev_event_id", "text", &prev_event_ids), + ], + None, + ) + .await?; + update_backward_extremities(txn, instance_name, &backward).await +} + +/// `_store_state_dag_edges`: the MSC4242 state DAG edges of the state +/// events of `events` in a room version with a state DAG. +pub async fn store_state_dag_edges( + txn: &WriteTxn, + events: &[Persisting<'_>], +) -> Result<(), WriteError> { + let mut room_ids = Vec::new(); + let mut event_ids = Vec::new(); + let mut prev_state_event_ids: Vec> = Vec::new(); + for persisting in events { + let event = persisting.event(); + if !event.room_version().msc4242_state_dags || event.state_key().is_none() { + continue; + } + let prevs = event.prev_state_event_ids().unwrap_or_default(); + if prevs.is_empty() { + // `get_state_dag` reads every row, so the create event, which has + // no edge, gets one to nothing. + if event.type_() != event_type::M_ROOM_CREATE { + return Err(anyhow!( + "state DAG event {} has no prev_state_events", + event.event_id() + ) + .into()); + } + room_ids.push(event.room_id()); + event_ids.push(event.event_id()); + prev_state_event_ids.push(None); + continue; + } + for prev in prevs { + room_ids.push(event.room_id()); + event_ids.push(event.event_id()); + prev_state_event_ids.push(Some(prev)); + } + } + // Python inserts the create event's row and upserts the rest. Only the + // create event has a NULL edge, and the key of a NULL never conflicts, + // so one upsert does both. + txn.upsert_many( + "msc4242_state_dag_edges", + &[ + Column::new("room_id", "text", &room_ids), + Column::new("event_id", "text", &event_ids), + Column::new("prev_state_event_id", "text", &prev_state_event_ids), + ], + &[], + ) + .await?; + Ok(()) +} diff --git a/rust/src/persist/txn/metadata.rs b/rust/src/persist/txn/metadata.rs new file mode 100644 index 0000000000..b0c260154c --- /dev/null +++ b/rust/src/persist/txn/metadata.rs @@ -0,0 +1,1011 @@ +/* + * 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: + * . + * + */ + +//! `_update_metadata_tables_txn`: the tables about the events, written once +//! the events themselves are stored. Push actions, redactions, relations +//! and threads, search, retention, labels, expiry and room members. +//! +//! The event cache prefill at its end, `_add_to_cache`, stays in Python. + +use anyhow::anyhow; +use serde_json::{Map, Value}; +use tokio_postgres::types::ToSql; + +use super::{caches::invalidate_cache_and_stream, edges, redacts, relation, Persisting, TxnInput}; +use crate::events::constants::event_type::{ + M_ROOM_CREATE as CREATE, M_ROOM_ENCRYPTION as ENCRYPTION, M_ROOM_MESSAGE as MESSAGE, + M_ROOM_NAME as NAME, M_ROOM_RETENTION as RETENTION, M_ROOM_TOMBSTONE as TOMBSTONE, + M_ROOM_TOPIC as TOPIC, +}; +use crate::{ + database::postgres::write::{Column, WriteError, WriteTxn}, + events::{ + constants::{event_type, relation_type}, + json_object::JsonObject, + Event, + }, + persist::{effects::PersistEffects, extremities::is_mine_id}, +}; + +/// `EventContentFields.LABELS`. +const LABELS: &str = "org.matrix.labels"; +/// `EventContentFields.SELF_DESTRUCT_AFTER`. +const SELF_DESTRUCT_AFTER: &str = "org.matrix.self_destruct_after"; + +/// `_update_metadata_tables_txn` over `events`, the events left once the +/// rejected ones are stored. `input.events` is the whole chunk. +pub async fn update_metadata_tables( + txn: &WriteTxn, + input: &TxnInput<'_>, + events: &[Persisting<'_>], + effects: &mut PersistEffects, +) -> Result<(), WriteError> { + set_push_actions(txn, input, events).await?; + if events.is_empty() { + return Ok(()); + } + + for persisting in events { + let event = persisting.event(); + if event.type_() != event_type::M_ROOM_REDACTION { + continue; + } + let Some(redacted) = redacts(event) else { + continue; + }; + txn.execute( + "DELETE FROM event_push_actions WHERE room_id = $1 AND event_id = $2", + &[&event.room_id(), &redacted], + ) + .await?; + handle_redact_relations(txn, input, effects, redacted).await?; + } + + edges::handle_mult_prev_events(txn, input.instance_name, events).await?; + + store_search(txn, input, events).await?; + store_redactions(txn, input, events).await?; + for persisting in events { + store_retention_policy(txn, input, effects, persisting.event()).await?; + } + store_relations(txn, events).await?; + edges::store_state_dag_edges(txn, events).await?; + store_labels(txn, events).await?; + if input.enable_ephemeral_messages { + store_expiry(txn, events).await?; + } + + store_room_members(txn, input, events).await +} + +/// `_set_push_actions_for_event_and_users_txn`: move the push actions of +/// `events` from the staging table, and clear the staging rows of the whole +/// chunk, `input.events`. Only notifiable events have push actions. +async fn set_push_actions( + txn: &WriteTxn, + input: &TxnInput<'_>, + events: &[Persisting<'_>], +) -> Result<(), WriteError> { + let mut event_ids = Vec::new(); + let mut stream_orderings = Vec::new(); + let mut depths = Vec::new(); + for persisting in events { + let event = persisting.event(); + if event + .internal_metadata() + .is_notifiable() + .map_err(py_error)? + { + event_ids.push(event.event_id()); + stream_orderings.push(persisting.stream_ordering()); + depths.push(event.depth()); + } + } + if !event_ids.is_empty() { + txn.execute( + "INSERT INTO event_push_actions ( + room_id, event_id, user_id, actions, stream_ordering, + topological_ordering, notif, highlight, unread, thread_id + ) + SELECT $1, event_id, s.user_id, s.actions, e.stream_ordering, + e.depth, s.notif, s.highlight, s.unread, s.thread_id + FROM unnest($2::text[], $3::bigint[], $4::bigint[]) + AS e(event_id, stream_ordering, depth) + INNER JOIN event_push_actions_staging AS s USING (event_id)", + &[&input.room_id, &event_ids, &stream_orderings, &depths], + ) + .await?; + } + + let mut staged = Vec::new(); + for event in input.events { + if event + .event + .internal_metadata() + .is_notifiable() + .map_err(py_error)? + { + staged.push(event.event_id()); + } + } + if !staged.is_empty() { + txn.execute( + "DELETE FROM event_push_actions_staging WHERE event_id = ANY($1)", + &[&staged], + ) + .await?; + } + Ok(()) +} + +/// `_handle_redact_relations`: the redacted event no longer relates to +/// anything. Invalidate what was computed from its relation, and if it was +/// in a thread, find the thread's new latest event, or delete the thread. +async fn handle_redact_relations( + txn: &WriteTxn, + input: &TxnInput<'_>, + effects: &mut PersistEffects, + redacted_event_id: &str, +) -> Result<(), WriteError> { + let room_id = input.room_id; + let rows = txn + .query( + "SELECT relates_to_id, relation_type FROM event_relations WHERE event_id = $1", + &[&redacted_event_id], + ) + .await?; + let Some(row) = rows.first() else { + return Ok(()); + }; + let relates_to: String = row.try_get(0).map_err(anyhow::Error::new)?; + let rel_type: String = row.try_get(1).map_err(anyhow::Error::new)?; + txn.execute( + "DELETE FROM event_relations WHERE event_id = $1", + &[&redacted_event_id], + ) + .await?; + + // Any relation information for the related event must be cleared. + let mut invalidations = vec![( + "get_relations_for_event", + vec![room_id.to_owned(), relates_to.clone()], + )]; + match rel_type.as_str() { + relation_type::REFERENCE => { + invalidations.push(("get_references_for_event", vec![relates_to.clone()])); + } + relation_type::REPLACE => { + invalidations.push(("get_applicable_edit", vec![relates_to.clone()])); + } + relation_type::THREAD => { + invalidations.push(("get_thread_summary", vec![relates_to.clone()])); + invalidations.push(("get_thread_participated", vec![relates_to.clone()])); + invalidations.push(("get_threads", vec![room_id.to_owned()])); + } + _ => {} + } + for (cache_name, keys) in invalidations { + invalidate_cache_and_stream(txn, input, effects, cache_name, keys).await?; + } + if rel_type != relation_type::THREAD { + return Ok(()); + } + + let rows = txn + .query( + "SELECT event_id, topological_ordering, stream_ordering + FROM event_relations + INNER JOIN events USING (event_id) + WHERE relates_to_id = $1 AND relation_type = $2 + ORDER BY topological_ordering DESC, stream_ordering DESC + LIMIT 1", + &[&relates_to, &relation_type::THREAD], + ) + .await?; + if let Some(latest) = rows.first() { + // This may be the thread's current latest event, if an earlier + // event of the thread was redacted. + let event_id: String = latest.try_get(0).map_err(anyhow::Error::new)?; + let topological_ordering: i64 = latest.try_get(1).map_err(anyhow::Error::new)?; + let stream_ordering: i64 = latest.try_get(2).map_err(anyhow::Error::new)?; + txn.execute( + "INSERT INTO threads + (room_id, thread_id, latest_event_id, topological_ordering, stream_ordering) + VALUES ($1, $2, $3, $4, $5) + ON CONFLICT (room_id, thread_id) DO UPDATE SET + latest_event_id = EXCLUDED.latest_event_id, + topological_ordering = EXCLUDED.topological_ordering, + stream_ordering = EXCLUDED.stream_ordering", + &[ + &room_id, + &relates_to, + &event_id, + &topological_ordering, + &stream_ordering, + ], + ) + .await?; + } else { + // The thread no longer exists. As in Python's + // `simple_delete_one_txn`, by thread ID alone, and exactly one row. + let deleted = txn + .execute("DELETE FROM threads WHERE thread_id = $1", &[&relates_to]) + .await?; + if deleted != 1 { + return Err(anyhow!("deleting thread {relates_to}: {deleted} rows matched").into()); + } + } + Ok(()) +} + +/// The `event_search` rows of the events: room names, topics and message +/// bodies, as `_store_room_name_txn`, `_store_room_topic_txn` and +/// `_store_room_message_txn` add them with `store_search_entries_txn`. +async fn store_search( + txn: &WriteTxn, + input: &TxnInput<'_>, + events: &[Persisting<'_>], +) -> Result<(), WriteError> { + if !input.enable_search { + return Ok(()); + } + let mut event_ids = Vec::new(); + let mut room_ids = Vec::new(); + let mut keys = Vec::new(); + let mut values = Vec::new(); + let mut stream_orderings = Vec::new(); + let mut origin_server_tss = Vec::new(); + for persisting in events { + let event = persisting.event(); + let Some((key, value)) = search_entry(event) else { + continue; + }; + event_ids.push(event.event_id()); + room_ids.push(event.room_id()); + keys.push(key); + // `_clean_value_for_search`: Postgres takes no NUL in text. + values.push(value.replace('\0', " ")); + stream_orderings.push(persisting.stream_ordering()); + origin_server_tss.push(event.origin_server_ts()); + } + if event_ids.is_empty() { + return Ok(()); + } + txn.execute( + "INSERT INTO event_search + (event_id, room_id, key, vector, stream_ordering, origin_server_ts) + SELECT event_id, room_id, key, to_tsvector('english', value), + stream_ordering, origin_server_ts + FROM unnest($1::text[], $2::text[], $3::text[], $4::text[], $5::bigint[], $6::bigint[]) + AS t(event_id, room_id, key, value, stream_ordering, origin_server_ts)", + &[ + &event_ids, + &room_ids, + &keys, + &values, + &stream_orderings, + &origin_server_tss, + ], + ) + .await?; + Ok(()) +} + +/// The search key and text of an event, if it has one. +fn search_entry(event: &Event) -> Option<(&'static str, &str)> { + let content = event.content(); + match event.type_() { + NAME => Some(("content.name", content.get_field("name")?.as_str()?)), + TOPIC => { + content.get_field("topic")?.as_str()?; + Some(("content.topic", plain_text_topic(content).unwrap_or(""))) + } + MESSAGE => Some(("content.body", content.get_field("body")?.as_str()?)), + _ => None, + } +} + +/// `get_plain_text_topic_from_event_content`: the first plain text +/// representation in `m.topic`, else `topic`. A malformed `m.topic` is +/// ignored, as are its malformed representations. +fn plain_text_topic(content: &JsonObject) -> Option<&str> { + let topic = content.get_field("topic")?.as_str()?; + let representations = content + .get_field("m.topic") + .and_then(Value::as_object) + .and_then(|m_topic| match m_topic.get("m.text") { + Some(Value::Array(representations)) => Some(representations.as_slice()), + // Without `m.text` the `m.topic` is valid but has no text. + None => Some(&[][..]), + // Anything else makes the whole `m.topic` invalid. + Some(_) => None, + }) + .unwrap_or_default(); + for representation in representations { + let Some(representation) = representation.as_object() else { + continue; + }; + let Some(Value::String(body)) = representation.get("body") else { + continue; + }; + match representation.get("mimetype") { + // The mimetype defaults to `text/plain`, and Python takes an + // empty one as missing. + None | Some(Value::Null) => return Some(body), + Some(Value::String(mimetype)) if mimetype.is_empty() || mimetype == "text/plain" => { + return Some(body) + } + // A representation of another type, or a malformed one. + Some(_) => continue, + } + } + Some(topic) +} + +/// `_store_redaction`: a `redactions` row for each redaction. +async fn store_redactions( + txn: &WriteTxn, + input: &TxnInput<'_>, + events: &[Persisting<'_>], +) -> Result<(), WriteError> { + let mut event_ids = Vec::new(); + let mut redacted = Vec::new(); + let mut rechecks = Vec::new(); + for persisting in events { + let event = persisting.event(); + if event.type_() != event_type::M_ROOM_REDACTION { + continue; + } + let Some(redacts) = redacts(event) else { + continue; + }; + event_ids.push(event.event_id()); + redacted.push(redacts); + rechecks.push( + event + .internal_metadata() + .need_to_check_redaction() + .map_err(py_error)?, + ); + } + if event_ids.is_empty() { + return Ok(()); + } + let received_tss = vec![input.now_ms()?; event_ids.len()]; + txn.upsert_many( + "redactions", + &[Column::new("event_id", "text", &event_ids)], + &[ + Column::new("redacts", "text", &redacted), + Column::new("received_ts", "bigint", &received_tss), + Column::new("recheck", "boolean", &rechecks), + ], + ) + .await?; + Ok(()) +} + +/// `_store_retention_policy_for_room_txn`: a `room_retention` row for a +/// retention state event whose lifetimes are integers, if it has either. +async fn store_retention_policy( + txn: &WriteTxn, + input: &TxnInput<'_>, + effects: &mut PersistEffects, + event: &Event, +) -> Result<(), WriteError> { + if event.type_() != RETENTION || !event.is_state() { + return Ok(()); + } + let content = event.content(); + let min_lifetime = content.get_field("min_lifetime"); + let max_lifetime = content.get_field("max_lifetime"); + if min_lifetime.is_none() && max_lifetime.is_none() { + return Ok(()); + } + // Python ignores the event if either is present but not an integer. + let (Some(min_lifetime), Some(max_lifetime)) = ( + min_lifetime.map_or(Ok(Some(None)), |v| integer(v).map(|i| i.map(Some)))?, + max_lifetime.map_or(Ok(Some(None)), |v| integer(v).map(|i| i.map(Some)))?, + ) else { + return Ok(()); + }; + txn.execute( + "INSERT INTO room_retention (room_id, event_id, min_lifetime, max_lifetime) + VALUES ($1, $2, $3, $4)", + &[ + &event.room_id(), + &event.event_id(), + &min_lifetime, + &max_lifetime, + ], + ) + .await?; + invalidate_cache_and_stream( + txn, + input, + effects, + "get_retention_policy_for_room", + vec![event.room_id().to_owned()], + ) + .await +} + +/// A JSON integer, as Python's `type(value) is int` takes it, or `None`. +/// An integer too big for a `bigint` fails, as Python's insert does. +/// +/// serde reads an integer beyond `u64` as a float, which Python reads as an +/// integer. Python's insert then fails, while here it is not an integer. +fn integer(value: &Value) -> anyhow::Result> { + match value { + Value::Number(number) if !number.is_f64() => number + .as_i64() + .map(Some) + .ok_or_else(|| anyhow!("{number} is out of range for a bigint")), + _ => Ok(None), + } +} + +/// `_handle_event_relations`: an `event_relations` row for each event with +/// a relation, and for a thread reply, the thread's latest event, unless +/// the thread already has a later one. +async fn store_relations(txn: &WriteTxn, events: &[Persisting<'_>]) -> Result<(), WriteError> { + let mut event_ids = Vec::new(); + let mut parent_ids = Vec::new(); + let mut rel_types = Vec::new(); + let mut aggregation_keys = Vec::new(); + // The thread replies, as `(room_id, thread_id, event_id, depth, + // stream_ordering)`. + let mut replies: Vec<(&str, &str, &str, i64, i64)> = Vec::new(); + for persisting in events { + let event = persisting.event(); + let Some(relation) = relation(event) else { + continue; + }; + event_ids.push(event.event_id()); + parent_ids.push(relation.parent_id); + rel_types.push(relation.rel_type); + aggregation_keys.push(relation.aggregation_key); + if relation.rel_type == relation_type::THREAD { + replies.push(( + event.room_id(), + relation.parent_id, + event.event_id(), + event.depth(), + persisting.stream_ordering(), + )); + } + } + txn.insert_many( + "event_relations", + &[ + Column::new("event_id", "text", &event_ids), + Column::new("relates_to_id", "text", &parent_ids), + Column::new("relation_type", "text", &rel_types), + Column::new("aggregation_key", "text", &aggregation_keys), + ], + None, + ) + .await?; + + // Only a later event becomes the latest. The stream ordering is always + // later, unless the event was backfilled. In order, as each reply + // compares with the one before. + let thread_rows: Vec> = replies + .iter() + .map(|(room_id, thread_id, event_id, depth, stream_ordering)| { + vec![ + room_id as &(dyn ToSql + Sync), + thread_id, + event_id, + depth, + stream_ordering, + ] + }) + .collect(); + txn.execute_batch( + "INSERT INTO threads + (room_id, thread_id, latest_event_id, topological_ordering, stream_ordering) + VALUES ($1, $2, $3, $4, $5) + ON CONFLICT (room_id, thread_id) DO UPDATE SET + latest_event_id = EXCLUDED.latest_event_id, + topological_ordering = EXCLUDED.topological_ordering, + stream_ordering = EXCLUDED.stream_ordering + WHERE + threads.topological_ordering <= EXCLUDED.topological_ordering AND + threads.stream_ordering < EXCLUDED.stream_ordering", + &thread_rows, + ) + .await?; + Ok(()) +} + +/// `insert_labels_for_event_txn`: the event's labels, if it has a list of +/// them. +/// +/// Python iterates whatever the field holds and stores each element, so it +/// also stores labels from a string or an object, and non-string labels. +/// Only string labels from a list are stored here. +async fn store_labels(txn: &WriteTxn, events: &[Persisting<'_>]) -> Result<(), WriteError> { + let mut event_ids = Vec::new(); + let mut labels = Vec::new(); + let mut room_ids = Vec::new(); + let mut depths = Vec::new(); + for persisting in events { + let event = persisting.event(); + let Some(Value::Array(event_labels)) = event.content().get_field(LABELS) else { + continue; + }; + for label in event_labels.iter().filter_map(Value::as_str) { + event_ids.push(event.event_id()); + labels.push(label); + room_ids.push(event.room_id()); + depths.push(event.depth()); + } + } + txn.insert_many( + "event_labels", + &[ + Column::new("event_id", "text", &event_ids), + Column::new("label", "text", &labels), + Column::new("room_id", "text", &room_ids), + Column::new("topological_ordering", "bigint", &depths), + ], + None, + ) + .await?; + Ok(()) +} + +/// `_insert_event_expiry_txn`: the expiry of each non-state event that +/// sets an integer one. +async fn store_expiry(txn: &WriteTxn, events: &[Persisting<'_>]) -> Result<(), WriteError> { + let mut event_ids = Vec::new(); + let mut expiry_tss = Vec::new(); + for persisting in events { + let event = persisting.event(); + if event.is_state() { + continue; + } + let Some(expiry_ts) = event + .content() + .get_field(SELF_DESTRUCT_AFTER) + .map(integer) + .transpose()? + .flatten() + else { + continue; + }; + event_ids.push(event.event_id()); + expiry_tss.push(expiry_ts); + } + txn.insert_many( + "event_expiry", + &[ + Column::new("event_id", "text", &event_ids), + Column::new("expiry_ts", "bigint", &expiry_tss), + ], + None, + ) + .await?; + Ok(()) +} + +/// `event.membership`, if it is a string. +fn membership_of(event: &Event) -> Option<&str> { + event + .content() + .get_field("membership") + .and_then(Value::as_str) +} + +/// `non_null_str_or_none`: the value if it is a string without a NUL. +fn non_null_str(value: Option<&Value>) -> Option<&str> { + value.and_then(Value::as_str).filter(|s| !s.contains('\0')) +} + +/// `_store_room_members_txn`: a `room_memberships` row for each membership +/// event. An out-of-band membership of one of our users, such as a remote +/// invite or its rejection, is also current, so it updates +/// `local_current_membership` and the user's sliding sync snapshot, unless +/// the chunk was backfilled. +async fn store_room_members( + txn: &WriteTxn, + input: &TxnInput<'_>, + events: &[Persisting<'_>], +) -> Result<(), WriteError> { + let members: Vec<&Persisting<'_>> = events + .iter() + .filter(|p| p.event().type_() == event_type::M_ROOM_MEMBER) + .collect(); + if members.is_empty() { + return Ok(()); + } + let event_ids: Vec<&str> = members.iter().map(|p| p.event().event_id()).collect(); + let stream_orderings: Vec = members.iter().map(|p| p.stream_ordering()).collect(); + let user_ids: Vec> = members.iter().map(|p| p.event().state_key()).collect(); + let senders: Vec<&str> = members.iter().map(|p| p.event().sender()).collect(); + let room_ids: Vec<&str> = members.iter().map(|p| p.event().room_id()).collect(); + let memberships: Vec> = members.iter().map(|p| membership_of(p.event())).collect(); + let display_names: Vec> = members + .iter() + .map(|p| non_null_str(p.event().content().get_field("displayname"))) + .collect(); + let avatar_urls: Vec> = members + .iter() + .map(|p| non_null_str(p.event().content().get_field("avatar_url"))) + .collect(); + txn.insert_many( + "room_memberships", + &[ + Column::new("event_id", "text", &event_ids), + Column::new("event_stream_ordering", "bigint", &stream_orderings), + Column::new("user_id", "text", &user_ids), + Column::new("sender", "text", &senders), + Column::new("room_id", "text", &room_ids), + Column::new("membership", "text", &memberships), + Column::new("display_name", "text", &display_names), + Column::new("avatar_url", "text", &avatar_urls), + ], + None, + ) + .await?; + + if input.inhibit_local_membership_updates { + return Ok(()); + } + for persisting in members { + let event = persisting.event(); + let metadata = event.internal_metadata(); + let Some(user_id) = event.state_key() else { + continue; + }; + if !is_mine_id(user_id, input.server_name) + || !persisting.source.is_outlier()? + || !metadata.is_out_of_band_membership().map_err(py_error)? + { + continue; + } + let membership = membership_of(event).unwrap_or_default(); + let stripped_state = match membership { + "invite" => Some(event.unsigned_ref().invite_room_state().map_err(py_error)?), + "knock" => Some(event.unsigned_ref().knock_room_state().map_err(py_error)?), + // A rejected or retracted invite, or a rescinded knock, keeps the + // snapshot's state, so the room does not come and go from + // sliding sync filters. + "leave" => None, + _ => { + return Err(anyhow!( + "Unexpected out-of-band membership {membership} ({}) that we don't know how to handle yet", + event.event_id() + ) + .into()) + } + }; + + txn.execute( + "INSERT INTO local_current_membership + (room_id, user_id, event_id, event_stream_ordering, membership) + VALUES ($1, $2, $3, $4, $5) + ON CONFLICT (room_id, user_id) DO UPDATE SET + event_id = EXCLUDED.event_id, + event_stream_ordering = EXCLUDED.event_stream_ordering, + membership = EXCLUDED.membership", + &[ + &event.room_id(), + &user_id, + &event.event_id(), + &persisting.stream_ordering(), + &membership, + ], + ) + .await?; + + let snapshot_state = stripped_state + .map(|state| snapshot_values_from_stripped_state(state.as_deref())) + .transpose()?; + let room_id = event.room_id(); + let sender = event.sender(); + let event_id = event.event_id(); + // A new membership is not forgotten. + let forgotten = 0i32; + let stream_ordering = persisting.stream_ordering(); + let mut columns: Vec<(&'static str, &(dyn ToSql + Sync))> = vec![ + ("room_id", &room_id), + ("user_id", &user_id), + ("sender", &sender), + ("membership_event_id", &event_id), + ("membership", &membership), + ("forgotten", &forgotten), + ("event_stream_ordering", &stream_ordering), + ("event_instance_name", &input.instance_name), + ]; + if let Some(values) = &snapshot_state { + values.add_columns(&mut columns); + } + upsert_one( + txn, + "sliding_sync_membership_snapshots", + &["room_id", "user_id"], + &columns, + ) + .await?; + } + Ok(()) +} + +/// Insert a row, or update all but the `keys` columns of the row with the +/// same keys, as Python's `simple_upsert_txn`. +async fn upsert_one( + txn: &WriteTxn, + table: &str, + keys: &[&str], + columns: &[(&str, &(dyn ToSql + Sync))], +) -> Result<(), WriteError> { + let names: Vec<&str> = columns.iter().map(|(name, _)| *name).collect(); + let placeholders: Vec = (1..=columns.len()).map(|i| format!("${i}")).collect(); + let updates: Vec = names + .iter() + .filter(|name| !keys.contains(name)) + .map(|name| format!("{name} = EXCLUDED.{name}")) + .collect(); + let sql = format!( + "INSERT INTO {table} ({}) VALUES ({}) ON CONFLICT ({}) DO UPDATE SET {}", + names.join(", "), + placeholders.join(", "), + keys.join(", "), + updates.join(", ") + ); + let params: Vec<&(dyn ToSql + Sync)> = columns.iter().map(|(_, value)| *value).collect(); + txn.execute(&sql, ¶ms).await?; + Ok(()) +} + +/// What `_get_sliding_sync_insert_values_from_stripped_state` takes from an +/// invite's or knock's stripped state for its snapshot. A field is `None` +/// where Python leaves the column out. +#[derive(Debug, PartialEq, Eq)] +struct StrippedSnapshotValues { + has_known_state: bool, + room_type: Option>, + is_encrypted: Option, + room_name: Option>, + tombstone_successor_room_id: Option>, +} + +impl StrippedSnapshotValues { + fn add_columns<'a>(&'a self, columns: &mut Vec<(&'static str, &'a (dyn ToSql + Sync))>) { + columns.push(("has_known_state", &self.has_known_state)); + if let Some(room_type) = &self.room_type { + columns.push(("room_type", room_type)); + } + if let Some(is_encrypted) = &self.is_encrypted { + columns.push(("is_encrypted", is_encrypted)); + } + if let Some(room_name) = &self.room_name { + columns.push(("room_name", room_name)); + } + if let Some(successor) = &self.tombstone_successor_room_id { + columns.push(("tombstone_successor_room_id", successor)); + } + } +} + +/// `_get_sliding_sync_insert_values_from_stripped_state`. +fn snapshot_values_from_stripped_state( + stripped_state: Option<&[Value]>, +) -> anyhow::Result { + let no_state = StrippedSnapshotValues { + has_known_state: false, + room_type: Some(None), + is_encrypted: Some(false), + room_name: Some(None), + tombstone_successor_room_id: None, + }; + let Some(stripped_state) = stripped_state else { + return Ok(no_state); + }; + + // `StrippedStateEvent.from_json_dict`, keeping the last of each key. + let mut contents: std::collections::HashMap<(&str, &str), &Map> = + std::collections::HashMap::new(); + for raw in stripped_state { + let Some(raw) = raw.as_object() else { + continue; + }; + let field = |name: &str| raw.get(name).and_then(Value::as_str); + let (Some(event_type), Some(state_key), Some(_sender), Some(Value::Object(content))) = ( + field("type"), + field("state_key"), + field("sender"), + raw.get("content"), + ) else { + continue; + }; + contents.insert((event_type, state_key), content); + } + + // If there is any stripped state, the remote server sent all of it. + // There must at least be a create event. + let Some(create) = contents.get(&(CREATE, "")) else { + return Ok(no_state); + }; + let field = |event_type: &str, name: &str| { + contents + .get(&(event_type, "")) + .and_then(|content| content.get(name)) + .filter(|v| !v.is_null()) + }; + Ok(StrippedSnapshotValues { + has_known_state: true, + room_type: stripped_text(create.get("type").filter(|v| !v.is_null()))?, + is_encrypted: Some(field(ENCRYPTION, "algorithm").is_some()), + room_name: stripped_text(field(NAME, "name"))?, + tombstone_successor_room_id: stripped_text(field(TOMBSTONE, "replacement_room"))?, + }) +} + +/// A text column from stripped state: `Some(None)` for a missing value, and +/// `None`, leaving the column out, where Python's `"\0" in value` holds. That +/// is a string with a NUL, which Postgres does not allow, or a list holding +/// `"\0"`, or an object with that key. Python fails on any other value, in +/// the check or when storing it. +fn stripped_text(value: Option<&Value>) -> anyhow::Result>> { + match value { + None => Ok(Some(None)), + Some(Value::String(s)) if s.contains('\0') => Ok(None), + Some(Value::String(s)) => Ok(Some(Some(s.clone()))), + Some(Value::Array(items)) if items.iter().any(|item| item == "\0") => Ok(None), + Some(Value::Object(object)) if object.contains_key("\0") => Ok(None), + Some(other) => Err(anyhow!("stripped state value {other} is not a string")), + } +} + +fn py_error(e: pyo3::PyErr) -> WriteError { + WriteError::Other(anyhow::Error::new(e)) +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + + fn topic(content: Value) -> Option { + let content: JsonObject = serde_json::from_value(content).unwrap(); + plain_text_topic(&content).map(str::to_owned) + } + + #[test] + fn plain_text_topic_as_python() { + let plain = |body: &str| Some(body.to_owned()); + assert_eq!(topic(json!({"topic": "t"})), plain("t")); + assert_eq!( + topic(json!({"topic": "t", "m.topic": {"m.text": [ + {"mimetype": "text/html", "body": "h"}, + {"body": "p"}, + ]}})), + plain("p") + ); + assert_eq!( + topic(json!({"topic": "t", "m.topic": {"m.text": [ + {"mimetype": "", "body": "empty"}, + ]}})), + plain("empty") + ); + // Malformed representations are skipped. + assert_eq!( + topic(json!({"topic": "t", "m.topic": {"m.text": [ + "x", {"body": 1}, {"mimetype": 1, "body": "m"}, {"mimetype": null, "body": "n"}, + ]}})), + plain("n") + ); + // A malformed `m.topic` falls back to `topic`. + assert_eq!( + topic(json!({"topic": "t", "m.topic": {"m.text": {"body": "x"}}})), + plain("t") + ); + assert_eq!(topic(json!({"topic": "t", "m.topic": "x"})), plain("t")); + assert_eq!(topic(json!({"topic": "t", "m.topic": {}})), plain("t")); + assert_eq!(topic(json!({"topic": 1})), None); + } + + #[test] + fn stripped_state_as_python() { + let values = |state: Value| { + let state: Vec = serde_json::from_value(state).unwrap(); + snapshot_values_from_stripped_state(Some(&state)) + }; + let event = |event_type: &str, content: Value| json!({"type": event_type, "state_key": "", "sender": "@s:x", "content": content}); + + let no_state = StrippedSnapshotValues { + has_known_state: false, + room_type: Some(None), + is_encrypted: Some(false), + room_name: Some(None), + tombstone_successor_room_id: None, + }; + assert_eq!(snapshot_values_from_stripped_state(None).unwrap(), no_state); + // Without a create event, the state is unknown. + assert_eq!( + values(json!([event(NAME, json!({"name": "n"}))])).unwrap(), + no_state + ); + + assert_eq!( + values(json!([ + event(CREATE, json!({"type": "m.space"})), + event(ENCRYPTION, json!({"algorithm": "a"})), + event(NAME, json!({"name": "n"})), + event(TOMBSTONE, json!({"replacement_room": "!r:x"})), + // Malformed events are ignored. + {"type": NAME, "state_key": "", "content": {"name": "no sender"}}, + ])) + .unwrap(), + StrippedSnapshotValues { + has_known_state: true, + room_type: Some(Some("m.space".into())), + is_encrypted: Some(true), + room_name: Some(Some("n".into())), + tombstone_successor_room_id: Some(Some("!r:x".into())), + } + ); + + // NULs leave the column out, and a null algorithm is unencrypted. + assert_eq!( + values(json!([ + event(CREATE, json!({"type": "a\u{0}b"})), + event(ENCRYPTION, json!({"algorithm": null})), + event(NAME, json!({"name": "a\u{0}b"})), + ])) + .unwrap(), + StrippedSnapshotValues { + has_known_state: true, + room_type: None, + is_encrypted: Some(false), + room_name: None, + tombstone_successor_room_id: Some(None), + } + ); + + assert!(values(json!([event(CREATE, json!({"type": 1}))])).is_err()); + assert!(values(json!([event(CREATE, json!({"type": ["x"]}))])).is_err()); + // Python's `"\0" in value` also holds for a list or an object. + assert_eq!( + values(json!([ + event(CREATE, json!({"type": ["\u{0}"]})), + event(NAME, json!({"name": {"\u{0}": 1}})), + ])) + .unwrap(), + StrippedSnapshotValues { + has_known_state: true, + room_type: None, + is_encrypted: Some(false), + room_name: None, + tombstone_successor_room_id: Some(None), + } + ); + } + + #[test] + fn integers_as_python() { + assert_eq!(integer(&json!(5)).unwrap(), Some(5)); + assert_eq!(integer(&json!(-5)).unwrap(), Some(-5)); + assert_eq!(integer(&json!(5.0)).unwrap(), None); + assert_eq!(integer(&json!(true)).unwrap(), None); + assert_eq!(integer(&json!("5")).unwrap(), None); + // Too big for a `bigint`. + assert!(integer(&json!(u64::MAX)).is_err()); + } +} diff --git a/rust/src/persist/txn/mod.rs b/rust/src/persist/txn/mod.rs index a33c73b538..a03ec4f8e7 100644 --- a/rust/src/persist/txn/mod.rs +++ b/rust/src/persist/txn/mod.rs @@ -21,15 +21,19 @@ //! cache updates Python queues with `txn.call_after`. //! //! This is being ported a group of steps at a time, from the start of the -//! transaction. So far it covers the steps up to storing the rejections: -//! the events themselves ([`events`]) and the room's forward extremities -//! ([`edges`]). It returns the events left at that point. The Python steps after it run -//! only in `tests/storage/test_rust_persist_txn.py`, which persists the -//! rest of a chunk in Python and compares the rows with Python's. +//! transaction. So far it covers the steps up to `_update_metadata_tables_txn`: +//! the events themselves ([`events`]), the room's DAG ([`edges`]) and the +//! tables about the events ([`metadata`]), with the cache stream rows +//! ([`caches`]). It returns the events left at that point. The Python steps +//! after it run only in `tests/storage/test_rust_persist_txn.py`, which +//! persists the rest of a chunk in Python and compares the rows with +//! Python's. +pub mod caches; pub mod edges; pub mod events; pub mod json; +pub mod metadata; use std::collections::HashMap; @@ -37,6 +41,7 @@ use anyhow::anyhow; use pyo3::PyErr; use serde_json::Value; +use self::caches::CacheStream; use super::{ context::PersistContext, effects::{EventInvalidation, PersistEffects}, @@ -118,7 +123,17 @@ pub struct TxnInput<'a> { pub prepared: &'a PreparedChunk, /// This instance, which writes the events. pub instance_name: &'a str, + /// This server's name, for which users are ours. + pub server_name: &'a str, pub clock: &'a Clock, + /// The chunk was backfilled, so its memberships are not current. + pub inhibit_local_membership_updates: bool, + /// `server.enable_search`. + pub enable_search: bool, + /// `server.enable_ephemeral_messages`. + pub enable_ephemeral_messages: bool, + /// The cache stream IDs of this persist, over its attempts. + pub cache_stream: &'a CacheStream, } impl TxnInput<'_> { @@ -160,11 +175,28 @@ fn metadata_error(e: PyErr) -> WriteError { /// Python would queue with `txn.call_after`. Returns what the steps still /// to be ported take. /// -/// `effects` must start empty: a retried transaction starts again. +/// `effects` must start empty: a retried transaction starts again. Once +/// it commits, hand `effects` to `input.cache_stream`. pub async fn persist_events_txn<'a>( txn: &WriteTxn, input: &TxnInput<'a>, effects: &mut PersistEffects, +) -> Result, WriteError> { + // An earlier attempt whose commit failed left its cache stream IDs. + input.cache_stream.finish_pending(); + let result = persist_events_txn_steps(txn, input, effects).await; + if result.is_err() { + // The rows are rolled back, so their IDs are done with, as Python's + // `call_on_exception` has it. + input.cache_stream.finish_pending(); + } + result +} + +async fn persist_events_txn_steps<'a>( + txn: &WriteTxn, + input: &TxnInput<'a>, + effects: &mut PersistEffects, ) -> Result, WriteError> { let room_id = input.room_id; @@ -239,6 +271,8 @@ pub async fn persist_events_txn<'a>( // From here on, only events that were not rejected. let events = events::store_rejections(txn, input, events).await?; + metadata::update_metadata_tables(txn, input, &events, effects).await?; + Ok(TxnPrefix { events, de_outliered, diff --git a/synapse/storage/databases/main/__init__.py b/synapse/storage/databases/main/__init__.py index 65435ab473..2e20766bc5 100644 --- a/synapse/storage/databases/main/__init__.py +++ b/synapse/storage/databases/main/__init__.py @@ -222,6 +222,10 @@ class DataStore( for cache_name, keys in effects.cache_invalidations: self._attempt_to_invalidate_cache(cache_name, tuple(keys)) + if effects.cache_stream_ids: + assert self._cache_id_gen is not None + self._cache_id_gen._mark_ids_as_finished(effects.cache_stream_ids) + self.hs.get_replication_notifier().notify_replication() stream_change_caches: dict[str, StreamChangeCache] = { "events": self._events_stream_cache, diff --git a/synapse/synapse_rust/persist.pyi b/synapse/synapse_rust/persist.pyi index f3dac84778..33faa22700 100644 --- a/synapse/synapse_rust/persist.pyi +++ b/synapse/synapse_rust/persist.pyi @@ -56,10 +56,13 @@ class Persister: room_id: str, events_and_contexts: list[tuple[EventBase, EventContext]], prepared: PreparedChunk, + *, + inhibit_local_membership_updates: bool, ) -> tuple[PersistEffects, list[int], dict[int, str], list[int]]: """Run the part of the write transaction ported to Rust so far over the chunk, whose events have their stream orderings set, and commit - it. `prepared` is what Rust prepared for it. + it. `prepared` is what Rust prepared for it. The cache stream IDs it + wrote come back unfinished in the effects. Returns what it collected for Python to apply, and for the Python steps after it, by position in the chunk: the events left, the @@ -120,6 +123,7 @@ class PersistEffects: event_cache_invalidations: list[str] = ..., state_caches: list[tuple[str, list[str]]] = ..., cache_invalidations: list[tuple[str, list[str]]] = ..., + cache_stream_ids: list[int] = ..., state_group_prefills: list[tuple[str, int]] = ..., latest_event_ids: tuple[str, list[str]] | None = None, stream_changes: list[tuple[str, str, int]] = ..., @@ -150,6 +154,12 @@ class PersistEffects: """`(cache name, key)`: the local half of `_invalidate_cache_and_stream`.""" + @property + def cache_stream_ids(self) -> list[int]: + """The IDs of the cache stream rows the transaction wrote, still + unfinished. Python marks them finished once the caches are + invalidated.""" + @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 f523b9f9fa..a5520841e8 100644 --- a/tests/storage/test_rust_persist_prereqs.py +++ b/tests/storage/test_rust_persist_prereqs.py @@ -568,6 +568,54 @@ class PersistEffectsTestCase(unittest.HomeserverTestCase): self.assertEqual(seen_by_cache, [True]) self.assertEqual(id_gen.get_current_token(), stream) + def test_cache_stream_ids_finish_after_the_caches(self) -> None: + """As Python's `get_next_txn`: a cache stream ID is finished once the + cache is invalidated locally.""" + id_gen = self.store._cache_id_gen + if id_gen is None: + raise SkipTest("the cache stream needs Postgres") + allocation = id_gen.get_next() + stream_id = self.get_success(allocation.__aenter__()) + self.assertLess( + id_gen.get_current_token_for_writer(id_gen._instance_name), stream_id + ) + + calls: list[str] = [] + original_finish = id_gen._mark_ids_as_finished + original_invalidate = self.store._attempt_to_invalidate_cache + + def finish(ids: list[int]) -> None: + calls.append("finish") + original_finish(ids) + + def invalidate(cache_name: str, key: Collection[Any] | None) -> bool: + calls.append("invalidate") + return original_invalidate(cache_name, key) + + notifier = self.hs.get_replication_notifier() + original_notify = notifier.notify_replication + + def notify() -> None: + calls.append("notify") + original_notify() + + with ( + patch.object(id_gen, "_mark_ids_as_finished", finish), + patch.object(self.store, "_attempt_to_invalidate_cache", invalidate), + patch.object(notifier, "notify_replication", notify), + ): + self.store._apply_rust_persist_effects( + PersistEffects( + cache_invalidations=[("get_threads", [self.room_id])], + cache_stream_ids=[stream_id], + ) + ) + + self.assertEqual(calls, ["invalidate", "finish", "notify"]) + self.assertEqual( + id_gen.get_current_token_for_writer(id_gen._instance_name), stream_id + ) + class RustPersistConfigTestCase(unittest.TestCase): def _parse(self, level: Any) -> ExperimentalConfig: diff --git a/tests/storage/test_rust_persist_txn.py b/tests/storage/test_rust_persist_txn.py index d8882ff182..5a598d119a 100644 --- a/tests/storage/test_rust_persist_txn.py +++ b/tests/storage/test_rust_persist_txn.py @@ -60,6 +60,7 @@ from tests import unittest # The tables `_persist_events_txn` writes. Each is compared on the rows of # the room, by its `room_id` column if it has one, else by `event_id`. +# `event_edges` is also compared by `event_id`, see `_UNFILLED_ROOM_ID`. _TABLES = ( "current_state_delta_stream", "current_state_events", @@ -102,6 +103,10 @@ _TABLES = ( "timeline_gaps", ) +# Tables with a `room_id` column that is no longer filled in, compared by +# `event_id` instead. +_UNFILLED_ROOM_ID = {"event_edges"} + # Integer columns holding a position on a stream or a state group, by # column name or `(table, column)`, and which sequence the value is from. # They differ between the twins, so they are compared by their rank among @@ -245,6 +250,8 @@ class _TwinTestBase(unittest.HomeserverTestCase): experimental = config.setdefault("experimental_features", {}) experimental["rust_event_persistence"] = "prepare" experimental["msc4354_enabled"] = True + # For `event_expiry`. + config["enable_ephemeral_messages"] = True return config def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer: @@ -343,10 +350,19 @@ class _TwinTestBase(unittest.HomeserverTestCase): ) effects, left, redacted, de_outliered = ( self.persister.persist_events_txn_for_test( - room_id, events_and_contexts, prepared + room_id, + events_and_contexts, + prepared, + inhibit_local_membership_updates=kwargs[ + "inhibit_local_membership_updates" + ], ) ) twin.effects.extend(_rust_effects(effects)) + # As applying the effects would, so the cache stream moves on. + if effects.cache_stream_ids: + assert self.store._cache_id_gen is not None + self.store._cache_id_gen._mark_ids_as_finished(effects.cache_stream_ids) # The events left, as the steps after Rust's see them. left_events_and_contexts = [] for i in left: @@ -374,7 +390,8 @@ class _TwinTestBase(unittest.HomeserverTestCase): for callback, args, kwargs in (txn.after_callbacks or [])[start:]: name = callback.__name__ owner = getattr(callback, "__self__", None) - # A single-key cache's `prefill` is a lambda over the cache. + # A single-key cache's `prefill` and `invalidate` are lambdas + # over the cache. cache = next( ( cell.cell_contents @@ -385,18 +402,32 @@ class _TwinTestBase(unittest.HomeserverTestCase): ) if owner is self.store: owner_name = None + elif isinstance(owner, DeferredCache): + owner_name = self._cache_names().get(id(owner), "DeferredCache") elif isinstance(owner, MultiWriterIdGenerator): owner_name = owner._stream_name elif owner is not None: owner_name = getattr(owner, "name", None) or type(owner).__name__ elif name == "" and cache is not None: - # The transaction prefills only `_get_state_group_for_event`. - name, owner_name = "prefill", "_get_state_group_for_event" + # A single-key cache's `prefill` or `invalidate`, by the + # method the lambda calls on the cache. + code_names = callback.__code__.co_names + name = "prefill" if "prefill" in code_names else "invalidate" + owner_name = self._cache_names().get(id(cache), "DeferredCache") else: owner_name = callback.__qualname__ effects.append((name, owner_name, args, kwargs)) return effects + def _cache_names(self) -> dict[int, str]: + """The store's cached functions' names, by the ID of their cache.""" + names = {} + for name in dir(type(self.store)): + cache = getattr(getattr(self.store, name, None), "cache", None) + if isinstance(cache, DeferredCache): + names[id(cache)] = name + return names + def _cache_stream_position(self) -> int: rows = self.get_success( self.store.db_pool.execute( @@ -410,7 +441,7 @@ class _TwinTestBase(unittest.HomeserverTestCase): return self.get_success( self.store.db_pool.execute( "test", - """SELECT stream_id, instance_name, cache_func, keys + """SELECT stream_id, instance_name, cache_func, keys, invalidation_ts FROM cache_invalidation_stream_by_instance WHERE stream_id > ? ORDER BY stream_id""", after, @@ -431,10 +462,12 @@ class _TwinTestBase(unittest.HomeserverTestCase): sender: str | None = None, txn_id: str | None = None, sticky_ms: int | None = None, + push: bool = False, ) -> EventPersistencePair: """An event of the twin's, not yet persisted, added to its events. `prev` and `redacts` are indices of the twin's events. `content` can - be a function of the twin, for content that names its events.""" + be a function of the twin, for content that names its events. With + `push`, its push actions are staged, as sending it would.""" sender = sender or self.alice if callable(content): content = content(twin) @@ -464,6 +497,12 @@ class _TwinTestBase(unittest.HomeserverTestCase): ) ) context = self.get_success(unpersisted_context.persist(event)) + if push: + self.get_success( + self.hs.get_bulk_push_rule_evaluator().action_for_events_by_user( + [(event, context)] + ) + ) twin.events.append(event.event_id) return event, context @@ -519,7 +558,7 @@ class _TwinTestBase(unittest.HomeserverTestCase): raw: dict[str, tuple[list[str], list[tuple]]] = {} for table in _TABLES: columns = self._columns(table) - if "room_id" in columns: + if "room_id" in columns and table not in _UNFILLED_ROOM_ID: where = "room_id = ?" elif table == "event_auth_chain_links": where = """origin_chain_id IN ( @@ -556,8 +595,9 @@ class _TwinTestBase(unittest.HomeserverTestCase): key=repr, ) dump["cache_invalidation_stream_by_instance"] = [ - (instance_name, cache_func, normaliser.value("", "", keys)) - for _, instance_name, cache_func, keys in twin.cache_rows + # The twins share the clock, so the times compare as they are. + (instance_name, cache_func, normaliser.value("", "", keys), ts) + for _, instance_name, cache_func, keys, ts in twin.cache_rows ] dump["effects"] = sorted( ( @@ -700,6 +740,15 @@ def _rust_effects( ) for event_id in effects.event_cache_invalidations: calls.append(("_invalidate_local_get_event_cache", None, (event_id,), {})) + for cache_name, keys in effects.cache_invalidations: + calls.append(("invalidate", cache_name, (tuple(keys),), {})) + # Python's `get_next_txn` queues these per cache stream row. Applying + # the effects finishes the IDs at once, and notifies once. + for stream_id in effects.cache_stream_ids: + calls.append(("_mark_ids_as_finished", "caches", ([stream_id],), {})) + calls.append(("notify_replication", "ReplicationNotifier", (), {})) + if effects.new_replication_data: + calls.append(("on_new_replication_data", "Notifier", (), {})) for cache, entity, stream_id in effects.stream_changes: calls.append( ( @@ -759,12 +808,8 @@ def _python_steps_after_rust( txn, [event for event in de_outliered if event.sticky_duration()] ) - store._update_metadata_tables_txn( - txn, - events_and_contexts=events_and_contexts, - all_events_and_contexts=all_events_and_contexts, - inhibit_local_membership_updates=inhibit_local_membership_updates, - ) + # The end of `_update_metadata_tables_txn`, a Python cache prefill. + store._add_to_cache(txn, events_and_contexts) if state_delta_for_room: assert sliding_sync_table_changes is not None assert min_stream_order is not None @@ -841,7 +886,17 @@ class PersistTxnTestCase(_TwinTestBase): self._step(target) target_index = len(self.twins[0].events) - 1 - self._send(EventTypes.Redaction, {}, redacts=target_index, prev=[base]) + + def redaction(twin: _Twin) -> list[EventPersistencePair]: + event, context = self._build( + twin, EventTypes.Redaction, {}, redacts=target_index, prev=[base] + ) + # Without the target, the recheck cannot pass yet, so it is + # stored as still needed. + event.internal_metadata.recheck_redaction = True + return [(event, context)] + + self._step(redaction) self._step(lambda twin: [built[twin.name]]) self._assert_twins_match() @@ -1044,6 +1099,304 @@ class PersistTxnTestCase(_TwinTestBase): ) self._assert_twins_match() + def _join_bob(self) -> None: + self._send(EventTypes.Member, {"membership": "invite"}, state_key=self.bob) + self._send( + EventTypes.Member, + {"membership": "join", "displayname": "Bob", "avatar_url": "mxc://a/b"}, + state_key=self.bob, + sender=self.bob, + ) + + def test_relations_and_their_redaction(self) -> None: + """Each kind of relation, then redactions of them that clear what + was computed from them. Redacting the thread's latest reply moves + the thread back, and redacting its last reply deletes it.""" + root = self._message("root") + + def relates(rel_type: str, **extra: Any) -> int: + return self._send( + EventTypes.Message, + lambda twin: { + "msgtype": "m.text", + "body": rel_type, + "m.relates_to": { + "rel_type": rel_type, + "event_id": twin.events[root], + **extra, + }, + }, + ) + + edit = relates("m.replace") + reference = relates("m.reference") + annotation = relates("m.annotation", key="\U0001f44d") + first_reply = relates("m.thread") + second_reply = relates("m.thread") + for target in (edit, reference, annotation, second_reply, first_reply): + self._send(EventTypes.Redaction, {}, redacts=target) + self._assert_twins_match() + + def test_relation_and_redaction_in_one_chunk(self) -> None: + """A redaction runs its relation cleanup before the chunk's + relations are stored, so the relation it redacts in the same chunk + is kept.""" + root = self._message("root") + + def chunk(twin: _Twin) -> list[EventPersistencePair]: + reply = self._build( + twin, + EventTypes.Message, + { + "msgtype": "m.text", + "body": "reply", + "m.relates_to": { + "rel_type": "m.thread", + "event_id": twin.events[root], + }, + }, + ) + redaction = self._build( + twin, + EventTypes.Redaction, + {}, + redacts=len(twin.events) - 1, + prev=[root], + ) + return [reply, redaction] + + self._step(chunk) + self._assert_twins_match() + + def test_thread_replies_in_one_chunk(self) -> None: + """Each reply compares with the one before, so the thread's latest + event is the later reply.""" + root = self._message("root") + + def replies(twin: _Twin) -> list[EventPersistencePair]: + return [ + self._build( + twin, + EventTypes.Message, + { + "msgtype": "m.text", + "body": body, + "m.relates_to": { + "rel_type": "m.thread", + "event_id": twin.events[root], + }, + }, + prev=[root], + ) + for body in ("first", "second") + ] + + self._step(replies) + self._assert_twins_match() + + def test_retention_in_one_chunk(self) -> None: + """Two policies, whose cache stream rows keep the events' order.""" + + def policies(twin: _Twin) -> list[EventPersistencePair]: + # Different state keys, so state resolution has nothing to break + # a tie on. + return [ + self._build( + twin, + EventTypes.Retention, + {"max_lifetime": lifetime}, + state_key=state_key, + prev=[0], + ) + for state_key, lifetime in (("", 1000), ("other", 2000)) + ] + + self._message("base") + self._step(policies) + self._assert_twins_match() + + def test_search(self) -> None: + self._send(EventTypes.Name, {"name": "Name"}, state_key="") + self._send( + EventTypes.Topic, + { + "topic": "plain", + "m.topic": { + "m.text": [ + {"mimetype": "text/html", "body": "rich"}, + {"body": "from m.topic"}, + ] + }, + }, + state_key="", + ) + self._message("searchable words") + self._assert_twins_match() + + def test_retention(self) -> None: + self._send( + EventTypes.Retention, + {"min_lifetime": 1000, "max_lifetime": 86_400_000}, + state_key="", + ) + # No lifetimes, so no policy. + self._send(EventTypes.Retention, {}, state_key="") + self._assert_twins_match() + + def test_labels_and_expiry(self) -> None: + self._send( + EventTypes.Message, + { + "msgtype": "m.text", + "body": "labelled", + "org.matrix.labels": ["#fun", "#work"], + "org.matrix.self_destruct_after": 2_000_000_000_000, + }, + ) + # Not an integer, so no expiry. + self._send( + EventTypes.Message, + { + "msgtype": "m.text", + "body": "x", + "org.matrix.self_destruct_after": "1", + }, + ) + self._assert_twins_match() + + def test_members(self) -> None: + self._join_bob() + self._send( + EventTypes.Member, + # A NUL is not allowed in the database, so it is not stored. + {"membership": "join", "displayname": "B\u0000b", "avatar_url": 1}, + state_key=self.bob, + sender=self.bob, + ) + self._assert_twins_match() + + def test_push_actions(self) -> None: + """Push actions move from the staging table, and a redaction + removes them.""" + self._join_bob() + target = self._send( + EventTypes.Message, + {"msgtype": "m.text", "body": "hello bob"}, + push=True, + ) + self._assert_twins_match() + for twin in self.twins: + # Bob is notified. + self.assertTrue( + self.get_success( + self.store.db_pool.simple_select_onecol( + "event_push_actions", + {"room_id": twin.room_id, "user_id": self.bob}, + "event_id", + ) + ) + ) + self._send(EventTypes.Redaction, {}, redacts=target, push=True) + self._assert_twins_match() + rows = self.get_success( + self.store.db_pool.simple_select_onecol( + "event_push_actions", + {"room_id": self.twins[1].room_id}, + "event_id", + ) + ) + # The redaction itself does not notify. + self.assertEqual(rows, []) + self.assertEqual( + self.get_success( + self.store.db_pool.simple_select_onecol( + "event_push_actions_staging", {}, "event_id" + ) + ), + [], + ) + + def _skip_hand_built_in_state_dag_rooms(self) -> None: + if KNOWN_ROOM_VERSIONS[self.room_version].msc4242_state_dags: + raise SkipTest("a hand-built event has no state DAG context") + + def _out_of_band( + self, + twin: _Twin, + membership: str, + stripped_state: list[dict[str, Any]] | None, + ) -> EventPersistencePair: + """An out-of-band membership of Bob's, as a remote invite or knock, + or its rejection, arrives over federation: an outlier.""" + event, _ = self._build( + twin, EventTypes.Member, {"membership": "invite"}, state_key=self.bob + ) + pdu = event.get_pdu_json() + pdu["content"] = {"membership": membership} + if membership != "invite": + pdu["sender"] = self.bob + pdu["unsigned"] = {} + if stripped_state is not None: + pdu["unsigned"][f"{membership}_room_state"] = stripped_state + copy = make_event_from_dict(pdu, event.room_version, {}) + copy.internal_metadata.outlier = True + copy.internal_metadata.out_of_band_membership = True + copy.internal_metadata.stream_ordering = None + twin.events[-1] = copy.event_id + return copy, EventContext.for_outlier(self.hs.get_storage_controllers()) + + def test_out_of_band_invite_and_rejection(self) -> None: + self._skip_hand_built_in_state_dag_rooms() + + def stripped(event_type: str, content: dict[str, Any]) -> dict[str, Any]: + return { + "type": event_type, + "state_key": "", + "sender": "@remote:other", + "content": content, + } + + state = [ + stripped(EventTypes.Create, {"type": "m.space"}), + stripped(EventTypes.RoomEncryption, {"algorithm": "m.megolm.v1"}), + stripped(EventTypes.Name, {"name": "Remote"}), + # Malformed, so ignored. + {"type": EventTypes.Tombstone, "content": {}}, + ] + self._step(lambda twin: [self._out_of_band(twin, "invite", state)]) + self._step(lambda twin: [self._out_of_band(twin, "leave", None)]) + self._assert_twins_match() + + def test_out_of_band_knock(self) -> None: + self._skip_hand_built_in_state_dag_rooms() + state = [ + { + "type": EventTypes.Create, + "state_key": "", + "sender": "@remote:other", + "content": {}, + }, + { + "type": EventTypes.Name, + "state_key": "", + "sender": "@remote:other", + # A NUL leaves the column out. + "content": {"name": "a\u0000b"}, + }, + ] + self._step(lambda twin: [self._out_of_band(twin, "knock", state)]) + # No stripped state. + self._step(lambda twin: [self._out_of_band(twin, "invite", None)]) + self._assert_twins_match() + + def test_out_of_band_backfilled(self) -> None: + """A backfilled membership is not current.""" + self._skip_hand_built_in_state_dag_rooms() + self._step( + lambda twin: [self._out_of_band(twin, "invite", [])], backfilled=True + ) + self._assert_twins_match() + def test_mismatch_is_found(self) -> None: """The comparison sees a row that differs.""" self._message("hello") @@ -1061,7 +1414,10 @@ class PersistTxnTestCase(_TwinTestBase): def test_missing_room(self) -> None: with self.assertRaisesRegex(RuntimeError, "Room does not exist !nope:test"): self.persister.persist_events_txn_for_test( - "!nope:test", [], self.persister.last_prepared + "!nope:test", + [], + self.persister.last_prepared, + inhibit_local_membership_updates=False, )