diff --git a/rust/src/events/internal_metadata.rs b/rust/src/events/internal_metadata.rs index c57a9fa7bb..698caacc1a 100644 --- a/rust/src/events/internal_metadata.rs +++ b/rust/src/events/internal_metadata.rs @@ -606,19 +606,15 @@ impl EventInternalMetadata { Ok(()) } - /// The `internal_metadata` column, as `get_dict()` returns it, built - /// without the GIL. - pub fn to_persistence_value(&self) -> PyResult { - Ok(serde_json::Value::Object( - self.read_inner()? - .data - .iter() - .map(|entry| { - let (key, value) = entry.to_json_pair(); - (key.to_owned(), value) - }) - .collect(), - )) + /// The `internal_metadata` column as key-value pairs, in the order + /// `get_dict()` gives them, built without the GIL. + pub fn to_persistence_pairs(&self) -> PyResult> { + Ok(self + .read_inner()? + .data + .iter() + .map(|entry| entry.to_json_pair()) + .collect()) } /// The event's stream ordering, if known. diff --git a/rust/src/events/mod.rs b/rust/src/events/mod.rs index 13bcde147b..1488e789c4 100644 --- a/rust/src/events/mod.rs +++ b/rust/src/events/mod.rs @@ -52,22 +52,23 @@ use std::{cmp::min, sync::Arc}; use anyhow::Error; use pyo3::{ - exceptions::{PyAttributeError, PyKeyError, PyRuntimeError, PyValueError}, + exceptions::{PyAttributeError, PyKeyError, PyValueError}, pyclass, pyfunction, pymethods, types::{PyAnyMethods, PyDict, PyDictMethods, PyList, PyMapping, PyModule, PyModuleMethods}, wrap_pyfunction, Bound, IntoPyObject, PyAny, PyResult, Python, }; use pythonize::{depythonize, pythonize}; +use serde::Serialize; use serde_json::Value; use crate::events::{ constants::event_type::M_ROOM_MEMBER, formats::{ - EventFormatEnum, EventFormatV1, EventFormatV2V3, EventFormatV4, EventFormatVMSC4242, - FormattedEvent, + EventCommonFields, EventFormatEnum, EventFormatV1, EventFormatV2V3, EventFormatV4, + EventFormatVMSC4242, FormattedEvent, }, signatures::Signatures, - unsigned::Unsigned, + unsigned::{PersistedUnsignedFields, Unsigned}, utils::redact, }; use crate::{ @@ -765,24 +766,27 @@ impl Event { serde_json::to_value(&self.parsed_event) } - /// The `event_json.json` column: the event as - /// `get_dict_for_persistence()` returns it, with `unsigned` holding only - /// the fields kept in the database, built without the GIL. + /// Calls `f` with the `event_json.json` column: the event as + /// `get_dict_for_persistence()` returns it, with its keys in the same + /// order and `unsigned` holding only the fields kept in the database, + /// built without the GIL. /// /// The persist transaction drops `redacted` and `redacted_because` from /// that dict. Neither is ever in the Rust event, so there is nothing to /// drop. - pub fn to_persistence_value(&self) -> PyResult { - let mut value = serde_json::to_value(&self.parsed_event) - .map_err(|e| PyRuntimeError::new_err(format!("serializing event: {e}")))?; - let object = value - .as_object_mut() - .ok_or_else(|| PyRuntimeError::new_err("event did not serialize to an object"))?; - object.insert( - UNSIGNED.to_owned(), - self.parsed_event.unsigned.persistence_value()?, - ); - Ok(value) + pub fn with_persistence_dict( + &self, + f: impl FnOnce(&PersistenceDict<'_>) -> R, + ) -> PyResult { + let event = &self.parsed_event; + event.unsigned.with_persisted_fields(|unsigned| { + f(&PersistenceDict { + signatures: &event.signatures, + unsigned, + specific_fields: &event.specific_fields, + common_fields: &event.common_fields, + }) + }) } /// Returns the `redacts` field of this event, if it has one. @@ -797,6 +801,19 @@ impl Event { } } +/// An event as [`Event::with_persistence_dict`] serializes it. The fields +/// are in [`FormattedEvent`]'s order, which is the order `get_dict()` gives +/// them. +#[derive(Serialize)] +pub struct PersistenceDict<'a> { + signatures: &'a Signatures, + unsigned: &'a PersistedUnsignedFields, + #[serde(flatten)] + specific_fields: &'a EventFormatEnum, + #[serde(flatten)] + common_fields: &'a EventCommonFields, +} + /// Parses a JSON string into a [`FormattedEvent`] for the given room version. pub(crate) fn event_dict_from_json_str( room_version: &RoomVersion, diff --git a/rust/src/events/unsigned.rs b/rust/src/events/unsigned.rs index 0cf5c316ef..c2e7cd3773 100644 --- a/rust/src/events/unsigned.rs +++ b/rust/src/events/unsigned.rs @@ -35,7 +35,7 @@ pub struct Unsigned { /// The fields in the unsigned data of an event that are persisted in the /// database. #[derive(Debug, Clone, Serialize, Deserialize, Default)] -struct PersistedUnsignedFields { +pub struct PersistedUnsignedFields { #[serde(skip_serializing_if = "Option::is_none")] age_ts: Option, #[serde(skip_serializing_if = "Option::is_none")] @@ -292,11 +292,13 @@ impl Unsigned { } impl Unsigned { - /// The fields stored in the database, as [`Unsigned::for_persistence`] - /// returns them, built without the GIL. - pub fn persistence_value(&self) -> PyResult { - serde_json::to_value(&self.py_read()?.persisted_fields) - .map_err(|e| PyRuntimeError::new_err(format!("serializing unsigned: {e}"))) + /// Calls `f` with the fields stored in the database, as + /// [`Unsigned::for_persistence`] returns them, without the GIL. + pub fn with_persisted_fields( + &self, + f: impl FnOnce(&PersistedUnsignedFields) -> R, + ) -> PyResult { + Ok(f(&self.py_read()?.persisted_fields)) } } diff --git a/rust/src/persist/python.rs b/rust/src/persist/python.rs index e7d7e63cab..5ea6f1083f 100644 --- a/rust/src/persist/python.rs +++ b/rust/src/persist/python.rs @@ -21,6 +21,7 @@ use std::{ sync::{Arc, Mutex}, }; +use anyhow::Context; use futures::{future::BoxFuture, FutureExt}; use http::StatusCode; use pyo3::{exceptions::PyRuntimeError, intern, prelude::*, types::PyDict}; @@ -33,10 +34,12 @@ use super::{ prepare::{self, PrepareInput, PrepareOptions, Prepared}, reader::{self, PersistReader, PyDelta, PySlidingSync}, state_delta::NativeStateResolver, - txn::{self, TxnEvent, TxnInput}, + txn::{self, json, PartialStateConflict, TxnEvent, TxnInput}, }; use crate::{ - database::postgres::write::{is_retryable_error, run_write_interaction, MAX_ATTEMPTS}, + database::postgres::write::{ + is_retryable_error, run_write_interaction, WriteError, MAX_ATTEMPTS, + }, deferred::{create_deferred_with, OnCancel}, errors::SynapseError, event_auth::AuthFailure, @@ -200,7 +203,7 @@ impl Persister { { Ok(prepared) => { return Ok(PreparedChunk { - prepared, + prepared: Arc::new(prepared), counts: Mutex::new(Some(counts)), }) } @@ -270,41 +273,79 @@ impl Persister { /// Run the part of the write transaction ported so far over the chunk, /// whose events must have their stream orderings set, and commit it. - /// Returns what it collected for Python to apply and the positions in - /// the chunk of the events left for the Python steps after it. For + /// `prepared` is what Rust prepared for it. For /// `tests/storage/test_rust_persist_txn.py`. + /// + /// 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)] fn persist_events_txn_for_test( &self, py: Python<'_>, room_id: String, events_and_contexts: Vec<(Py, PersistContext)>, - ) -> PyResult<(PersistEffects, Vec)> { + prepared: &PreparedChunk, + ) -> PyResult<( + PersistEffects, + Vec, + HashMap, + Vec, + )> { + let prepared = Arc::clone(&prepared.prepared); + if !matches!(&*prepared, Prepared::Chunk(_)) { + return Err(PyRuntimeError::new_err("the chunk was prepared in Python")); + } let events = events_and_contexts .into_iter() .map(|(event, context)| TxnEvent::new(event.bind(py).get().shallow_clone(), context)) .collect::>>() .map_err(persist_error_to_py)?; let pool = self.inner.pool.get()?; + let inner = Arc::clone(&self.inner); let future = async move { + let Prepared::Chunk(chunk) = &*prepared else { + unreachable!("checked above"); + }; let input = TxnInput { room_id: &room_id, events: &events, + prepared: chunk, + instance_name: &inner.instance_name, + clock: inner.runtime.clock(), + }; + let position = |event: &TxnEvent| { + input + .events + .iter() + .position(|x| std::ptr::eq(x, event)) + .expect("an event of the chunk") }; run_write_interaction(&pool, "persist_events", |txn, _| { - let input = &input; + let (input, position) = (&input, &position); async move { let mut effects = PersistEffects::default(); - let left = txn::persist_events_txn(txn, input, &mut effects).await?; - let left = left - .iter() - .filter_map(|e| input.events.iter().position(|x| std::ptr::eq(x, *e))) - .collect(); - Ok((effects, left)) + let prefix = txn::persist_events_txn(txn, input, &mut effects).await?; + let left = prefix.events.iter().map(|e| position(e.source)).collect(); + let mut redacted = HashMap::new(); + for event in &prefix.events { + if let Some(copy) = &event.redacted { + let by = copy + .internal_metadata() + .redacted_by() + .map_err(anyhow::Error::new)? + .context("a redacted copy without redacted_by")?; + redacted.insert(position(event.source), by); + } + } + let de_outliered = prefix.de_outliered.iter().map(|e| position(e)).collect(); + Ok((effects, left, redacted, de_outliered)) } .boxed() }) .await - .map_err(|e| persist_error_to_py(anyhow::Error::new(e))) + .map_err(write_error_to_py) }; self.block_on(py, future.boxed()) } @@ -330,14 +371,15 @@ impl Persister { /// and Python should compute the chunk itself. #[pyclass(frozen)] pub struct PreparedChunk { - prepared: Prepared, + /// Shared with the transaction that writes the chunk. + prepared: Arc, /// What preparing counted, until it is recorded. counts: Mutex>, } impl PreparedChunk { fn chunk(&self) -> Option<&prepare::PreparedChunk> { - match &self.prepared { + match &*self.prepared { Prepared::Chunk(chunk) => Some(chunk), Prepared::StateResFallback(_) => None, } @@ -349,7 +391,7 @@ impl PreparedChunk { /// Why the Rust state resolution could not resolve the room. #[getter] fn state_res_fallback(&self) -> Option<&str> { - match &self.prepared { + match &*self.prepared { Prepared::Chunk(_) => None, Prepared::StateResFallback(reason) => Some(reason), } @@ -422,6 +464,15 @@ impl PreparedChunk { } } +/// Convert an error from a persist transaction to the Python exception the +/// Python transaction would raise. See [`persist_error_to_py`]. +pub fn write_error_to_py(e: WriteError) -> PyErr { + match e { + WriteError::Other(e) => persist_error_to_py(e), + e => persist_error_to_py(anyhow::Error::new(e)), + } +} + /// Convert an error from persisting to the Python exception the Python /// path would raise: a Python exception as it was raised, an auth /// failure as the auth rules raise it (from any step, state resolution @@ -442,6 +493,11 @@ pub fn persist_error_to_py(e: anyhow::Error) -> PyErr { Ok(failure) => return failure.into_py_err(), Err(e) => e, }; + if e.downcast_ref::().is_some() { + return match py_partial_state_conflict_error() { + Ok(err) | Err(err) => err, + }; + } if let Some(missing) = e.downcast_ref::() { return SynapseError::new( StatusCode::INTERNAL_SERVER_ERROR, @@ -454,14 +510,26 @@ pub fn persist_error_to_py(e: anyhow::Error) -> PyErr { PyRuntimeError::new_err(format!("{e:#}")) } +/// A `PartialStateConflictError`, as the Python transaction raises. +fn py_partial_state_conflict_error() -> PyResult { + crate::gil::with_gil("persist_partial_state_conflict", |py| { + let error = py + .import("synapse.api.errors")? + .getattr("PartialStateConflictError")? + .call0()?; + Ok(PyErr::from_value(error)) + }) +} + /// `(internal_metadata, json)`: the two `event_json` columns as Rust would /// write them for `event`, for tests to compare with Python's. #[pyfunction] pub fn persistence_json_for_test(event: &Bound<'_, Event>) -> PyResult<(String, String)> { let event = event.get(); - let internal_metadata = event.internal_metadata().to_persistence_value()?; - let json = event.to_persistence_value()?; - Ok((internal_metadata.to_string(), json.to_string())) + let internal_metadata = json::encode_object(&event.internal_metadata().to_persistence_pairs()?); + let json = event.with_persistence_dict(|dict| json::encode(dict))?; + let to_py = |e: serde_json::Error| PyRuntimeError::new_err(format!("encoding: {e}")); + Ok((internal_metadata.map_err(to_py)?, json.map_err(to_py)?)) } /// The [`PersistContext`] of `context` as a dict, for tests to check the diff --git a/rust/src/persist/txn/edges.rs b/rust/src/persist/txn/edges.rs new file mode 100644 index 0000000000..15dc2d4d85 --- /dev/null +++ b/rust/src/persist/txn/edges.rs @@ -0,0 +1,195 @@ +/* + * 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 room's extremities: `_update_forward_extremities_txn`, +//! `_set_state_dag_extremities_txn` and `_update_backward_extremeties`. + +use std::{collections::HashSet, sync::Arc}; + +use crate::{ + database::postgres::write::{Column, WriteError, WriteTxn}, + events::Event, +}; + +/// `_update_forward_extremities_txn`: replace the room's forward +/// extremities, and record them at `max_stream_order`. +pub async fn update_forward_extremities( + txn: &WriteTxn, + room_id: &str, + new_forward_extremities: &HashSet>, + max_stream_order: i64, +) -> Result<(), WriteError> { + txn.execute( + "DELETE FROM event_forward_extremities WHERE room_id = $1", + &[&room_id], + ) + .await?; + + let event_ids: Vec<&str> = new_forward_extremities.iter().map(|id| &**id).collect(); + let room_ids = vec![room_id; event_ids.len()]; + txn.insert_many( + "event_forward_extremities", + &[ + Column::new("event_id", "text", &event_ids), + Column::new("room_id", "text", &room_ids), + ], + None, + ) + .await?; + + let stream_orderings = vec![max_stream_order; event_ids.len()]; + txn.insert_many( + "stream_ordering_to_exterm", + &[ + Column::new("room_id", "text", &room_ids), + Column::new("event_id", "text", &event_ids), + Column::new("stream_ordering", "bigint", &stream_orderings), + ], + None, + ) + .await?; + Ok(()) +} + +/// `_set_state_dag_extremities_txn`: replace the room's MSC4242 state DAG +/// forward extremities. +pub async fn set_state_dag_extremities( + txn: &WriteTxn, + room_id: &str, + new_extremities: &HashSet>, +) -> Result<(), WriteError> { + txn.execute( + "DELETE FROM msc4242_state_dag_forward_extremities WHERE room_id = $1", + &[&room_id], + ) + .await?; + let event_ids: Vec<&str> = new_extremities.iter().map(|id| &**id).collect(); + let room_ids = vec![room_id; event_ids.len()]; + txn.insert_many( + "msc4242_state_dag_forward_extremities", + &[ + Column::new("room_id", "text", &room_ids), + Column::new("event_id", "text", &event_ids), + ], + None, + ) + .await?; + Ok(()) +} + +/// An event for [`update_backward_extremities`]. +pub struct BackwardEvent<'a> { + pub event: &'a Event, + pub outlier: bool, + pub stream_ordering: i64, +} + +/// `_update_backward_extremeties`: the prev events of `events` that we do +/// not have, other than as outliers, become backward extremities, with a +/// timeline gap at each event that points at one. `events` themselves stop +/// being backward extremities, and their failed pull attempts go. +/// +/// For new events, and for outliers that are now persisted as +/// non-outliers. `events` must be of one room. +pub async fn update_backward_extremities( + txn: &WriteTxn, + instance_name: &str, + events: &[BackwardEvent<'_>], +) -> Result<(), WriteError> { + let Some(first) = events.first() else { + return Ok(()); + }; + let room_id = first.event.room_id(); + let prev_event_ids: Vec> = + events.iter().map(|e| e.event.prev_event_ids()).collect(); + + let mut potential: HashSet<&str> = events + .iter() + .zip(&prev_event_ids) + .filter(|(e, _)| !e.outlier) + .flat_map(|(_, prevs)| prevs.iter().map(String::as_str)) + .collect(); + if potential.is_empty() { + return Ok(()); + } + + let candidates: Vec<&str> = potential.iter().copied().collect(); + let rows = txn + .query( + "SELECT event_id FROM events WHERE event_id = ANY($1) AND outlier = FALSE", + &[&candidates], + ) + .await?; + for row in rows { + let event_id: &str = row.try_get(0).map_err(anyhow::Error::new)?; + potential.remove(event_id); + } + + if !potential.is_empty() { + let event_ids: Vec<&str> = potential.iter().copied().collect(); + let room_ids = vec![room_id; event_ids.len()]; + txn.upsert_many( + "event_backward_extremities", + &[ + Column::new("room_id", "text", &room_ids), + Column::new("event_id", "text", &event_ids), + ], + &[], + ) + .await?; + + // Record the stream orderings where we have new gaps. + let gap_orderings: Vec = events + .iter() + .zip(&prev_event_ids) + .filter(|(_, prevs)| prevs.iter().any(|prev| potential.contains(prev.as_str()))) + .map(|(e, _)| e.stream_ordering) + .collect(); + let room_ids = vec![room_id; gap_orderings.len()]; + let instance_names = vec![instance_name; gap_orderings.len()]; + txn.insert_many( + "timeline_gaps", + &[ + Column::new("room_id", "text", &room_ids), + Column::new("instance_name", "text", &instance_names), + Column::new("stream_ordering", "bigint", &gap_orderings), + ], + None, + ) + .await?; + } + + // An event with no prev events will never have anything to backfill + // from, so it stops being a backward extremity even as an outlier. + let to_remove: Vec<&str> = events + .iter() + .zip(&prev_event_ids) + .filter(|(e, prevs)| !e.outlier || prevs.is_empty()) + .map(|(e, _)| e.event.event_id()) + .collect(); + if !to_remove.is_empty() { + txn.execute( + "DELETE FROM event_backward_extremities WHERE event_id = ANY($1) AND room_id = $2", + &[&to_remove, &room_id], + ) + .await?; + txn.execute( + "DELETE FROM event_failed_pull_attempts WHERE event_id = ANY($1) AND room_id = $2", + &[&to_remove, &room_id], + ) + .await?; + } + Ok(()) +} diff --git a/rust/src/persist/txn/events.rs b/rust/src/persist/txn/events.rs new file mode 100644 index 0000000000..7332bada04 --- /dev/null +++ b/rust/src/persist/txn/events.rs @@ -0,0 +1,772 @@ +/* + * 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 steps that store the events themselves, from `_update_room_depths_txn` +//! to `_store_rejected_events_txn`, as `_persist_events_txn` runs them. + +use std::{ + collections::{HashMap, HashSet}, + sync::Arc, +}; + +use anyhow::{anyhow, Context}; +use log::{debug, info}; + +use super::{ + edges::{update_backward_extremities, BackwardEvent}, + json, metadata_error, redacts, PartialStateConflict, Persisting, TxnEvent, TxnInput, +}; +use crate::{ + database::postgres::write::{Column, WriteError, WriteTxn}, + event_auth::get_domain_from_id, + events::{constants::event_type, redact_event_py}, + persist::{ + chain_cover::NewEventChainLinks, + effects::{PersistEffects, StreamChangeCache}, + }, + room_versions::{EventFormatVersions, RoomVersion}, +}; + +/// `_update_room_depths_txn`: move the room's front in the events stream +/// cache to the chunk's highest stream ordering, unless it was backfilled, +/// and lower its `min_depth` to the deepest new event if that is lower. +pub async fn update_room_depths( + txn: &WriteTxn, + room_id: &str, + events: &[Persisting<'_>], + effects: &mut PersistEffects, +) -> Result<(), WriteError> { + let mut stream_ordering: Option = None; + let mut depth_update = 0; + for event in events { + // Backfilled events have negative orderings and are in the past. + if event.stream_ordering() >= 0 { + stream_ordering = stream_ordering.max(Some(event.stream_ordering())); + } + if !event.source.is_outlier()? && !event.context().is_rejected() { + depth_update = depth_update.max(event.event().depth()); + } + } + if let Some(stream_ordering) = stream_ordering { + effects.add_stream_change( + StreamChangeCache::Events, + room_id.to_owned(), + stream_ordering, + ); + } + + // `_update_min_depth_for_room_txn`. + let rows = txn + .query( + "SELECT min_depth FROM room_depth WHERE room_id = $1", + &[&room_id], + ) + .await?; + let min_depth: Option = match rows.first() { + Some(row) => row.try_get(0).map_err(anyhow::Error::new)?, + None => None, + }; + if min_depth.is_some_and(|min_depth| depth_update >= min_depth) { + return Ok(()); + } + txn.upsert_many( + "room_depth", + &[Column::new("room_id", "text", &vec![room_id])], + &[Column::new("min_depth", "bigint", &vec![depth_update])], + ) + .await?; + Ok(()) +} + +/// `_update_outliers_txn`: drop the events we have already persisted. An +/// event we have as an outlier, now persisted as a non-outlier, stops being +/// one: it gets its state group, a row in `ex_outlier_stream`, and its +/// backward extremities. +/// +/// Returns the events left, and adds to `de_outliered` the events that +/// stopped being outliers. +pub async fn update_outliers<'a>( + txn: &WriteTxn, + input: &TxnInput<'_>, + events: Vec>, + effects: &mut PersistEffects, + de_outliered: &mut Vec<&'a TxnEvent>, +) -> Result>, WriteError> { + let event_ids: Vec<&str> = events.iter().map(|e| e.event().event_id()).collect(); + let rows = txn + .query( + "SELECT event_id, outlier FROM events WHERE event_id = ANY($1)", + &[&event_ids], + ) + .await?; + let mut have_persisted: HashMap = HashMap::with_capacity(rows.len()); + for row in rows { + have_persisted.insert( + row.try_get(0).map_err(anyhow::Error::new)?, + row.try_get(1).map_err(anyhow::Error::new)?, + ); + } + + let mut left = Vec::with_capacity(events.len()); + for event in events { + let Some(&outlier_persisted) = have_persisted.get(event.event().event_id()) else { + left.push(event); + continue; + }; + // What we have is at least as good as a rejected copy. + if event.context().is_rejected() { + continue; + } + if event.source.is_outlier()? || !outlier_persisted { + continue; + } + + // We have the event as an outlier, and now have state at it. Its + // `events` row keeps the stream ordering it was first persisted + // with. `ex_outlier_stream` and any timeline gap get this chunk's. + info!( + "_update_outliers_txn: Updating state for ex-outlier event {}", + event.event().event_id() + ); + store_state_mappings(txn, std::slice::from_ref(&event), effects).await?; + + txn.insert_many( + "ex_outlier_stream", + &[ + Column::new( + "event_stream_ordering", + "bigint", + &vec![event.stream_ordering()], + ), + Column::new("event_id", "text", &vec![event.event().event_id()]), + Column::new("state_group", "bigint", &vec![event.context().state_group]), + Column::new("instance_name", "text", &vec![input.instance_name]), + ], + None, + ) + .await?; + txn.execute( + "UPDATE events SET outlier = FALSE WHERE event_id = $1", + &[&event.event().event_id()], + ) + .await?; + update_backward_extremities( + txn, + input.instance_name, + &[BackwardEvent { + event: event.event(), + outlier: false, + stream_ordering: event.stream_ordering(), + }], + ) + .await?; + de_outliered.push(event.source); + } + Ok(left) +} + +/// A redaction that applies to an event of the chunk. +struct RedactionInfo<'a> { + event_id: &'a str, + redacts: &'a str, + sender: &'a str, + needs_v3_recheck: bool, +} + +/// `_apply_existing_redaction_txn`: store redacted the events of the chunk +/// that an earlier event or another event of the chunk redacts. Redaction +/// events are never redacted here, so that circular redactions work. +pub async fn apply_existing_redactions<'a>( + txn: &WriteTxn, + room_id: &str, + room_version: &RoomVersion, + events: Vec>, +) -> Result>, WriteError> { + let event_ids: HashSet<&str> = events + .iter() + .filter(|e| e.event().type_() != event_type::M_ROOM_REDACTION) + .map(|e| e.event().event_id()) + .collect(); + if event_ids.is_empty() { + return Ok(events); + } + + let ids: Vec<&str> = event_ids.iter().copied().collect(); + let rows = txn + .query( + "SELECT redactions.event_id, redactions.redacts, redaction_events.sender, + redactions.recheck + FROM redactions + INNER JOIN events AS redaction_events + ON redactions.event_id = redaction_events.event_id + AND redaction_events.room_id = $1 + WHERE redactions.redacts = ANY($2)", + &[&room_id, &ids], + ) + .await?; + let mut redactions = Vec::with_capacity(rows.len()); + for row in &rows { + redactions.push(RedactionInfo { + event_id: row.try_get(0).map_err(anyhow::Error::new)?, + redacts: row.try_get(1).map_err(anyhow::Error::new)?, + sender: row.try_get(2).map_err(anyhow::Error::new)?, + needs_v3_recheck: row + .try_get::<_, Option>(3) + .map_err(anyhow::Error::new)? + .unwrap_or(false), + }); + } + for event in &events { + let event_ = event.event(); + let Some(redacts) = redacts(event_) else { + continue; + }; + if event_.type_() == event_type::M_ROOM_REDACTION + && event_ids.contains(redacts) + && !event.context().is_rejected() + { + redactions.push(RedactionInfo { + event_id: event_.event_id(), + redacts, + sender: event_.sender(), + needs_v3_recheck: event_ + .internal_metadata() + .need_to_check_redaction() + .map_err(metadata_error)?, + }); + } + } + + let events_by_id: HashMap<&str, &Persisting<'a>> = + events.iter().map(|e| (e.event().event_id(), e)).collect(); + // From each redacted event to the event that redacts it. + let mut redacted_by: HashMap = HashMap::new(); + for redaction in &redactions { + if redacted_by.contains_key(redaction.redacts) { + continue; + } + let event = events_by_id + .get(redaction.redacts) + .ok_or_else(|| anyhow!("redaction of {} is not in the chunk", redaction.redacts))? + .event(); + // The recheck column defaults to true, so v1 and v2 rooms can have + // it set, but their redactions never need one. + if redaction.needs_v3_recheck + && room_version.event_format != EventFormatVersions::ROOM_V1_V2 + { + let redaction_domain = get_domain_from_id(redaction.sender).map_err(domain_error)?; + let event_domain = get_domain_from_id(event.sender()).map_err(domain_error)?; + if redaction_domain != event_domain { + debug!( + "redaction of {} by {} skipped as it required a v3 recheck and sender servers differ", + redaction.redacts, redaction.event_id + ); + continue; + } + } + redacted_by.insert(redaction.redacts.to_owned(), redaction.event_id.to_owned()); + } + drop(events_by_id); + + events + .into_iter() + .map(|mut event| { + if let Some(by) = redacted_by.get(event.event().event_id()) { + let redacted = redact_event_py(event.event()).map_err(metadata_error)?; + redacted + .internal_metadata() + .set_redacted_by(Some(by.clone())) + .map_err(metadata_error)?; + event.redacted = Some(redacted); + } + Ok(event) + }) + .collect() +} + +fn domain_error(e: impl std::fmt::Display) -> WriteError { + WriteError::Other(anyhow!("{e}")) +} + +/// `_compute_newly_redacted_event_ids_txn`: the events of the room already +/// persisted that the chunk's redactions redact, where we can confirm the +/// redaction. A redaction that needed a recheck and passes it no longer +/// needs one, which is stored with it. +pub async fn newly_redacted_event_ids( + txn: &WriteTxn, + room_id: &str, + events: &[Persisting<'_>], +) -> Result, WriteError> { + let redactions: Vec<(&Persisting<'_>, &str)> = events + .iter() + // As in Python. A new event's rejection is in its context, not + // here, so a rejected redaction still counts. + .filter(|e| e.event().rejected_reason().is_none()) + // Python takes any `redacts`, but only a string can match. + .filter_map(|e| Some((e, redacts(e.event())?))) + .collect(); + if redactions.is_empty() { + return Ok(HashSet::new()); + } + + let targets: Vec<&str> = redactions.iter().map(|(_, redacts)| *redacts).collect(); + // The room ID keeps redactions from reaching into other rooms. + let rows = txn + .query( + "SELECT event_id, sender, type FROM events + WHERE event_id = ANY($1) AND room_id = $2", + &[&targets, &room_id], + ) + .await?; + let mut redacted_events: HashMap = HashMap::new(); + for row in rows { + let sender: &str = row.try_get(1).map_err(anyhow::Error::new)?; + redacted_events.insert( + row.try_get(0).map_err(anyhow::Error::new)?, + ( + get_domain_from_id(sender).map_err(domain_error)?.to_owned(), + row.try_get(2).map_err(anyhow::Error::new)?, + ), + ); + } + + let mut confirmed = HashSet::new(); + for (redaction, redacts) in redactions { + let Some((sender_domain, event_type)) = redacted_events.get(redacts) else { + // We cannot confirm a redaction of an event we do not have. + continue; + }; + // As `_maybe_redact_event_row`, create events are never redacted. + if event_type == event_type::M_ROOM_CREATE { + continue; + } + let metadata = redaction.event().internal_metadata(); + if metadata.need_to_check_redaction().map_err(metadata_error)? { + let domain = get_domain_from_id(redaction.event().sender()).map_err(domain_error)?; + if domain != sender_domain { + continue; + } + metadata + .set_recheck_redaction(false) + .map_err(metadata_error)?; + } + confirmed.insert(redacts.to_owned()); + } + Ok(confirmed) +} + +/// `delete_sticky_events_txn`: the events stop being sticky. +pub async fn delete_sticky_events( + txn: &WriteTxn, + event_ids: &HashSet, +) -> Result<(), WriteError> { + if event_ids.is_empty() { + return Ok(()); + } + let event_ids: Vec<&str> = event_ids.iter().map(String::as_str).collect(); + // `sticky_events` has no index on `event_id`, but has one on the + // stream ordering. + txn.execute( + "DELETE FROM sticky_events WHERE event_stream_ordering IN ( + SELECT stream_ordering FROM events WHERE event_id = ANY($1) + )", + &[&event_ids], + ) + .await?; + Ok(()) +} + +/// `_store_event_txn`: the `event_json`, `events` and `state_events` rows, +/// and the redactions of these events need censoring again. +pub async fn store_events( + txn: &WriteTxn, + input: &TxnInput<'_>, + events: &[Persisting<'_>], +) -> Result<(), WriteError> { + if events.is_empty() { + return Ok(()); + } + + let n = events.len(); + let mut event_ids = Vec::with_capacity(n); + let mut room_ids = Vec::with_capacity(n); + let mut internal_metadata = Vec::with_capacity(n); + let mut jsons = Vec::with_capacity(n); + let mut format_versions = Vec::with_capacity(n); + let mut instance_names = Vec::with_capacity(n); + let mut stream_orderings = Vec::with_capacity(n); + let mut depths = Vec::with_capacity(n); + let mut types = Vec::with_capacity(n); + let mut outliers = Vec::with_capacity(n); + let mut origin_server_tss = Vec::with_capacity(n); + let mut received_tss = Vec::with_capacity(n); + let mut senders = Vec::with_capacity(n); + let mut contains_urls = Vec::with_capacity(n); + let mut state_keys = Vec::with_capacity(n); + let mut rejection_reasons = Vec::with_capacity(n); + let mut unredacted = Vec::new(); + for persisting in events { + let event = persisting.event(); + let metadata = event.internal_metadata(); + event_ids.push(event.event_id()); + room_ids.push(event.room_id()); + internal_metadata.push( + json::encode_object(&metadata.to_persistence_pairs().map_err(metadata_error)?) + .map_err(anyhow::Error::new)?, + ); + jsons.push( + event + .with_persistence_dict(|dict| json::encode(dict)) + .map_err(metadata_error)? + .map_err(anyhow::Error::new)?, + ); + format_versions.push(i64::from(event.room_version().event_format)); + instance_names.push(input.instance_name); + stream_orderings.push(persisting.stream_ordering()); + depths.push(event.depth()); + types.push(event.type_()); + outliers.push(persisting.source.is_outlier()?); + origin_server_tss.push(event.origin_server_ts()); + received_tss.push(metadata.get_received_ts().map_err(metadata_error)?); + senders.push(event.sender()); + contains_urls.push( + event + .content() + .get_field("url") + .is_some_and(serde_json::Value::is_string), + ); + state_keys.push(event.state_key()); + rejection_reasons.push(persisting.context().rejected.as_deref()); + if !metadata.redacted().map_err(metadata_error)? { + unredacted.push(event.event_id()); + } + } + + txn.insert_many( + "event_json", + &[ + Column::new("event_id", "text", &event_ids), + Column::new("room_id", "text", &room_ids), + Column::new("internal_metadata", "text", &internal_metadata), + Column::new("json", "text", &jsons), + Column::new("format_version", "bigint", &format_versions), + ], + None, + ) + .await?; + + let processed = vec![true; n]; + txn.insert_many( + "events", + &[ + Column::new("instance_name", "text", &instance_names), + Column::new("stream_ordering", "bigint", &stream_orderings), + Column::new("topological_ordering", "bigint", &depths), + Column::new("depth", "bigint", &depths), + Column::new("event_id", "text", &event_ids), + Column::new("room_id", "text", &room_ids), + Column::new("type", "text", &types), + Column::new("processed", "boolean", &processed), + Column::new("outlier", "boolean", &outliers), + Column::new("origin_server_ts", "bigint", &origin_server_tss), + Column::new("received_ts", "bigint", &received_tss), + Column::new("sender", "text", &senders), + Column::new("contains_url", "boolean", &contains_urls), + Column::new("state_key", "text", &state_keys), + Column::new("rejection_reason", "text", &rejection_reasons), + ], + None, + ) + .await?; + + // A redaction of an event we now have unredacted needs to censor it. + if !unredacted.is_empty() { + txn.execute( + "UPDATE redactions SET have_censored = FALSE WHERE redacts = ANY($1)", + &[&unredacted], + ) + .await?; + } + + let state: Vec<_> = events.iter().filter(|e| e.event().is_state()).collect(); + let event_ids: Vec<&str> = state.iter().map(|e| e.event().event_id()).collect(); + let room_ids: Vec<&str> = state.iter().map(|e| e.event().room_id()).collect(); + let types: Vec<&str> = state.iter().map(|e| e.event().type_()).collect(); + let state_keys: Vec> = state.iter().map(|e| e.event().state_key()).collect(); + txn.insert_many( + "state_events", + &[ + Column::new("event_id", "text", &event_ids), + Column::new("room_id", "text", &room_ids), + Column::new("type", "text", &types), + Column::new("state_key", "text", &state_keys), + ], + None, + ) + .await?; + Ok(()) +} + +/// `_persist_transaction_ids_txn`: map the client's transaction ID to each +/// event that has one and a device ID, for deduplicating its retries. +pub async fn persist_transaction_ids( + txn: &WriteTxn, + input: &TxnInput<'_>, + events: &[Persisting<'_>], +) -> Result<(), WriteError> { + let mut rows = Vec::new(); + for persisting in events { + let event = persisting.event(); + let metadata = event.internal_metadata(); + let (Some(txn_id), Some(device_id)) = ( + metadata.txn_id().map_err(metadata_error)?, + metadata.device_id().map_err(metadata_error)?, + ) else { + continue; + }; + rows.push((event, device_id, txn_id)); + } + if rows.is_empty() { + return Ok(()); + } + let inserted_ts = input.now_ms()?; + let event_ids: Vec<&str> = rows.iter().map(|(e, _, _)| e.event_id()).collect(); + let room_ids: Vec<&str> = rows.iter().map(|(e, _, _)| e.room_id()).collect(); + let user_ids: Vec<&str> = rows.iter().map(|(e, _, _)| e.sender()).collect(); + let device_ids: Vec<&str> = rows.iter().map(|(_, d, _)| d.as_str()).collect(); + let txn_ids: Vec<&str> = rows.iter().map(|(_, _, t)| t.as_str()).collect(); + let inserted_tss = vec![inserted_ts; rows.len()]; + txn.insert_many( + "event_txn_id_device_id", + &[ + Column::new("event_id", "text", &event_ids), + Column::new("room_id", "text", &room_ids), + Column::new("user_id", "text", &user_ids), + Column::new("device_id", "text", &device_ids), + Column::new("txn_id", "text", &txn_ids), + Column::new("inserted_ts", "bigint", &inserted_tss), + ], + None, + ) + .await?; + Ok(()) +} + +/// `_store_event_state_mappings_txn`: the state group of each event, and +/// which events have partial state. +/// +/// Fails with [`PartialStateConflict`] when an event has partial state in +/// a room that no longer has. +pub async fn store_state_mappings( + txn: &WriteTxn, + events: &[Persisting<'_>], + effects: &mut PersistEffects, +) -> Result<(), WriteError> { + // Event ID to state group, in first-seen order. + let mut state_groups: Vec<(&str, Option)> = Vec::new(); + let mut index_of: HashMap<&str, usize> = HashMap::new(); + for persisting in events { + let event = persisting.event(); + let context = persisting.context(); + // State DAG rooms give outliers state: the events + // `/get_missing_events` returns are nominally outliers but need + // state for their auth events. + if !event.room_version().msc4242_state_dags && persisting.source.is_outlier()? { + if context.partial_state { + return Err(anyhow!( + "Outlier event {} claims to have partial state", + event.event_id() + ) + .into()); + } + continue; + } + // A rejected event gets the state of its predecessor. + let group = if context.is_rejected() { + context.state_group_before_event + } else { + context.state_group + }; + match index_of.get(event.event_id()) { + Some(&i) => state_groups[i].1 = group, + None => { + index_of.insert(event.event_id(), state_groups.len()); + state_groups.push((event.event_id(), group)); + } + } + } + + let partial: Vec<&Persisting<'_>> = events + .iter() + .filter(|e| e.context().partial_state) + .collect(); + let event_ids: Vec<&str> = partial.iter().map(|e| e.event().event_id()).collect(); + let room_ids: Vec<&str> = partial.iter().map(|e| e.event().room_id()).collect(); + let inserted = txn + .insert_many( + "partial_state_events", + &[ + Column::new("room_id", "text", &room_ids), + Column::new("event_id", "text", &event_ids), + ], + None, + ) + .await; + if let Err(WriteError::Constraint { .. }) = inserted { + // The room's partial state rows are gone: it was un-partial stated. + let rooms: HashSet<&str> = room_ids.iter().copied().collect(); + info!( + "Cannot persist events {event_ids:?} in rooms {rooms:?}: room has been un-partial stated" + ); + return Err(WriteError::Other(anyhow::Error::new(PartialStateConflict))); + } + inserted?; + + let event_ids: Vec<&str> = state_groups.iter().map(|(id, _)| *id).collect(); + let groups: Vec> = state_groups.iter().map(|(_, group)| *group).collect(); + txn.upsert_many( + "event_to_state_groups", + &[Column::new("event_id", "text", &event_ids)], + &[Column::new("state_group", "bigint", &groups)], + ) + .await?; + + for (event_id, group) in state_groups { + // Python would prefill a missing group too, but the insert above + // has failed by then. + if let Some(group) = group { + effects.add_state_group_prefill(event_id.to_owned(), group); + } + } + Ok(()) +} + +/// `_persist_event_auth_chain_txn`: the chain cover index of the new state +/// events, which [`crate::persist::chain_cover`] calculated, and their +/// auth events, rejected ones included as state resolution reads them. +pub async fn persist_auth_chain( + txn: &WriteTxn, + events: &[Persisting<'_>], + new_event_links: &HashMap, NewEventChainLinks>, +) -> Result<(), WriteError> { + if !new_event_links.is_empty() { + // `_persist_chain_cover_index`. + let event_ids: Vec<&str> = new_event_links.keys().map(|id| &**id).collect(); + let chain_ids: Vec = new_event_links.values().map(|l| l.chain_id).collect(); + let sequence_numbers: Vec = new_event_links + .values() + .map(|l| l.sequence_number) + .collect(); + txn.insert_many( + "event_auth_chains", + &[ + Column::new("event_id", "text", &event_ids), + Column::new("chain_id", "bigint", &chain_ids), + Column::new("sequence_number", "bigint", &sequence_numbers), + ], + None, + ) + .await?; + txn.execute( + "DELETE FROM event_auth_chain_to_calculate WHERE event_id = ANY($1)", + &[&event_ids], + ) + .await?; + + let (mut origin_chains, mut origin_seqs) = (Vec::new(), Vec::new()); + let (mut target_chains, mut target_seqs) = (Vec::new(), Vec::new()); + for l in new_event_links.values() { + for &(chain, seq) in &l.links { + origin_chains.push(l.chain_id); + origin_seqs.push(l.sequence_number); + target_chains.push(chain); + target_seqs.push(seq); + } + } + txn.insert_many( + "event_auth_chain_links", + &[ + Column::new("origin_chain_id", "bigint", &origin_chains), + Column::new("origin_sequence_number", "bigint", &origin_seqs), + Column::new("target_chain_id", "bigint", &target_chains), + Column::new("target_sequence_number", "bigint", &target_seqs), + ], + None, + ) + .await?; + } + + let mut event_ids = Vec::new(); + let mut room_ids = Vec::new(); + let mut auth_ids = Vec::new(); + for persisting in events { + let event = persisting.event(); + if !event.is_state() { + continue; + } + let auth_events = event + .auth_event_ids() + .map_err(metadata_error) + .with_context(|| format!("reading the auth events of {}", event.event_id()))?; + for auth_id in auth_events { + event_ids.push(event.event_id()); + room_ids.push(event.room_id()); + auth_ids.push(auth_id); + } + } + txn.insert_many( + "event_auth", + &[ + Column::new("event_id", "text", &event_ids), + Column::new("room_id", "text", &room_ids), + Column::new("auth_id", "text", &auth_ids), + ], + None, + ) + .await?; + Ok(()) +} + +/// `_store_rejected_events_txn`: a `rejections` row for each rejected +/// event, which goes no further. Returns the events left. +pub async fn store_rejections<'a>( + txn: &WriteTxn, + input: &TxnInput<'_>, + events: Vec>, +) -> Result>, WriteError> { + let (rejected, left): (Vec<_>, Vec<_>) = + events.into_iter().partition(|e| e.context().is_rejected()); + if rejected.is_empty() { + return Ok(left); + } + let event_ids: Vec<&str> = rejected.iter().map(|e| e.event().event_id()).collect(); + let reasons: Vec<&str> = rejected + .iter() + .map(|e| e.context().rejected.as_deref().unwrap_or_default()) + .collect(); + // `last_check` is a text column. + let last_checks = vec![input.now_ms()?.to_string(); rejected.len()]; + txn.insert_many( + "rejections", + &[ + Column::new("event_id", "text", &event_ids), + Column::new("reason", "text", &reasons), + Column::new("last_check", "text", &last_checks), + ], + None, + ) + .await?; + Ok(left) +} diff --git a/rust/src/persist/txn/json.rs b/rust/src/persist/txn/json.rs new file mode 100644 index 0000000000..f883a60901 --- /dev/null +++ b/rust/src/persist/txn/json.rs @@ -0,0 +1,121 @@ +/* + * 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: + * . + * + */ + +//! JSON as Python's `json_encoder` writes it: compact, with every character +//! outside printable ASCII escaped (`ensure_ascii`). Keys come in the order +//! the value holds them, so a value with Python's key order gives Python's +//! text. +//! +//! Numbers are written as the value holds them. Python writes a float in +//! its own way, so number text from federation can differ (`1.50` is +//! stored as `1.5` by Python). The value is the same. + +use std::io; + +use serde::Serialize; +use serde_json::ser::{CompactFormatter, Formatter, Serializer}; + +/// A compact formatter that escapes non-ASCII characters and DEL as +/// `\uXXXX`, in UTF-16 code units as Python does. serde escapes the other +/// control characters before they reach it. +struct AsciiFormatter; + +impl Formatter for AsciiFormatter { + fn write_string_fragment(&mut self, writer: &mut W, fragment: &str) -> io::Result<()> + where + W: ?Sized + io::Write, + { + let mut start = 0; + for (i, c) in fragment.char_indices() { + if c.is_ascii() && c != '\x7f' { + continue; + } + writer.write_all(&fragment.as_bytes()[start..i])?; + let mut units = [0u16; 2]; + for unit in c.encode_utf16(&mut units) { + write!(writer, "\\u{unit:04x}")?; + } + start = i + c.len_utf8(); + } + writer.write_all(&fragment.as_bytes()[start..]) + } + + // The rest as the compact formatter writes it. + fn begin_array_value(&mut self, writer: &mut W, first: bool) -> io::Result<()> + where + W: ?Sized + io::Write, + { + CompactFormatter.begin_array_value(writer, first) + } + + fn begin_object_key(&mut self, writer: &mut W, first: bool) -> io::Result<()> + where + W: ?Sized + io::Write, + { + CompactFormatter.begin_object_key(writer, first) + } + + fn begin_object_value(&mut self, writer: &mut W) -> io::Result<()> + where + W: ?Sized + io::Write, + { + CompactFormatter.begin_object_value(writer) + } +} + +/// Encode `value` as Python's `json_encoder.encode` does. +pub fn encode(value: &T) -> serde_json::Result { + let mut out = Vec::new(); + value.serialize(&mut Serializer::with_formatter(&mut out, AsciiFormatter))?; + Ok(String::from_utf8(out).expect("the formatter writes ASCII")) +} + +/// Encode `pairs` as an object, with the keys in the order given. +pub fn encode_object(pairs: &[(K, V)]) -> serde_json::Result { + struct Object<'a, K, V>(&'a [(K, V)]); + + impl Serialize for Object<'_, K, V> { + fn serialize(&self, serializer: S) -> Result { + serializer.collect_map(self.0.iter().map(|(k, v)| (k, v))) + } + } + + encode(&Object(pairs)) +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + + #[test] + fn matches_python() { + // `json_encoder.encode({"a": ["é", "👍", "\x01", "\x7f~", "\"\\"], "b": 1.5})` + let value = json!({"a": ["é", "👍", "\u{1}", "\u{7f}~", "\"\\"], "b": 1.5}); + assert_eq!( + encode(&value).unwrap(), + r#"{"a":["\u00e9","\ud83d\udc4d","\u0001","\u007f~","\"\\"],"b":1.5}"# + ); + } + + #[test] + fn object_keeps_key_order() { + assert_eq!( + encode_object(&[("z", json!(1)), ("a", json!("é"))]).unwrap(), + r#"{"z":1,"a":"\u00e9"}"# + ); + } +} diff --git a/rust/src/persist/txn/mod.rs b/rust/src/persist/txn/mod.rs index 1a75b1e76d..a33c73b538 100644 --- a/rust/src/persist/txn/mod.rs +++ b/rust/src/persist/txn/mod.rs @@ -21,18 +21,29 @@ //! 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 dropping duplicate events, -//! and returns the events left at that point. The Python steps after it run +//! 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. +pub mod edges; +pub mod events; +pub mod json; + use std::collections::HashMap; use anyhow::anyhow; +use pyo3::PyErr; use serde_json::Value; -use super::{context::PersistContext, effects::EventInvalidation, effects::PersistEffects}; +use super::{ + context::PersistContext, + effects::{EventInvalidation, PersistEffects}, + prepare::PreparedChunk, +}; use crate::{ + clock::Clock, database::postgres::write::{WriteError, WriteTxn}, events::{ constants::{event_field::M_RELATES_TO, relation_type}, @@ -75,41 +86,163 @@ impl TxnEvent { } } +/// An event on its way through the transaction: an event of the chunk, or +/// its redacted copy once [`events::apply_existing_redactions`] finds it +/// already redacted. +pub struct Persisting<'a> { + pub source: &'a TxnEvent, + /// The event as it is stored, if it is redacted. + pub redacted: Option, +} + +impl Persisting<'_> { + pub fn event(&self) -> &Event { + self.redacted.as_ref().unwrap_or(&self.source.event) + } + + pub fn context(&self) -> &PersistContext { + &self.source.context + } + + pub fn stream_ordering(&self) -> i64 { + self.source.stream_ordering + } +} + /// What the transaction takes besides the chunk. pub struct TxnInput<'a> { pub room_id: &'a str, /// The chunk, in order. It may hold the same event twice. pub events: &'a [TxnEvent], + /// What [`super::prepare`] computed for the chunk. + pub prepared: &'a PreparedChunk, + /// This instance, which writes the events. + pub instance_name: &'a str, + pub clock: &'a Clock, +} + +impl TxnInput<'_> { + /// The time, read when a step needs it, as Python reads its clock. + fn now_ms(&self) -> Result { + Ok(i64::try_from(self.clock.now_millis()).map_err(anyhow::Error::new)?) + } +} + +/// What the steps ported so far leave for the Python steps after them. +pub struct TxnPrefix<'a> { + /// The events left: neither persisted before nor rejected. + pub events: Vec>, + /// The events that stopped being outliers. Python makes the sticky + /// ones sticky. + pub de_outliered: Vec<&'a TxnEvent>, +} + +/// The chunk has an event with partial state, but the room no longer has +/// partial state. Python raises `PartialStateConflictError`, and the caller +/// computes the event's context again. +#[derive(Debug)] +pub struct PartialStateConflict; + +impl std::fmt::Display for PartialStateConflict { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("Cannot persist partial state event in un-partial stated room") + } +} + +impl std::error::Error for PartialStateConflict {} + +/// An error reading an event's fields through the Python-facing accessors. +fn metadata_error(e: PyErr) -> WriteError { + WriteError::Other(anyhow::Error::new(e)) } /// Write the chunk, as `_persist_events_txn` does, adding to `effects` what -/// Python would queue with `txn.call_after`. Returns the events left once -/// duplicates are dropped, which the steps still to be ported take. +/// 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. pub async fn persist_events_txn<'a>( txn: &WriteTxn, input: &TxnInput<'a>, effects: &mut PersistEffects, -) -> Result, WriteError> { +) -> Result, WriteError> { + let room_id = input.room_id; + // Protect against a race with deleting the room: the share lock holds // off a delete until we commit. let rows = txn .query( "SELECT room_version FROM rooms WHERE room_id = $1 FOR SHARE", - &[&input.room_id], + &[&room_id], ) .await?; if rows.is_empty() { - return Err(anyhow!("Room does not exist {}", input.room_id).into()); + return Err(anyhow!("Room does not exist {room_id}").into()); } + let (Some(first), Some(last)) = (input.events.first(), input.events.last()) else { + return Err(anyhow!("no events to persist in {room_id}").into()); + }; + let max_stream_order = last.stream_ordering; + // Python takes the first event's room version for the whole chunk. + let room_version = first.event.room_version(); + // Recorded for every event of the chunk, before any are dropped. for event in input.events { add_event_effects(event, effects); } - filter_duplicates(input.events) + let events: Vec> = filter_duplicates(input.events)? + .into_iter() + .map(|source| Persisting { + source, + redacted: None, + }) + .collect(); + + events::update_room_depths(txn, room_id, &events, effects).await?; + + let mut de_outliered = Vec::new(); + let events = events::update_outliers(txn, input, events, effects, &mut de_outliered).await?; + + // From here on, only events we have not seen before. + let events = events::apply_existing_redactions(txn, room_id, room_version, events).await?; + let newly_redacted = events::newly_redacted_event_ids(txn, room_id, &events).await?; + // Redacted events stop being sticky. Python makes de-outliered events + // sticky in `update_outliers`, before this. The sticky port must do so + // there too, or a de-outliered event redacted in the same chunk stays + // sticky. + events::delete_sticky_events(txn, &newly_redacted).await?; + + events::store_events(txn, input, &events).await?; + + let state = &input.prepared.state; + if let Some(extremities) = state + .new_forward_extremities + .as_ref() + .filter(|e| !e.is_empty()) + { + edges::update_forward_extremities(txn, room_id, extremities, max_stream_order).await?; + } + if let Some(extremities) = state + .new_state_dag_extremities + .as_ref() + .filter(|e| !e.is_empty()) + { + edges::set_state_dag_extremities(txn, room_id, extremities).await?; + } + + events::persist_transaction_ids(txn, input, &events).await?; + events::store_state_mappings(txn, &events, effects).await?; + events::persist_auth_chain(txn, &events, &input.prepared.new_event_links).await?; + + // From here on, only events that were not rejected. + let events = events::store_rejections(txn, input, events).await?; + + Ok(TxnPrefix { + events, + de_outliered, + }) } /// The cache updates Python queues for each event of the chunk at the start diff --git a/synapse/synapse_rust/persist.pyi b/synapse/synapse_rust/persist.pyi index a3a5019461..f3dac84778 100644 --- a/synapse/synapse_rust/persist.pyi +++ b/synapse/synapse_rust/persist.pyi @@ -55,11 +55,16 @@ class Persister: self, room_id: str, events_and_contexts: list[tuple[EventBase, EventContext]], - ) -> tuple[PersistEffects, list[int]]: + prepared: PreparedChunk, + ) -> 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. Returns what it collected for Python to apply and the positions - in the chunk of the events left for the Python steps after it.""" + it. `prepared` is what Rust prepared for it. + + 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.""" class PreparedChunk: """What `Persister.prepare_chunk` computed. If `state_res_fallback` is diff --git a/tests/storage/test_rust_persist_prereqs.py b/tests/storage/test_rust_persist_prereqs.py index dba68117be..f523b9f9fa 100644 --- a/tests/storage/test_rust_persist_prereqs.py +++ b/tests/storage/test_rust_persist_prereqs.py @@ -20,7 +20,6 @@ native Rust Postgres driver: SYNAPSE_TEST_RUST_DRIVER=1 POSTGRES=1 run-trial.sh tests.storage.test_rust_persist_prereqs """ -import json from contextlib import ExitStack from typing import Any, Collection from unittest import SkipTest @@ -65,23 +64,19 @@ class PersistenceJsonTestCase(unittest.TestCase): """The `event_json` columns Rust writes match what `_store_event_txn` writes.""" - def _python_columns(self, event: EventBase) -> tuple[Any, Any]: + def _python_columns(self, event: EventBase) -> tuple[str, str]: # As `_store_event_txn` builds them. d = event.get_dict_for_persistence() d.pop("redacted", None) d.pop("redacted_because", None) - # Compared as stored: JSON has no tuples. return ( - json.loads(json_encoder.encode(event.internal_metadata.get_dict())), - json.loads(json_encoder.encode(d)), + json_encoder.encode(event.internal_metadata.get_dict()), + json_encoder.encode(d), ) def _assert_same(self, event: EventBase) -> None: - internal_metadata, event_json = persistence_json_for_test(event) - self.assertEqual( - (json.loads(internal_metadata), json.loads(event_json)), - self._python_columns(event), - ) + # The text, byte for byte, key order and escaping included. + self.assertEqual(persistence_json_for_test(event), self._python_columns(event)) def test_message(self) -> None: self._assert_same( @@ -104,6 +99,33 @@ class PersistenceJsonTestCase(unittest.TestCase): ) ) + def test_escapes_and_floats(self) -> None: + # Room versions before 6 allow floats. + self._assert_same( + make_event_from_dict( + { + "type": "m.room.message", + "sender": "@alice:test", + "room_id": "!room:test", + "content": { + "body": 'caf\u00e9 \U0001f44d \x01\x7f~ "\\ None: # `replaces_state` and `invite_room_state` are stored; `prev_content` # is not. diff --git a/tests/storage/test_rust_persist_txn.py b/tests/storage/test_rust_persist_txn.py index 605ff09d62..d8882ff182 100644 --- a/tests/storage/test_rust_persist_txn.py +++ b/tests/storage/test_rust_persist_txn.py @@ -35,11 +35,13 @@ from unittest import SkipTest import attr +from twisted.internet.defer import Deferred from twisted.internet.testing import MemoryReactor -from synapse.api.constants import EventTypes +from synapse.api.constants import EventTypes, StickyEvent +from synapse.api.errors import PartialStateConflictError from synapse.api.room_versions import KNOWN_ROOM_VERSIONS -from synapse.events import make_event_from_dict +from synapse.events import EventBase, make_event_from_dict from synapse.events.snapshot import EventContext, EventPersistencePair from synapse.rest import admin from synapse.rest.client import login, room @@ -47,6 +49,8 @@ from synapse.server import HomeServer from synapse.storage.database import LoggingTransaction from synapse.storage.databases.main.events import PersistEventsStore from synapse.storage.rust_pool import RustConnectionPool +from synapse.storage.util.id_generators import MultiWriterIdGenerator +from synapse.synapse_rust.events import redact_event from synapse.synapse_rust.persist import PersistEffects from synapse.types import create_requester from synapse.util.caches.deferred_cache import DeferredCache @@ -128,14 +132,23 @@ _JSON_COLUMNS = { } # Which positional arguments of a `call_after` callback hold a stream -# position or state group, by callback name. -_EFFECT_ARGS = { +# position or state group, by callback name, or by its name and owner. +_EFFECT_ARGS: dict[str | tuple[str, str | None], tuple[str | None, ...]] = { "_invalidate_caches_for_event": ("events",), "entity_has_changed": (None, "events"), "prefill": (None, "state_groups"), - "_mark_ids_as_finished": ("cache",), + # An ID generator's stream IDs, by the generator's stream name. + ("_mark_ids_as_finished", "caches"): ("cache",), + ("_mark_ids_as_finished", "sticky_events"): ("sticky",), } + +def _effect_kinds(name: str, owner: str | None, args: tuple) -> tuple[str | None, ...]: + """The kind of each argument of an effect, for `_Normaliser.position`.""" + kinds = _EFFECT_ARGS.get((name, owner), _EFFECT_ARGS.get(name, ())) + return kinds + (None,) * (len(args) - len(kinds)) + + # Marks each twin's create event, so the rooms get different IDs in room # versions that derive the room ID from it. _TWIN_KEY = "org.matrix.synapse.test.twin" @@ -254,12 +267,15 @@ class _TwinTestBase(unittest.HomeserverTestCase): self.persistence._rust_persist_strict = True persister = persistence._rust_persister assert persister is not None - self.persister = persister + self.persister = _RecordingPersister(persister) + persistence._rust_persister = self.persister # type: ignore[assignment] events_store = hs.get_datastores().persist_events assert events_store is not None self.events_store = events_store self.alice = self.register_user("alice", "pass") + # A device, for transaction IDs. + self.login("alice", "pass", device_id="DEVICE") self.bob = self.register_user("bob", "pass") self.twins = (_Twin("a"), _Twin("b")) for twin in self.twins: @@ -318,17 +334,35 @@ class _TwinTestBase(unittest.HomeserverTestCase): ) -> None: """Room B's transaction: the Rust part, committed first, then the rest in Python.""" - effects, left = self.persister.persist_events_txn_for_test( - room_id, events_and_contexts + prepared = self.persister.last_prepared + assert prepared is not None + # The transaction must take the chunk prepared for it. + self.assertEqual( + self.persister.last_chunk, + (room_id, [event.event_id for event, _ in events_and_contexts]), + ) + effects, left, redacted, de_outliered = ( + self.persister.persist_events_txn_for_test( + room_id, events_and_contexts, prepared + ) ) twin.effects.extend(_rust_effects(effects)) + # The events left, as the steps after Rust's see them. + left_events_and_contexts = [] + for i in left: + event, context = events_and_contexts[i] + if i in redacted: + event = redact_event(event) + event.internal_metadata.redacted_by = redacted[i] + left_events_and_contexts.append((event, context)) start = len(txn.after_callbacks or []) _python_steps_after_rust( self.events_store, txn, room_id=room_id, all_events_and_contexts=events_and_contexts, - events_and_contexts=[events_and_contexts[i] for i in left], + events_and_contexts=left_events_and_contexts, + de_outliered=[events_and_contexts[i][0] for i in de_outliered], **kwargs, ) twin.effects.extend(self._python_effects(txn, start)) @@ -351,6 +385,8 @@ class _TwinTestBase(unittest.HomeserverTestCase): ) if owner is self.store: owner_name = None + 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: @@ -393,6 +429,8 @@ class _TwinTestBase(unittest.HomeserverTestCase): prev: list[int] | None = None, redacts: int | None = None, sender: str | None = None, + txn_id: str | None = None, + sticky_ms: int | None = None, ) -> 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 @@ -408,6 +446,8 @@ class _TwinTestBase(unittest.HomeserverTestCase): } if state_key is not None: event_dict["state_key"] = state_key + if sticky_ms is not None: + event_dict[StickyEvent.EVENT_FIELD_NAME] = {"duration_ms": sticky_ms} if redacts is not None: redacts_id = twin.events[redacts] event_dict["redacts"] = redacts_id @@ -415,8 +455,9 @@ class _TwinTestBase(unittest.HomeserverTestCase): event_dict["content"]["redacts"] = redacts_id event, unpersisted_context = self.get_success( self.hs.get_event_creation_handler().create_event( - create_requester(sender), + create_requester(sender, device_id="DEVICE" if txn_id else None), event_dict, + txn_id=txn_id, prev_event_ids=( [twin.events[i] for i in prev] if prev is not None else None ), @@ -527,9 +568,7 @@ class _TwinTestBase(unittest.HomeserverTestCase): normaliser.position(kind, arg) if kind else normaliser.value("", "", arg) - for kind, arg in zip( - _EFFECT_ARGS.get(name, ()) + (None,) * len(args), args - ) + for kind, arg in zip(_effect_kinds(name, owner, args), args) ), tuple(sorted(kwargs.items())), ) @@ -563,6 +602,17 @@ class _TwinTestBase(unittest.HomeserverTestCase): row[i] for row in rows if isinstance(row[i], int) ) values.setdefault("cache", set()).update(row[0] for row in twin.cache_rows) + # Positions that only an effect holds, such as the stream ID of a + # sticky event that was redacted. + for name, owner, args, _ in twin.effects: + for kind, arg in zip(_effect_kinds(name, owner, args), args): + if not kind: + continue + found = values.setdefault(kind, set()) + if isinstance(arg, int): + found.add(arg) + elif isinstance(arg, (list, tuple)): + found.update(v for v in arg if isinstance(v, int)) ranks = { kind: {value: rank for rank, value in enumerate(sorted(found))} for kind, found in values.items() @@ -602,6 +652,38 @@ class _TwinTestBase(unittest.HomeserverTestCase): self.assertEqual(a[table], b[table], f"{table} differs") +class _RecordingPersister: + """The Rust persister, remembering the last chunk it prepared, which + room B's transaction takes.""" + + def __init__(self, persister: Any) -> None: + self._persister = persister + self.last_prepared: Any = None + # The room and event IDs `last_prepared` is for. + self.last_chunk: tuple[str, list[str]] | None = None + + def prepare_chunk( + self, + room_id: str, + events_and_contexts: list[EventPersistencePair], + *args: Any, + **kwargs: Any, + ) -> "Deferred[Any]": + chunk = (room_id, [event.event_id for event, _ in events_and_contexts]) + + def record(prepared: Any) -> Any: + self.last_prepared = prepared + self.last_chunk = chunk + return prepared + + return self._persister.prepare_chunk( + room_id, events_and_contexts, *args, **kwargs + ).addCallback(record) + + def __getattr__(self, name: str) -> Any: + return getattr(self._persister, name) + + def _rust_effects( effects: PersistEffects, ) -> list[tuple[str, str | None, tuple, dict[str, Any]]]: @@ -618,9 +700,31 @@ def _rust_effects( ) for event_id in effects.event_cache_invalidations: calls.append(("_invalidate_local_get_event_cache", None, (event_id,), {})) + for cache, entity, stream_id in effects.stream_changes: + calls.append( + ( + "entity_has_changed", + _STREAM_CHANGE_CACHES[cache], + (entity, stream_id), + {}, + ) + ) + for event_id, state_group in effects.state_group_prefills: + calls.append( + ("prefill", "_get_state_group_for_event", ((event_id,), state_group), {}) + ) return calls +# The names of the stream change caches, by `PersistEffects.stream_changes` +# key. +_STREAM_CHANGE_CACHES = { + "events": "EventsRoomStreamChangeCache", + "current_state_deltas": "_curr_state_delta_stream_cache", + "membership": "_membership_stream_cache", +} + + def _python_steps_after_rust( store: PersistEventsStore, txn: LoggingTransaction, @@ -628,6 +732,7 @@ def _python_steps_after_rust( room_id: str, all_events_and_contexts: list[EventPersistencePair], events_and_contexts: list[EventPersistencePair], + de_outliered: list[EventBase], inhibit_local_membership_updates: bool, state_delta_for_room: Any, new_forward_extremities: set[str] | None, @@ -645,44 +750,15 @@ def _python_steps_after_rust( transaction meanwhile runs in room A only, so its rows differ and the comparison fails.""" min_stream_order = all_events_and_contexts[0][0].internal_metadata.stream_ordering - room_version = all_events_and_contexts[0][0].room_version - store._update_room_depths_txn(txn, room_id, events_and_contexts=events_and_contexts) - events_and_contexts = store._update_outliers_txn( - txn, events_and_contexts=events_and_contexts - ) - events_and_contexts = store._apply_existing_redaction_txn( - txn, room_id, room_version, events_and_contexts=events_and_contexts - ) - redacted = store._compute_newly_redacted_event_ids_txn( - txn, room_id, room_version, events_and_contexts=events_and_contexts - ) - if redacted: - store.store.delete_sticky_events_txn(txn, redacted) - store._store_event_txn(txn, events_and_contexts=events_and_contexts) - if new_forward_extremities: - max_stream_order = all_events_and_contexts[-1][ - 0 - ].internal_metadata.stream_ordering - assert max_stream_order is not None - store._update_forward_extremities_txn( - txn, - room_id, - new_forward_extremities=new_forward_extremities, - max_stream_order=max_stream_order, + # Left by `_update_outliers_txn`, until Rust inserts sticky events. + # Python inserts these before deleting the redacted events' rows, so a + # chunk that de-outliers a sticky event and redacts it differs here. + if store._msc4354_enabled: + store.store.insert_sticky_events_txn( + txn, [event for event in de_outliered if event.sticky_duration()] ) - if new_state_dag_forward_extremities: - store._set_state_dag_extremities_txn( - txn, room_id, new_state_dag_forward_extremities - ) - store._persist_transaction_ids_txn(txn, events_and_contexts) - store._store_event_state_mappings_txn(txn, events_and_contexts) - store._persist_event_auth_chain_txn( - txn, [e for e, _ in events_and_contexts], new_event_links - ) - events_and_contexts = store._store_rejected_events_txn( - txn, events_and_contexts=events_and_contexts - ) + store._update_metadata_tables_txn( txn, events_and_contexts=events_and_contexts, @@ -739,6 +815,9 @@ class PersistTxnTestCase(_TwinTestBase): def test_fork(self) -> None: base = self._message("base") self._send(EventTypes.Topic, {"topic": "A"}, state_key="", prev=[base]) + # State resolution orders the topics by timestamp, then by event ID, + # which differs between the twins. + self.reactor.advance(1) self._send(EventTypes.Topic, {"topic": "B"}, state_key="", prev=[base]) # Joins the two branches, resolving state. self._message("merge") @@ -766,6 +845,34 @@ class PersistTxnTestCase(_TwinTestBase): self._step(lambda twin: [built[twin.name]]) self._assert_twins_match() + def test_redaction_of_later_redaction(self) -> None: + """A redaction is never redacted as it is persisted, so a redaction + of it that came first needs censoring again.""" + message = self._message("secret") + built: dict[str, EventPersistencePair] = {} + + def target(twin: _Twin) -> list[EventPersistencePair]: + built[twin.name] = self._build( + twin, EventTypes.Redaction, {}, redacts=message + ) + return [] + + self._step(target) + self._send( + EventTypes.Redaction, + {}, + redacts=len(self.twins[0].events) - 1, + prev=[message], + ) + # As if the censor job had run. + self.get_success( + self.store.db_pool.simple_update( + "redactions", {}, {"have_censored": True}, desc="test" + ) + ) + self._step(lambda twin: [built[twin.name]]) + self._assert_twins_match() + def test_relation(self) -> None: root = self._message("root") self._send( @@ -778,6 +885,74 @@ class PersistTxnTestCase(_TwinTestBase): ) self._assert_twins_match() + def test_transaction_ids(self) -> None: + self._message("hello", txn_id="txn1") + self._assert_twins_match() + + def test_url_and_unicode(self) -> None: + self._send( + EventTypes.Message, + {"msgtype": "m.image", "body": "h\u00e9llo \U0001f44d", "url": "mxc://a/b"}, + ) + # Not a string, so not a URL. + self._send(EventTypes.Message, {"msgtype": "m.text", "body": "x", "url": 1}) + self._assert_twins_match() + + def test_redaction_recheck(self) -> None: + """A redaction that needs a recheck, which passes as Alice redacts + her own event, is stored as no longer needing one.""" + target = self._message("secret") + + def redaction(twin: _Twin) -> list[EventPersistencePair]: + event, context = self._build(twin, EventTypes.Redaction, {}, redacts=target) + event.internal_metadata.recheck_redaction = True + return [(event, context)] + + self._step(redaction) + self._assert_twins_match() + + def test_redaction_recheck_in_chunk(self) -> None: + """The same, with the redaction and its target in one chunk.""" + base = self._message("base") + + def chunk(twin: _Twin) -> list[EventPersistencePair]: + target = self._build( + twin, EventTypes.Message, {"msgtype": "m.text", "body": "secret"} + ) + event, context = self._build( + twin, + EventTypes.Redaction, + {}, + redacts=len(twin.events) - 1, + prev=[base], + ) + event.internal_metadata.recheck_redaction = True + return [target, (event, context)] + + self._step(chunk) + self._assert_twins_match() + + def test_partial_state_conflict(self) -> None: + """An event with partial state in a room without fails, as the room + must have been un-partial stated meanwhile.""" + for twin in self.twins: + event, context = self._build( + twin, EventTypes.Message, {"msgtype": "m.text", "body": "partial"} + ) + context.partial_state = True + with self._persisting(twin): + self.get_failure( + self.persistence.persist_events([(event, context)]), + PartialStateConflictError, + ) + self._assert_twins_match() + + def test_sticky_redacted(self) -> None: + """Redacting a sticky event makes it stop being sticky.""" + target = self._message("sticky", sticky_ms=60_000) + self._send(EventTypes.Redaction, {}, redacts=target) + self._assert_twins_match() + def test_rejected(self) -> None: def rejected(twin: _Twin) -> list[EventPersistencePair]: event, context = self._build( @@ -804,6 +979,51 @@ class PersistTxnTestCase(_TwinTestBase): self._step(lambda twin: [built[twin.name]]) self._assert_twins_match() + def test_de_outlier_with_gap(self) -> None: + """An event de-outliered while its prev event is an outlier: the + prev event becomes a backward extremity, with a timeline gap. The + event's own extremity and failed pull attempts go.""" + prev = self._message("prev") + built: dict[str, EventPersistencePair] = {} + + def outlier(twin: _Twin) -> list[EventPersistencePair]: + built[twin.name] = self._build( + twin, EventTypes.Topic, {"topic": "gap"}, state_key="" + ) + return [self._outlier(built[twin.name])] + + self._step(outlier, backfilled=True) + for twin in self.twins: + # The prev event as if we had only fetched it as an outlier. + self.get_success( + self.store.db_pool.simple_update_one( + "events", {"event_id": twin.events[prev]}, {"outlier": True} + ) + ) + event_id = built[twin.name][0].event_id + for table, row in ( + ("event_backward_extremities", {}), + ( + "event_failed_pull_attempts", + {"num_attempts": 1, "last_attempt_ts": 0, "last_cause": "x"}, + ), + ): + self.get_success( + self.store.db_pool.simple_insert( + table, + {"room_id": twin.room_id, "event_id": event_id, **row}, + desc="test", + ) + ) + self._step(lambda twin: [built[twin.name]]) + self._assert_twins_match() + gaps = self.get_success( + self.store.db_pool.simple_select_onecol( + "timeline_gaps", {"room_id": self.twins[1].room_id}, "stream_ordering" + ) + ) + self.assertEqual(len(gaps), 1) + def test_duplicates_in_chunk(self) -> None: def duplicated(twin: _Twin) -> list[EventPersistencePair]: pair = self._build(twin, EventTypes.Topic, {"topic": "twice"}, state_key="") @@ -840,7 +1060,9 @@ 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.persist_events_txn_for_test( + "!nope:test", [], self.persister.last_prepared + ) class StateDagPersistTxnTestCase(PersistTxnTestCase):