From 41c7808711fc4ebf09aacc03c0dedb981f03cfa8 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Sat, 17 Oct 2026 14:09:44 +0000 Subject: [PATCH] Store the event metadata in the Rust persist transaction The third commit of phase 2. The Rust transaction now also runs `_update_metadata_tables_txn`, apart from the event cache prefill at its end (`_add_to_cache`), which stays in Python: - `txn/metadata.rs`: push actions move from the staging table, and the chunk's staging rows go. A redaction removes the push actions and the relation of the event it redacts, and moves or deletes its thread. Then search (with the plain text topic from `m.topic`), redactions, retention, relations and threads, labels, expiry, and room members. An out-of-band membership of a local user also updates `local_current_membership` and the user's sliding sync snapshot, with the stripped state of an invite or knock. - `txn/edges.rs`: `event_edges` with the backward extremities, and the MSC4242 state DAG edges. - `txn/caches.rs`: `_invalidate_cache_and_stream`. The rows go into `cache_invalidation_stream_by_instance` inside the transaction, with IDs from the cache stream's tracker. The committed attempt hands its IDs to Python in `PersistEffects.cache_stream_ids`. Python finishes them after invalidating the caches, as `get_next_txn` does. A failed attempt's IDs are finished when its steps fail, or for a failed commit, by the next attempt or when the persist ends. A fetch of an ID that is dropped before the sequence answers still ends, so a cancelled persist cannot hold the stream's position back. Where Python and Rust differ, the input is malformed: - Python stores labels from any value it can iterate, and non-string labels. Rust stores only string labels from a list. - An integer lifetime or expiry beyond `u64` is a float to serde, so Rust ignores it where Python's insert fails. Within `u64` but beyond `bigint`, both fail. The persister reads `enable_search` and `enable_ephemeral_messages` from the server config. `persist_events_txn_for_test` takes `inhibit_local_membership_updates`. New twin scenarios: each kind of relation and its redaction, which moves the thread back and then deletes it, a relation redacted in its own chunk, two thread replies in one chunk, search, retention (also two policies in one chunk), labels and expiry, member rows, push actions and their redaction, an out-of-band invite and its rejection, a knock, and a backfilled invite. The out-of-band ones are skipped in state DAG rooms, as a hand-built event has no state DAG context. A redaction before its target is now stored with a recheck still needed. The test harness compared `event_edges` by its `room_id` column, which is no longer filled in, so it compared no rows. It now compares by `event_id`. It also names the cache behind each invalidation, compares the cache stream rows' times, and finishes the cache stream IDs Rust hands back. Fifteen mutants of the new steps each fail the comparison. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LXMkcZVkcZhi21augBff4a --- rust/src/config/mod.rs | 2 + rust/src/events/constants.rs | 12 + rust/src/events/internal_metadata.rs | 2 +- rust/src/events/unsigned.rs | 12 + rust/src/persist/effects.rs | 21 + rust/src/persist/python.rs | 24 +- rust/src/persist/sliding_sync.rs | 4 +- rust/src/persist/txn/caches.rs | 215 +++++ rust/src/persist/txn/edges.rs | 96 +- rust/src/persist/txn/metadata.rs | 1011 ++++++++++++++++++++ rust/src/persist/txn/mod.rs | 46 +- synapse/storage/databases/main/__init__.py | 4 + synapse/synapse_rust/persist.pyi | 12 +- tests/storage/test_rust_persist_prereqs.py | 48 + tests/storage/test_rust_persist_txn.py | 390 +++++++- 15 files changed, 1866 insertions(+), 33 deletions(-) create mode 100644 rust/src/persist/txn/caches.rs create mode 100644 rust/src/persist/txn/metadata.rs 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, )