Store the events themselves in the Rust persist transaction

The second commit of phase 2. The Rust transaction now runs the steps of
`_persist_events_txn` from `_update_room_depths_txn` up to
`_store_rejected_events_txn`:

- `txn/events.rs`: the room's depth and its place in the events stream
  cache, de-outliering, redactions that already apply (the event is
  stored pruned) and the earlier events the chunk's redactions confirm
  (they stop being sticky, and a passed recheck is stored), the
  `event_json`, `events` and `state_events` rows, transaction IDs, state
  groups and partial state, the chain cover index and `event_auth`, and
  rejections.
- `txn/edges.rs`: the forward and state DAG extremities, and the
  backward extremities and timeline gaps of a de-outliered event.
  `_handle_mult_prev_events` will use the latter for new events.
- `txn/json.rs`: JSON as `json_encoder` writes it, with non-ASCII and
  DEL escaped. The event and its internal metadata keep the key order
  `get_dict_for_persistence()` and `get_dict()` give, so the `event_json`
  text is Python's byte for byte. Only number text from federation can
  differ (`1.50` against Python's `1.5`).

An event with partial state in a room that is no longer partial stated
fails with `PartialStateConflictError`, as in Python.

`Persister.persist_events_txn_for_test` takes the chunk Rust prepared,
which `PreparedChunk` now shares through an `Arc`. It returns which
events were stored redacted and which were de-outliered, so the twin
test's Python half can carry on from there. Making a de-outliered event
sticky stays in Python until the sticky commit, as it needs the sticky
stream's ID generator. Python makes those sticky before deleting
redacted events' sticky rows, so the sticky commit has to insert them
inside `update_outliers`.

New twin scenarios: transaction IDs, URLs and non-ASCII text, a
redaction recheck (alone and in a chunk with its target), a partial
state conflict, a redacted sticky event, and a redaction of a later
redaction, which needs censoring again, and an event de-outliered while
its prev event is an outlier (a backward extremity and a timeline gap).
The test checks that room B's transaction takes the chunk prepared for
it. The fork scenario now advances
the clock between its topics, as state resolution otherwise breaks the
tie by event ID, which differs between the rooms. Each ID generator's
stream is ranked separately. Nineteen mutants of the new steps each
fail the comparison. `PersistenceJsonTestCase` compares the `event_json`
text byte for byte.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LXMkcZVkcZhi21augBff4a

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<serde_json::Value> {
-        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<Vec<(&'static str, serde_json::Value)>> {
+        Ok(self
+            .read_inner()?
+            .data
+            .iter()
+            .map(|entry| entry.to_json_pair())
+            .collect())
     }
 
     /// The event's stream ordering, if known.

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<Value> {
-        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<R>(
+        &self,
+        f: impl FnOnce(&PersistenceDict<'_>) -> R,
+    ) -> PyResult<R> {
+        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,

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<Number>,
     #[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::Value> {
-        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<R>(
+        &self,
+        f: impl FnOnce(&PersistedUnsignedFields) -> R,
+    ) -> PyResult<R> {
+        Ok(f(&self.py_read()?.persisted_fields))
     }
 }
 

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<Event>, PersistContext)>,
-    ) -> PyResult<(PersistEffects, Vec<usize>)> {
+        prepared: &PreparedChunk,
+    ) -> PyResult<(
+        PersistEffects,
+        Vec<usize>,
+        HashMap<usize, String>,
+        Vec<usize>,
+    )> {
+        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::<anyhow::Result<Vec<_>>>()
             .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<Prepared>,
     /// What preparing counted, until it is recorded.
     counts: Mutex<Option<PersistCounts>>,
 }
 
 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::<PartialStateConflict>().is_some() {
+        return match py_partial_state_conflict_error() {
+            Ok(err) | Err(err) => err,
+        };
+    }
     if let Some(missing) = e.downcast_ref::<MissingPrevStateEvents>() {
         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<PyErr> {
+    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

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:
+ * <https://www.gnu.org/licenses/agpl-3.0.html>.
+ *
+ */
+
+//! 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<Arc<str>>,
+    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<Arc<str>>,
+) -> 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<Vec<String>> =
+        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<i64> = 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(())
+}

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:
+ * <https://www.gnu.org/licenses/agpl-3.0.html>.
+ *
+ */
+
+//! 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<i64> = 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<i64> = 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<Persisting<'a>>,
+    effects: &mut PersistEffects,
+    de_outliered: &mut Vec<&'a TxnEvent>,
+) -> Result<Vec<Persisting<'a>>, 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<String, bool> = 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<Persisting<'a>>,
+) -> Result<Vec<Persisting<'a>>, 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<bool>>(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<String, String> = 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<HashSet<String>, 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<String, (String, String)> = 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<String>,
+) -> 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<Option<&str>> = 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<i64>)> = 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<Option<i64>> = 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<Arc<str>, 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<i64> = new_event_links.values().map(|l| l.chain_id).collect();
+        let sequence_numbers: Vec<i64> = 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<Persisting<'a>>,
+) -> Result<Vec<Persisting<'a>>, 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)
+}

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:
+ * <https://www.gnu.org/licenses/agpl-3.0.html>.
+ *
+ */
+
+//! 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<W>(&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<W>(&mut self, writer: &mut W, first: bool) -> io::Result<()>
+    where
+        W: ?Sized + io::Write,
+    {
+        CompactFormatter.begin_array_value(writer, first)
+    }
+
+    fn begin_object_key<W>(&mut self, writer: &mut W, first: bool) -> io::Result<()>
+    where
+        W: ?Sized + io::Write,
+    {
+        CompactFormatter.begin_object_key(writer, first)
+    }
+
+    fn begin_object_value<W>(&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<T: Serialize + ?Sized>(value: &T) -> serde_json::Result<String> {
+    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<K: Serialize, V: Serialize>(pairs: &[(K, V)]) -> serde_json::Result<String> {
+    struct Object<'a, K, V>(&'a [(K, V)]);
+
+    impl<K: Serialize, V: Serialize> Serialize for Object<'_, K, V> {
+        fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+            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"}"#
+        );
+    }
+}

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<Event>,
+}
+
+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<i64, WriteError> {
+        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<Persisting<'a>>,
+    /// 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<Vec<&'a TxnEvent>, WriteError> {
+) -> Result<TxnPrefix<'a>, 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<Persisting<'a>> = 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

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

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~ "\\ </',
+                        "f": [1.5, 1e5, 1e16, 1e-7, -0.0],
+                        "big": 12345678901234567890,
+                    },
+                    "depth": 5,
+                    "origin_server_ts": 1234,
+                    "prev_events": [["$prev:test", {}]],
+                    "auth_events": [],
+                    "event_id": "$event:test",
+                    "hashes": {"sha256": "abc"},
+                    "signatures": {"test": {"ed25519:a": "sig"}},
+                    "unsigned": {"age_ts": 1000},
+                },
+                RoomVersions.V1,
+                {"txn_id": "txn1", "token_id": 5, "device_id": "DEV"},
+            )
+        )
+
     def test_state_with_unsigned(self) -> None:
         # `replaces_state` and `invite_room_state` are stored; `prev_content`
         # is not.

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 == "<lambda>" 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):