Take the event stream orderings in Rust at rust_event_persistence: full

At `full`, `Persister.persist_chunk` takes the chunk's stream orderings
from the events or backfill stream tracker, which Python's
`_stream_id_gen` and `_backfill_id_gen` share. It runs `nextval` in
autocommit after prepare, as late as `get_next_mult` does, and sets each
event's stream ordering, received time and instance before the write
transaction. Backfilled events take theirs negated.

The orderings come back unfinished in `PersistEffects.stream_ids`.
`_persist_events_and_state_updates` skips `get_next_mult` when Rust takes
them. Its post-transaction block (metrics, MSC4293
`room_ban_redactions`, the extremities prefill) moves to
`_after_persisting_events`, which both paths call. At `full` it runs
after the effects are applied and before the orderings are finished, as
it ran inside `get_next_mult`. So a ban that redacts a user's events is
recorded before other workers see it.

`finish_rust_persist_stream_ids` finishes the orderings as
`get_next_mult` does on exit: it tells replication, and on success moves
`stream_positions` on. If the transaction fails, Rust finishes them. If
the apply or the follow-ups raise, Python finishes them.

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

rust/src/persist/python.rs

@@ -36,8 +36,9 @@ use super::{
     txn::{self, json, streams::TxnStream, PartialStateConflict, TxnEvent, TxnInput},
 };
 use crate::{
-    database::postgres::write::{
-        is_retryable_error, run_write_interaction, WriteError, MAX_ATTEMPTS,
+    database::postgres::{
+        pool::ConnectionPool,
+        write::{is_retryable_error, run_write_interaction, WriteError, MAX_ATTEMPTS},
     },
     deferred::{create_deferred_with, OnCancel},
     errors::SynapseError,
@@ -48,6 +49,7 @@ use crate::{
     state::{ConflictCache, StateResStats},
     storage::{
         native_pool::NativePool,
+        pool_querier::PoolQuerier,
         store::Store,
         stream_ids::{PyStreamIdTracker, StreamIdTracker},
         sync_store::{SyncStore, SyncStoreConfig},
@@ -170,30 +172,55 @@ impl Persister {
 
     /// Convert the chunk while the GIL is held, and return the future that
     /// runs the write transaction over it without the GIL. See
-    /// [`Self::persist_events_txn`].
+    /// [`Self::persist_events_txn`] and [`Self::persist_chunk`].
     fn start_persist_events_txn(
         &self,
         py: Python<'_>,
         room_id: String,
         events_and_contexts: Vec<(Py<Event>, PersistContext)>,
         prepared: &PreparedChunk,
+        orderings: StreamOrderings,
         inhibit_local_membership_updates: bool,
     ) -> PyResult<BoxFuture<'static, PyResult<PersistEffects>>> {
         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
+        let events: Vec<(Event, PersistContext)> = 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)?;
+            .map(|(event, context)| (event.bind(py).get().shallow_clone(), context))
+            .collect();
         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");
             };
+            // Taken as late as possible, as Python takes them: until the
+            // chunk is written, no later event on the stream can be
+            // announced.
+            let event_stream = match orderings {
+                StreamOrderings::Set => None,
+                StreamOrderings::Allocate { backfilled } => {
+                    let stream = if backfilled {
+                        TxnStream::backfill(Arc::clone(&inner.backfill_stream))
+                    } else {
+                        TxnStream::events(Arc::clone(&inner.events_stream))
+                    };
+                    let ids = allocate_stream_orderings(&pool, &stream, events.len())
+                        .await
+                        .map_err(persist_error_to_py)?;
+                    set_persisted_positions(&inner, &events, &ids, backfilled)
+                        .map_err(persist_error_to_py)?;
+                    Some((stream, ids, backfilled))
+                }
+            };
+            let events = events
+                .into_iter()
+                .map(|(event, context)| TxnEvent::new(event, context))
+                .collect::<anyhow::Result<Vec<_>>>()
+                .map_err(persist_error_to_py)?;
+
             let cache_stream = TxnStream::caches(Arc::clone(&inner.cache_stream));
             let sticky_stream = TxnStream::sticky_events(Arc::clone(&inner.sticky_stream));
             let profile_stream = TxnStream::profile_updates(Arc::clone(&inner.profile_stream));
@@ -213,7 +240,7 @@ impl Persister {
                 sticky_stream: &sticky_stream,
                 profile_stream: &profile_stream,
             };
-            let effects = run_write_interaction(&pool, "persist_events", |txn, _| {
+            let mut effects = run_write_interaction(&pool, "persist_events", |txn, _| {
                 let input = &input;
                 async move {
                     let mut effects = PersistEffects::default();
@@ -227,6 +254,12 @@ impl Persister {
             cache_stream.hand_over(effects.cache_stream_ids());
             sticky_stream.hand_over(effects.sticky_stream_ids());
             profile_stream.hand_over(effects.profile_stream_ids());
+            // Finished on drop if the transaction failed, as `get_next_mult`
+            // finishes them when its block raises.
+            if let Some((stream, ids, backfilled)) = event_stream {
+                stream.hand_over(&ids);
+                effects.add_stream_ids(ids, backfilled);
+            }
             Ok(effects)
         };
         Ok(future.boxed())
@@ -372,11 +405,61 @@ impl Persister {
             room_id,
             events_and_contexts,
             prepared,
+            StreamOrderings::Set,
             inhibit_local_membership_updates,
         )?;
         create_deferred_with(py, &self.inner.runtime, future, OnCancel::RunOn)
     }
 
+    /// As [`Self::persist_events_txn`], but Rust takes the chunk's stream
+    /// orderings, from the backfill stream if `backfilled`, and sets them on
+    /// the events with the time and this instance, as
+    /// `_persist_events_and_state_updates` does. They come back in the
+    /// effects' `stream_ids`, unfinished. If the persist fails, Rust
+    /// finishes them.
+    #[pyo3(signature = (room_id, events_and_contexts, prepared, *, backfilled, inhibit_local_membership_updates))]
+    fn persist_chunk<'py>(
+        &self,
+        py: Python<'py>,
+        room_id: String,
+        events_and_contexts: Vec<(Py<Event>, PersistContext)>,
+        prepared: &PreparedChunk,
+        backfilled: bool,
+        inhibit_local_membership_updates: bool,
+    ) -> PyResult<Bound<'py, PyAny>> {
+        let future = self.start_persist_events_txn(
+            py,
+            room_id,
+            events_and_contexts,
+            prepared,
+            StreamOrderings::Allocate { backfilled },
+            inhibit_local_membership_updates,
+        )?;
+        create_deferred_with(py, &self.inner.runtime, future, OnCancel::RunOn)
+    }
+
+    /// As [`Self::persist_chunk`], but blocks.
+    #[pyo3(signature = (room_id, events_and_contexts, prepared, *, backfilled, inhibit_local_membership_updates))]
+    fn persist_chunk_blocking(
+        &self,
+        py: Python<'_>,
+        room_id: String,
+        events_and_contexts: Vec<(Py<Event>, PersistContext)>,
+        prepared: &PreparedChunk,
+        backfilled: bool,
+        inhibit_local_membership_updates: bool,
+    ) -> PyResult<PersistEffects> {
+        let future = self.start_persist_events_txn(
+            py,
+            room_id,
+            events_and_contexts,
+            prepared,
+            StreamOrderings::Allocate { backfilled },
+            inhibit_local_membership_updates,
+        )?;
+        self.block_on(py, future)
+    }
+
     /// As [`Self::persist_events_txn`], but blocks.
     #[pyo3(signature = (room_id, events_and_contexts, prepared, *, inhibit_local_membership_updates))]
     fn persist_events_txn_blocking(
@@ -392,6 +475,7 @@ impl Persister {
             room_id,
             events_and_contexts,
             prepared,
+            StreamOrderings::Set,
             inhibit_local_membership_updates,
         )?;
         self.block_on(py, future)
@@ -413,6 +497,69 @@ impl Persister {
     }
 }
 
+/// Where the stream orderings of a chunk come from.
+#[derive(Clone, Copy)]
+enum StreamOrderings {
+    /// Python set them on the events.
+    Set,
+    /// Rust takes them from the events stream, or from the backfill stream
+    /// if the chunk is backfilled.
+    Allocate { backfilled: bool },
+}
+
+/// `n` IDs from `stream`, unfinished, retried as Python's `runInteraction`
+/// retries the fetch.
+async fn allocate_stream_orderings(
+    pool: &ConnectionPool,
+    stream: &TxnStream,
+    n: usize,
+) -> anyhow::Result<Vec<i64>> {
+    let mut attempt = 1;
+    loop {
+        let result = async {
+            let querier = PoolQuerier::checkout(pool).await?;
+            stream.next_ids_outside_txn(&querier, n).await
+        }
+        .await;
+        match result {
+            Err(e) if attempt < MAX_ATTEMPTS && is_retryable_error(&e) => {
+                log::warn!("[TXN RETRY] {{_load_next_mult_id}} {e:#} {attempt}/{MAX_ATTEMPTS}");
+                attempt += 1;
+            }
+            result => return result,
+        }
+    }
+}
+
+/// Set each event's stream ordering from `ids`, in order, with the time
+/// and this instance, as `_persist_events_and_state_updates` does once it
+/// has its stream orderings. Backfilled events take theirs negated. An
+/// event that appears twice keeps the later ordering, as in Python.
+fn set_persisted_positions(
+    inner: &PersisterInner,
+    events: &[(Event, PersistContext)],
+    ids: &[i64],
+    backfilled: bool,
+) -> anyhow::Result<()> {
+    if ids.len() != events.len() {
+        anyhow::bail!(
+            "took {} stream orderings for {} events",
+            ids.len(),
+            events.len()
+        );
+    }
+    let received_ts = i64::try_from(inner.runtime.clock().now_millis())?;
+    for ((event, _), &id) in events.iter().zip(ids) {
+        let stream_ordering = if backfilled { -id } else { id };
+        event.internal_metadata().set_persisted_position(
+            Some(stream_ordering),
+            Some(received_ts),
+            Some(inner.instance_name.clone()),
+        )?;
+    }
+    Ok(())
+}
+
 /// What [`Persister::prepare_chunk`] computed, in the shapes Python builds
 /// its own types from. If `state_res_fallback` is set, the rest is empty
 /// and Python should compute the chunk itself.

rust/src/persist/txn/streams.rs

@@ -13,9 +13,11 @@
  *
  */
 
-//! The streams a persist takes IDs from inside its transaction, as
-//! Python's `get_next_txn` and `get_next_mult_txn` do: the cache stream,
-//! the sticky events stream and the profile updates stream.
+//! The streams a persist takes IDs from: inside its transaction, as
+//! Python's `get_next_txn` and `get_next_mult_txn` do, the cache stream,
+//! the sticky events stream and the profile updates stream. Before its
+//! transaction, as Python's `get_next_mult` does, the events or the
+//! backfill stream, for the chunk's stream orderings.
 //!
 //! The IDs stay unfinished until the transaction has committed and Python
 //! has applied its effects, as Python finishes them only after the
@@ -24,12 +26,16 @@
 use std::sync::{Arc, Mutex};
 
 use anyhow::anyhow;
+use tokio_postgres::Row;
 
 use crate::{
     database::postgres::write::{WriteError, WriteTxn},
-    storage::stream_ids::StreamIdTracker,
+    storage::{events::Querier, stream_ids::StreamIdTracker},
 };
 
+/// `_load_next_mult_id_txn`'s statement: `$2` IDs from the sequence `$1`.
+const NEXTVAL_SQL: &str = "SELECT nextval($1::text) FROM generate_series(1, $2::bigint)";
+
 /// A stream's IDs taken by one persist, over all its attempts. An ID the
 /// committed attempt took is handed to Python with the effects. The rest
 /// are marked finished here: when an attempt fails, at the start of the
@@ -66,6 +72,17 @@ impl TxnStream {
         Self::new(tracker, "profile_updates_sequence")
     }
 
+    /// The events stream, `_stream_id_gen`.
+    pub fn events(tracker: Arc<StreamIdTracker>) -> Self {
+        Self::new(tracker, "events_stream_seq")
+    }
+
+    /// The backfill stream, `_backfill_id_gen`. Its IDs are positive, as
+    /// the tracker holds them. The events take them negated.
+    pub fn backfill(tracker: Arc<StreamIdTracker>) -> Self {
+        Self::new(tracker, "events_backfill_stream_seq")
+    }
+
     fn lock_pending(&self) -> std::sync::MutexGuard<'_, Vec<i64>> {
         self.pending.lock().unwrap_or_else(|e| e.into_inner())
     }
@@ -81,21 +98,23 @@ impl TxnStream {
     /// `_load_next_mult_id_txn`: the next `n` IDs on the stream,
     /// unfinished.
     pub async fn next_ids(&self, txn: &WriteTxn, n: usize) -> Result<Vec<i64>, WriteError> {
-        let fetch = Fetch::begin(self);
         let n = i64::try_from(n).map_err(anyhow::Error::new)?;
-        let rows = txn
-            .query(
-                "SELECT nextval($1::text) FROM generate_series(1, $2::bigint)",
-                &[&self.sequence, &n],
-            )
-            .await?;
-        let ids = rows
-            .iter()
-            .map(|row| row.try_get(0))
-            .collect::<Result<Vec<i64>, _>>()
-            .map_err(anyhow::Error::new)?;
-        fetch.allocated(&ids);
-        Ok(ids)
+        let fetch = Fetch::begin(self);
+        let rows = txn.query(NEXTVAL_SQL, &[&self.sequence, &n]).await?;
+        Ok(fetch.allocated_rows(&rows)?)
+    }
+
+    /// As [`Self::next_ids`], but outside a transaction, as `get_next_mult`
+    /// fetches its IDs in autocommit. The sequence ignores transactions.
+    pub async fn next_ids_outside_txn(
+        &self,
+        querier: &impl Querier,
+        n: usize,
+    ) -> anyhow::Result<Vec<i64>> {
+        let n = i64::try_from(n)?;
+        let fetch = Fetch::begin(self);
+        let rows = querier.query(NEXTVAL_SQL, &[&self.sequence, &n]).await?;
+        fetch.allocated_rows(&rows)
     }
 
     /// Mark finished the IDs of an attempt that failed, as Python's
@@ -131,6 +150,17 @@ impl<'a> Fetch<'a> {
         }
     }
 
+    /// The fetch returned `rows` of IDs, which stay unfinished. Ends the
+    /// fetch without allocating if a row cannot be read.
+    fn allocated_rows(self, rows: &[Row]) -> anyhow::Result<Vec<i64>> {
+        let ids = rows
+            .iter()
+            .map(|row| row.try_get(0))
+            .collect::<Result<Vec<i64>, _>>()?;
+        self.allocated(&ids);
+        Ok(ids)
+    }
+
     /// The fetch allocated `ids`, which stay unfinished.
     fn allocated(mut self, ids: &[i64]) {
         if let Some(token) = self.token.take() {

synapse/config/experimental.py

@@ -415,8 +415,8 @@ class ExperimentalConfig(Config):
         #  - "prepare": Rust's results for that are used.
         #  - "transaction": Rust also runs the write transaction.
         #  - "full": Rust persists each queued batch.
-        # The levels are being built up in that order. So far "transaction"
-        # and "full" act as "prepare".
+        # The levels are being built up in that order. So far "full" only
+        # has Rust take the stream orderings of each chunk as well.
         rust_event_persistence = experimental.get("rust_event_persistence", "off")
         # YAML reads an unquoted `off` as a boolean.
         if rust_event_persistence is False:

synapse/storage/controllers/persist_events.py

@@ -928,16 +928,22 @@ class EventsPersistenceStorageController:
 
     def _rust_transaction(self, prepared: _PreparedChunk) -> RustTransaction | None:
         """Whether to write the chunk in Rust: at `rust_event_persistence:
-        transaction`, if Rust prepared it."""
+        transaction`, if Rust prepared it. At `full`, Rust takes its stream
+        orderings too."""
+        experimental = self.hs.config.experimental
         if (
             prepared.rust is None
             or self._rust_persister is None
-            or not self.hs.config.experimental.rust_event_persistence_at_least(
-                "transaction"
-            )
+            or not experimental.rust_event_persistence_at_least("transaction")
         ):
             return None
-        return RustTransaction(self._rust_persister, prepared.rust)
+        return RustTransaction(
+            self._rust_persister,
+            prepared.rust,
+            allocate_stream_orderings=experimental.rust_event_persistence_at_least(
+                "full"
+            ),
+        )
 
     async def _prepare_chunk(
         self, room_id: str, chunk: list[EventPersistencePair], backfilled: bool

synapse/storage/databases/main/events.py

@@ -68,7 +68,10 @@ from synapse.storage.database import (
     make_tuple_in_list_sql_clause,
 )
 from synapse.storage.databases.main.event_federation import EventFederationStore
-from synapse.storage.databases.main.rust_persist import apply_rust_persist_effects
+from synapse.storage.databases.main.rust_persist import (
+    apply_rust_persist_effects,
+    finish_rust_persist_stream_ids,
+)
 from synapse.storage.databases.main.search import SearchEntry
 from synapse.storage.engines import PostgresEngine
 from synapse.storage.util.id_generators import AbstractStreamIdGenerator
@@ -253,6 +256,8 @@ class RustTransaction:
 
     persister: "Persister"
     prepared: "PreparedChunk"
+    # Rust takes the chunk's stream orderings too (`full`).
+    allocate_stream_orderings: bool = False
 
 
 @attr.s(slots=True, auto_attribs=True)
@@ -405,7 +410,8 @@ class PersistEventsStore:
             table_changes: The sliding sync table changes and sticky events
                 to un-soft-fail, if the caller computed them. Otherwise they
                 are computed here.
-            rust_transaction: Write the chunk in Rust, which prepared it.
+            rust_transaction: Write the chunk in Rust, which prepared it, and
+                which may take its stream orderings too.
 
         Returns:
             Resolves when the events have been persisted
@@ -415,6 +421,18 @@ class PersistEventsStore:
                 a room that has been un-partial stated.
         """
 
+        if rust_transaction is not None and rust_transaction.allocate_stream_orderings:
+            # Rust takes the stream orderings itself, as late as it can.
+            await self._persist_events_in_rust(
+                rust_transaction,
+                room_id,
+                events_and_contexts,
+                backfilled=use_negative_stream_ordering,
+                inhibit_local_membership_updates=inhibit_local_membership_updates,
+                new_forward_extremities=new_forward_extremities,
+            )
+            return
+
         # We want to calculate the stream orderings as late as possible, as
         # we only notify after all events with a lesser stream ordering have
         # been persisted. I.e. if we spend 10s inside the with block then
@@ -465,6 +483,7 @@ class PersistEventsStore:
                     rust_transaction,
                     room_id,
                     events_and_contexts,
+                    backfilled=use_negative_stream_ordering,
                     inhibit_local_membership_updates=inhibit_local_membership_updates,
                 )
             else:
@@ -481,116 +500,135 @@ class PersistEventsStore:
                     new_state_dag_forward_extremities=new_state_dag_forward_extremities,
                     sticky_events_to_un_soft_fail=sticky_events_to_un_soft_fail,
                 )
-            persist_event_counter.labels(**{SERVER_NAME_LABEL: self.server_name}).inc(
-                len(events_and_contexts)
+            await self._after_persisting_events(
+                room_id,
+                events_and_contexts,
+                use_negative_stream_ordering=use_negative_stream_ordering,
+                new_forward_extremities=new_forward_extremities,
             )
 
-            if not use_negative_stream_ordering:
-                # we don't want to set the event_persisted_position to a negative
-                # stream_ordering.
-                synapse.metrics.event_persisted_position.labels(
-                    **{SERVER_NAME_LABEL: self.server_name}
-                ).set(stream)
+    async def _after_persisting_events(
+        self,
+        room_id: str,
+        events_and_contexts: list[EventPersistencePair],
+        *,
+        use_negative_stream_ordering: bool,
+        new_forward_extremities: set[str] | None,
+    ) -> None:
+        """What follows writing a chunk: the metrics, the MSC4293
+        `room_ban_redactions` updates and the forward extremities prefill."""
+        # The chunk's last stream ordering, which its last event took.
+        last_stream_ordering = events_and_contexts[-1][
+            0
+        ].internal_metadata.stream_ordering
+        assert last_stream_ordering is not None
+        persist_event_counter.labels(**{SERVER_NAME_LABEL: self.server_name}).inc(
+            len(events_and_contexts)
+        )
 
-            for event, context in events_and_contexts:
-                if context.app_service:
-                    origin_type = "application_service"
-                elif self.hs.is_mine_id(event.sender):
-                    origin_type = "local"
-                else:
-                    origin_type = "remote"
+        if not use_negative_stream_ordering:
+            # we don't want to set the event_persisted_position to a negative
+            # stream_ordering.
+            synapse.metrics.event_persisted_position.labels(
+                **{SERVER_NAME_LABEL: self.server_name}
+            ).set(last_stream_ordering)
 
-                # We only track a subset of event types, to avoid high
-                # cardinality in the metrics.
-                metrics_event_type = (
-                    event.type if event.type in TRACKED_EVENT_TYPES else "*other*"
-                )
+        for event, context in events_and_contexts:
+            if context.app_service:
+                origin_type = "application_service"
+            elif self.hs.is_mine_id(event.sender):
+                origin_type = "local"
+            else:
+                origin_type = "remote"
 
-                event_counter.labels(
-                    type=metrics_event_type,
-                    origin_type=origin_type,
-                    **{SERVER_NAME_LABEL: self.server_name},
-                ).inc()
+            # We only track a subset of event types, to avoid high
+            # cardinality in the metrics.
+            metrics_event_type = (
+                event.type if event.type in TRACKED_EVENT_TYPES else "*other*"
+            )
 
+            event_counter.labels(
+                type=metrics_event_type,
+                origin_type=origin_type,
+                **{SERVER_NAME_LABEL: self.server_name},
+            ).inc()
+
+            if (
+                not self.hs.config.experimental.msc4293_enabled
+                or event.type != EventTypes.Member
+                or event.state_key is None
+            ):
+                continue
+
+            # check if this is an unban/join that will undo a ban/kick redaction for
+            # a user in the room
+            if event.membership in [Membership.LEAVE, Membership.JOIN]:
                 if (
-                    not self.hs.config.experimental.msc4293_enabled
-                    or event.type != EventTypes.Member
-                    or event.state_key is None
-                ):
-                    continue
-
-                # check if this is an unban/join that will undo a ban/kick redaction for
-                # a user in the room
-                if event.membership in [Membership.LEAVE, Membership.JOIN]:
-                    if (
-                        event.membership == Membership.LEAVE
-                        and event.sender == event.state_key
-                    ):
-                        # self-leave, ignore
-                        continue
-
-                    # if there is an existing ban/leave causing redactions for
-                    # this user/room combination update the entry with the stream
-                    # ordering when the redactions should stop - in the case of a backfilled
-                    # event where the stream ordering is negative, use the current max stream
-                    # ordering
-                    stream_ordering = event.internal_metadata.stream_ordering
-                    assert stream_ordering is not None
-                    if stream_ordering < 0:
-                        stream_ordering = self._stream_id_gen.get_current_token()
-                    await self.db_pool.simple_update(
-                        "room_ban_redactions",
-                        {"room_id": event.room_id, "user_id": event.state_key},
-                        {"redact_end_ordering": stream_ordering},
-                        desc="room_ban_redactions update redact_end_ordering",
-                    )
-
-                # check for msc4293 redact_events flag and apply if found
-                if event.membership not in [Membership.LEAVE, Membership.BAN]:
-                    continue
-                redact = event.content.get("org.matrix.msc4293.redact_events", False)
-                if not redact or not isinstance(redact, bool):
-                    continue
-                # self-bans currently are not authorized so we don't check for that
-                # case
-                if (
-                    event.membership == Membership.BAN
+                    event.membership == Membership.LEAVE
                     and event.sender == event.state_key
                 ):
+                    # self-leave, ignore
                     continue
 
-                # check that sender can redact
-                redact_allowed = await self._can_sender_redact(event)
-
-                # Signal that this user's past events in this room
-                # should be redacted by adding an entry to
-                # `room_ban_redactions`.
-                if redact_allowed:
-                    await self.db_pool.simple_upsert(
-                        "room_ban_redactions",
-                        {"room_id": event.room_id, "user_id": event.state_key},
-                        {
-                            "redacting_event_id": event.event_id,
-                            "redact_end_ordering": None,
-                        },
-                        {
-                            "room_id": event.room_id,
-                            "user_id": event.state_key,
-                            "redacting_event_id": event.event_id,
-                            "redact_end_ordering": None,
-                        },
-                    )
-
-                    # normally the cache entry for a redacted event would be invalidated
-                    # by an arriving redaction event, but since we are not creating redaction
-                    # events we invalidate manually
-                    self.store._invalidate_local_get_event_cache_room_id(event.room_id)
-
-            if new_forward_extremities:
-                self.store.get_latest_event_ids_in_room.prefill(
-                    (room_id,), frozenset(new_forward_extremities)
+                # if there is an existing ban/leave causing redactions for
+                # this user/room combination update the entry with the stream
+                # ordering when the redactions should stop - in the case of a backfilled
+                # event where the stream ordering is negative, use the current max stream
+                # ordering
+                stream_ordering = event.internal_metadata.stream_ordering
+                assert stream_ordering is not None
+                if stream_ordering < 0:
+                    stream_ordering = self._stream_id_gen.get_current_token()
+                await self.db_pool.simple_update(
+                    "room_ban_redactions",
+                    {"room_id": event.room_id, "user_id": event.state_key},
+                    {"redact_end_ordering": stream_ordering},
+                    desc="room_ban_redactions update redact_end_ordering",
                 )
 
+            # check for msc4293 redact_events flag and apply if found
+            if event.membership not in [Membership.LEAVE, Membership.BAN]:
+                continue
+            redact = event.content.get("org.matrix.msc4293.redact_events", False)
+            if not redact or not isinstance(redact, bool):
+                continue
+            # self-bans currently are not authorized so we don't check for that
+            # case
+            if event.membership == Membership.BAN and event.sender == event.state_key:
+                continue
+
+            # check that sender can redact
+            redact_allowed = await self._can_sender_redact(event)
+
+            # Signal that this user's past events in this room
+            # should be redacted by adding an entry to
+            # `room_ban_redactions`.
+            if redact_allowed:
+                await self.db_pool.simple_upsert(
+                    "room_ban_redactions",
+                    {"room_id": event.room_id, "user_id": event.state_key},
+                    {
+                        "redacting_event_id": event.event_id,
+                        "redact_end_ordering": None,
+                    },
+                    {
+                        "room_id": event.room_id,
+                        "user_id": event.state_key,
+                        "redacting_event_id": event.event_id,
+                        "redact_end_ordering": None,
+                    },
+                )
+
+                # normally the cache entry for a redacted event would be invalidated
+                # by an arriving redaction event, but since we are not creating redaction
+                # events we invalidate manually
+                self.store._invalidate_local_get_event_cache_room_id(event.room_id)
+
+        if new_forward_extremities:
+            self.store.get_latest_event_ids_in_room.prefill(
+                (room_id,), frozenset(new_forward_extremities)
+            )
+
     async def _can_sender_redact(self, event: EventBase) -> bool:
         state_filter = StateFilter.from_types(
             [(EventTypes.PowerLevels, ""), (EventTypes.Create, "")]
@@ -1137,11 +1175,16 @@ class PersistEventsStore:
         room_id: str,
         events_and_contexts: list[EventPersistencePair],
         *,
+        backfilled: bool,
         inhibit_local_membership_updates: bool,
+        new_forward_extremities: set[str] | None = None,
     ) -> None:
         """Write the chunk in Rust, as `_persist_events_txn` does, then apply
         the cache updates its transaction asks for. Called with the chunk's
-        stream orderings allocated, which are finished after.
+        stream orderings allocated, which are finished after, unless
+        `rust_transaction` takes them itself, from the backfill stream if
+        `backfilled`. Then this also runs `_after_persisting_events`, with
+        `new_forward_extremities`, before it finishes them.
 
         Raises:
             PartialStateConflictError: as `_persist_events_txn`.
@@ -1152,22 +1195,61 @@ class PersistEventsStore:
         # caller waits for both, as Python's transaction cannot be cancelled.
 
         async def persist_and_apply() -> None:
-            effects = await make_deferred_yieldable(
-                rust_transaction.persister.persist_events_txn(
-                    room_id,
-                    events_and_contexts,
-                    rust_transaction.prepared,
-                    inhibit_local_membership_updates=inhibit_local_membership_updates,
+            persister = rust_transaction.persister
+            allocate = rust_transaction.allocate_stream_orderings
+            if allocate:
+                try:
+                    effects = await make_deferred_yieldable(
+                        persister.persist_chunk(
+                            room_id,
+                            events_and_contexts,
+                            rust_transaction.prepared,
+                            backfilled=backfilled,
+                            inhibit_local_membership_updates=inhibit_local_membership_updates,
+                        )
+                    )
+                except BaseException:
+                    # Rust finished the stream orderings it took. Tell
+                    # replication, as `get_next_mult` does when its block
+                    # raises.
+                    self.hs.get_replication_notifier().notify_replication()
+                    raise
+            else:
+                effects = await make_deferred_yieldable(
+                    persister.persist_events_txn(
+                        room_id,
+                        events_and_contexts,
+                        rust_transaction.prepared,
+                        inhibit_local_membership_updates=inhibit_local_membership_updates,
+                    )
                 )
-            )
             # The copy of a duplicated event that Rust wrote, as Python's
             # `_add_to_cache` reads the deduplicated chunk.
             written = self._filter_events_and_contexts_for_duplicates(
                 events_and_contexts
             )
-            apply_rust_persist_effects(
-                self.store, effects, {event.event_id: event for event, _ in written}
-            )
+            events = {event.event_id: event for event, _ in written}
+            if not allocate:
+                apply_rust_persist_effects(self.store, effects, events)
+                return
+            # What follows the write runs before the stream orderings are
+            # finished, as it runs inside `get_next_mult` in Python: the
+            # events must not be announced before, say, the MSC4293
+            # redactions they start are recorded.
+            try:
+                apply_rust_persist_effects(
+                    self.store, effects, events, finish_stream_ids=False
+                )
+                await self._after_persisting_events(
+                    room_id,
+                    events_and_contexts,
+                    use_negative_stream_ordering=backfilled,
+                    new_forward_extremities=new_forward_extremities,
+                )
+            except BaseException:
+                finish_rust_persist_stream_ids(self.store, effects, succeeded=False)
+                raise
+            finish_rust_persist_stream_ids(self.store, effects, succeeded=True)
 
         await delay_cancellation(persist_and_apply())
 

synapse/storage/databases/main/rust_persist.py

@@ -30,6 +30,8 @@ def apply_rust_persist_effects(
     store: "DataStore",
     effects: PersistEffects,
     events: Mapping[str, "EventBase"] | None = None,
+    *,
+    finish_stream_ids: bool = True,
 ) -> None:
     """Apply what a Rust persist transaction asks for once it has
     committed: the cache updates the Python transaction queues with
@@ -45,13 +47,20 @@ def apply_rust_persist_effects(
     transaction's earlier callbacks. The events stream IDs are finished last, as Python's
     `get_next_mult` does once the transaction's callbacks have run, so
     the current token never passes an event before the stream change
-    caches know about it.
+    caches know about it. Then `stream_positions` moves on, as
+    `get_next_mult` moves it on success.
 
-    If a step raises, the IDs of those three streams are still
-    finished, as `call_on_exception` finishes them in Python, so the
+    If a step raises, the IDs of all four streams are still finished, as
+    `call_on_exception` and `get_next_mult` finish them in Python, so the
     streams do not stall.
+
+    Without `finish_stream_ids`, the events stream IDs are left to the
+    caller, who must call `finish_rust_persist_stream_ids` whatever
+    happens.
     """
     cache_ids_finished = profile_ids_finished = sticky_ids_finished = False
+    # Left unfinished on purpose, without `finish_stream_ids`.
+    event_ids_finished = not finish_stream_ids
     try:
         for (
             stream_ordering,
@@ -120,12 +129,9 @@ def apply_rust_persist_effects(
                 (room_id,), frozenset(extremities)
             )
 
-        if effects.stream_ids:
-            id_gen = (
-                store._backfill_id_gen if effects.backfilled else store._stream_id_gen
-            )
-            id_gen._mark_ids_as_finished(effects.stream_ids)
-            store.hs.get_replication_notifier().notify_replication()
+        if not event_ids_finished:
+            finish_rust_persist_stream_ids(store, effects, succeeded=True)
+            event_ids_finished = True
         if effects.new_replication_data:
             store.hs.get_notifier().on_new_replication_data()
     except BaseException:
@@ -138,9 +144,27 @@ def apply_rust_persist_effects(
             )
         if effects.sticky_stream_ids and not sticky_ids_finished:
             store._sticky_events_id_gen._mark_ids_as_finished(effects.sticky_stream_ids)
+        if not event_ids_finished:
+            finish_rust_persist_stream_ids(store, effects, succeeded=False)
         raise
 
 
+def finish_rust_persist_stream_ids(
+    store: "DataStore", effects: PersistEffects, *, succeeded: bool
+) -> None:
+    """Finish the events or backfill stream IDs a Rust persist took, as
+    `get_next_mult` finishes them when its block exits, and tell
+    replication. Only on success does `stream_positions` move on."""
+    if not effects.stream_ids:
+        return
+    id_gen = store._backfill_id_gen if effects.backfilled else store._stream_id_gen
+    if succeeded:
+        _finish_txn_stream_ids(store, id_gen, effects.stream_ids)
+    else:
+        id_gen._mark_ids_as_finished(effects.stream_ids)
+        store.hs.get_replication_notifier().notify_replication()
+
+
 def _finish_txn_stream_ids(
     store: "DataStore", id_gen: MultiWriterIdGenerator, stream_ids: list[int]
 ) -> None:

synapse/synapse_rust/persist.pyi

@@ -80,6 +80,31 @@ class Persister:
         inhibit_local_membership_updates: bool,
     ) -> PersistEffects:
         """As `persist_events_txn`, but blocks. For tests."""
+    def persist_chunk(
+        self,
+        room_id: str,
+        events_and_contexts: list[tuple[EventBase, EventContext]],
+        prepared: PreparedChunk,
+        *,
+        backfilled: bool,
+        inhibit_local_membership_updates: bool,
+    ) -> Deferred[PersistEffects]:
+        """As `persist_events_txn`, but Rust takes the chunk's stream
+        orderings, from the backfill stream if `backfilled`, and sets them on
+        the events with the time and this instance, as
+        `_persist_events_and_state_updates` does. They come back in the
+        effects' `stream_ids`, unfinished. If the persist fails, Rust
+        finishes them."""
+    def persist_chunk_blocking(
+        self,
+        room_id: str,
+        events_and_contexts: list[tuple[EventBase, EventContext]],
+        prepared: PreparedChunk,
+        *,
+        backfilled: bool,
+        inhibit_local_membership_updates: bool,
+    ) -> PersistEffects:
+        """As `persist_chunk`, but blocks. For tests."""
 
 class PreparedChunk:
     """What `Persister.prepare_chunk` computed. If `state_res_fallback` is

tests/storage/test_rust_persist_prereqs.py

@@ -33,6 +33,7 @@ from twisted.internet.testing import MemoryReactor
 
 import synapse.metrics
 from synapse.api.constants import EventTypes, Membership, RelationTypes
+from synapse.api.errors import PartialStateConflictError
 from synapse.api.room_versions import RoomVersions
 from synapse.config import ConfigError
 from synapse.config.experimental import ExperimentalConfig
@@ -55,6 +56,7 @@ from synapse.synapse_rust.persist import (
     persist_gil_stats,
     persistence_json_for_test,
 )
+from synapse.types import create_requester
 from synapse.util.caches.deferred_cache import DeferredCache
 from synapse.util.caches.stream_change_cache import StreamChangeCache
 from synapse.util.clock import Clock
@@ -867,6 +869,8 @@ class _CountingPersister:
         self._persister = persister
         self.prepares = 0
         self.transactions = 0
+        # Chunks written with Rust taking their stream orderings.
+        self.chunks = 0
         self.last_effects: PersistEffects | None = None
 
     def prepare_chunk(self, *args: Any, **kwargs: Any) -> Any:
@@ -882,6 +886,15 @@ class _CountingPersister:
 
         return self._persister.persist_events_txn(*args, **kwargs).addCallback(keep)
 
+    def persist_chunk(self, *args: Any, **kwargs: Any) -> Any:
+        self.chunks += 1
+
+        def keep(result: Any) -> Any:
+            self.last_effects = result
+            return result
+
+        return self._persister.persist_chunk(*args, **kwargs).addCallback(keep)
+
     def __getattr__(self, name: str) -> Any:
         return getattr(self._persister, name)
 
@@ -1074,9 +1087,9 @@ class RustTransactionTestCase(unittest.HomeserverTestCase):
         )
 
 
-class RustTransactionWorkerTestCase(BaseMultiWorkerStreamTestCase):
-    """`rust_event_persistence: transaction` on a worker event persister:
-    the worker writes in Rust, to the main database."""
+class RustFullTestCase(unittest.HomeserverTestCase):
+    """`rust_event_persistence: full`: Rust also takes the chunks' stream
+    orderings."""
 
     servlets = [
         admin.register_servlets,
@@ -1087,7 +1100,345 @@ class RustTransactionWorkerTestCase(BaseMultiWorkerStreamTestCase):
     def default_config(self) -> dict[str, Any]:
         config = super().default_config()
         config.setdefault("experimental_features", {})["rust_event_persistence"] = (
-            "transaction"
+            "full"
+        )
+        return config
+
+    def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
+        if RUST_PERSIST_FOR_TESTS not in (None, "full"):
+            raise SkipTest("SYNAPSE_TEST_RUST_PERSIST overrides the level")
+        hs = self.setup_test_homeserver()
+        if not isinstance(
+            hs.get_datastores().main.db_pool._db_pool, RustConnectionPool
+        ):
+            raise SkipTest("needs the native Rust Postgres driver")
+        return hs
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        self.store = hs.get_datastores().main
+        persistence = hs.get_storage_controllers().persistence
+        assert persistence is not None
+        self.persistence = persistence
+        self.persister = _CountingPersister(persistence._rust_persister)
+        persistence._rust_persister = self.persister  # type: ignore[assignment]
+        self.user = self.register_user("alice", "pass")
+        self.token = self.login("alice", "pass")
+        self.room_id = self.helper.create_room_as(self.user, tok=self.token)
+
+    def _build(self, body: str) -> tuple[EventBase, EventContext]:
+        event, unpersisted_context = self.get_success(
+            self.hs.get_event_creation_handler().create_event(
+                create_requester(self.user),
+                {
+                    "type": EventTypes.Message,
+                    "content": {"msgtype": "m.text", "body": body},
+                    "room_id": self.room_id,
+                    "sender": self.user,
+                },
+            )
+        )
+        context = self.get_success(unpersisted_context.persist(event))
+        return event, context
+
+    def _stream_position(self, stream_name: str) -> int | None:
+        rows = self.get_success(
+            self.store.db_pool.simple_select_onecol(
+                "stream_positions",
+                {"stream_name": stream_name, "instance_name": "master"},
+                "stream_id",
+            )
+        )
+        return rows[0] if rows else None
+
+    def _event_row(self, event_id: str) -> tuple[int, str, int]:
+        """`(stream_ordering, instance_name, received_ts)` of the event's
+        `events` row."""
+        row = self.get_success(
+            self.store.db_pool.simple_select_one(
+                "events",
+                {"event_id": event_id},
+                ("stream_ordering", "instance_name", "received_ts"),
+            )
+        )
+        return row[0], row[1], row[2]
+
+    def test_rust_takes_stream_orderings(self) -> None:
+        """Rust takes the chunk's stream orderings and sets them on the
+        events, and the applier finishes them and moves `stream_positions`
+        on, as `get_next_mult` does."""
+        id_gen = self.store._stream_id_gen
+        chunks = self.persister.chunks
+        with patch.object(
+            id_gen, "get_next_mult", side_effect=AssertionError("Python allocated")
+        ):
+            event_id = self.helper.send(self.room_id, "hello", tok=self.token)[
+                "event_id"
+            ]
+        self.assertEqual(self.persister.chunks, chunks + 1)
+        effects = self.persister.last_effects
+        assert effects is not None
+        self.assertFalse(effects.backfilled)
+        (stream_ordering,) = effects.stream_ids
+
+        self.assertEqual(
+            self._event_row(event_id),
+            (stream_ordering, "master", self.clock.time_msec()),
+        )
+        self.assertGreaterEqual(id_gen.get_current_token(), stream_ordering)
+        assert id_gen._writers
+        self.assertEqual(self._stream_position("events"), stream_ordering)
+
+    def test_chunk_with_duplicate(self) -> None:
+        """A chunk takes one ordering per entry, in order. An event that
+        appears twice keeps the later one, as in Python, and every ordering
+        is finished. What follows the write sees the last ordering."""
+        first = self._build("one")
+        second = self._build("two")
+        with patch.object(
+            self.store,
+            "get_latest_event_ids_in_room",
+            Mock(wraps=self.store.get_latest_event_ids_in_room),
+        ) as latest:
+            self.get_success(self.persistence.persist_events([first, second, first]))
+        effects = self.persister.last_effects
+        assert effects is not None
+        self.assertEqual(len(effects.stream_ids), 3)
+        self.assertEqual(effects.stream_ids, sorted(effects.stream_ids))
+        _, second_id, last_id = effects.stream_ids
+
+        self.assertEqual(self._event_row(first[0].event_id)[0], last_id)
+        self.assertEqual(self._event_row(second[0].event_id)[0], second_id)
+        self.assertGreaterEqual(self.store._stream_id_gen.get_current_token(), last_id)
+        # `_after_persisting_events` ran with the chunk.
+        self.assertEqual(
+            REGISTRY.get_sample_value(
+                "synapse_event_persisted_position", {SERVER_NAME_LABEL: "test"}
+            ),
+            last_id,
+        )
+        # Both were built on the same extremity.
+        latest.prefill.assert_called_with(
+            (self.room_id,), frozenset([first[0].event_id, second[0].event_id])
+        )
+
+    def test_follow_ups_run_before_orderings_finish(self) -> None:
+        """What follows the write (the MSC4293 redactions among it) runs
+        before the token passes the chunk, as inside `get_next_mult`. If it
+        raises, the orderings are still finished."""
+        persist_events = self.hs.get_datastores().persist_events
+        assert persist_events is not None
+        id_gen = self.store._stream_id_gen
+        original = persist_events._after_persisting_events
+        seen: list[tuple[int, int | None]] = []
+
+        async def after(room_id: str, events_and_contexts: list, **kwargs: Any) -> None:
+            seen.append(
+                (
+                    id_gen.get_current_token(),
+                    events_and_contexts[-1][0].internal_metadata.stream_ordering,
+                )
+            )
+            await original(room_id, events_and_contexts, **kwargs)
+
+        with patch.object(
+            persist_events, "_after_persisting_events", side_effect=after
+        ):
+            self.helper.send(self.room_id, "hello", tok=self.token)
+        ((token, stream_ordering),) = seen
+        assert stream_ordering is not None
+        self.assertLess(token, stream_ordering)
+        self.assertGreaterEqual(id_gen.get_current_token(), stream_ordering)
+
+        class AfterFailed(Exception):
+            pass
+
+        event, context = self._build("fails after")
+        with patch.object(
+            persist_events, "_after_persisting_events", side_effect=AfterFailed()
+        ):
+            self.get_failure(
+                self.persistence.persist_events([(event, context)]), AfterFailed
+            )
+        failed = event.internal_metadata.stream_ordering
+        assert failed is not None
+        self.assertGreaterEqual(id_gen.get_current_token(), failed)
+
+    def test_gil_count(self) -> None:
+        """Taking the stream orderings and setting them on the events takes
+        the GIL at no persist site."""
+        # The first persist opens the pool, at `persist_db_pool`.
+        self.helper.send(self.room_id, "warm up", tok=self.token)
+        before = persist_gil_stats()
+        chunks = self.persister.chunks
+        self.helper.send(self.room_id, "hello", tok=self.token)
+        self.assertEqual(self.persister.chunks, chunks + 1)
+        after = persist_gil_stats()
+        changed = {
+            site: after[site][0] - before.get(site, (0, 0.0))[0]
+            for site in after
+            if after[site][0] != before.get(site, (0, 0.0))[0]
+        }
+        self.assertEqual(changed, {})
+
+    def test_backfill_takes_negative_orderings(self) -> None:
+        """A backfilled chunk takes its stream orderings from the backfill
+        stream, negated on the events."""
+        id_gen = self.store._backfill_id_gen
+        event, context = self._build("old")
+        with patch.object(
+            id_gen, "get_next_mult", side_effect=AssertionError("Python allocated")
+        ):
+            self.get_success(
+                self.persistence.persist_events([(event, context)], backfilled=True)
+            )
+        effects = self.persister.last_effects
+        assert effects is not None
+        self.assertTrue(effects.backfilled)
+        (backfill_id,) = effects.stream_ids
+        self.assertGreater(backfill_id, 0)
+        self.assertEqual(event.internal_metadata.stream_ordering, -backfill_id)
+        self.assertEqual(
+            self._event_row(event.event_id),
+            (-backfill_id, "master", self.clock.time_msec()),
+        )
+        # The backfill token is negated too.
+        self.assertLessEqual(id_gen.get_current_token(), -backfill_id)
+        assert id_gen._writers
+        self.assertEqual(self._stream_position("backfill"), -backfill_id)
+
+    def test_failed_persist_finishes_its_orderings(self) -> None:
+        """A chunk whose transaction fails leaves no stream ordering
+        unfinished, so the stream does not stall, and replication is told."""
+        event, context = self._build("partial")
+        context.partial_state = True
+        notifier = self.hs.get_replication_notifier()
+        with patch.object(
+            notifier, "notify_replication", wraps=notifier.notify_replication
+        ) as notify:
+            self.get_failure(
+                self.persistence.persist_events([(event, context)]),
+                PartialStateConflictError,
+            )
+        notify.assert_called()
+        failed = event.internal_metadata.stream_ordering
+        assert failed is not None
+
+        event_id = self.helper.send(self.room_id, "after", tok=self.token)["event_id"]
+        position = self.get_success(self.store.get_position_for_event(event_id))
+        self.assertGreater(position.stream, failed)
+        # A leaked ordering would hold the token below it.
+        self.assertGreaterEqual(
+            self.store._stream_id_gen.get_current_token(), position.stream
+        )
+
+    def test_applier_failure_finishes_orderings(self) -> None:
+        """If applying the effects raises before the stream orderings are
+        finished, they are still finished, and replication is told."""
+
+        class ApplyFailed(Exception):
+            pass
+
+        event, context = self._build("hello")
+        notifier = self.hs.get_replication_notifier()
+        with (
+            patch.object(
+                self.store._events_stream_cache,
+                "entity_has_changed",
+                side_effect=ApplyFailed(),
+            ),
+            patch.object(
+                notifier, "notify_replication", wraps=notifier.notify_replication
+            ) as notify,
+        ):
+            self.get_failure(
+                self.persistence.persist_events([(event, context)]), ApplyFailed
+            )
+        notify.assert_called()
+        effects = self.persister.last_effects
+        assert effects is not None
+        (stream_ordering,) = effects.stream_ids
+        self.assertGreaterEqual(
+            self.store._stream_id_gen.get_current_token(), stream_ordering
+        )
+
+    def test_cancelled_persist_applies_its_effects(self) -> None:
+        """Cancelling the persist while Rust takes the orderings and writes
+        the chunk waits for both, and the orderings are finished."""
+        gate: list[tuple[Any, Any, defer.Deferred]] = []
+        blocking = self.persister._persister
+
+        class Gated:
+            def persist_chunk(self, *args: Any, **kwargs: Any) -> Any:
+                d: defer.Deferred = defer.Deferred()
+                gate.append((args, kwargs, d))
+                return d
+
+            def __getattr__(self, name: str) -> Any:
+                return getattr(blocking, name)
+
+        self.persister._persister = Gated()
+
+        persist_events = self.hs.get_datastores().persist_events
+        assert persist_events is not None
+        original = persist_events._persist_events_in_rust
+        cancelled = []
+
+        async def cancel_then_wait(*args: Any, **kwargs: Any) -> None:
+            d = defer.ensureDeferred(original(*args, **kwargs))
+            d.cancel()
+            try:
+                await make_deferred_yieldable(d)
+            except defer.CancelledError:
+                cancelled.append(True)
+
+        with patch.object(
+            persist_events, "_persist_events_in_rust", side_effect=cancel_then_wait
+        ):
+            channel = make_request(
+                self.reactor,
+                self.site,
+                "PUT",
+                f"/rooms/{self.room_id}/send/m.room.message/txn1",
+                {"msgtype": "m.text", "body": "hello"},
+                access_token=self.token,
+                await_result=False,
+            )
+            self.pump()
+            self.assertEqual(len(gate), 1)
+            # Cancelled, but waiting for the write.
+            self.assertEqual(cancelled, [])
+            args, kwargs, d = gate[0]
+            blocking.persist_chunk(*args, **kwargs).chainDeferred(d)
+            channel.await_result()
+        self.assertEqual(channel.code, 200, channel.json_body)
+        event_id = channel.json_body["event_id"]
+        self.assertEqual(cancelled, [True])
+
+        effects = self.persister.last_effects
+        assert effects is not None
+        (stream_ordering,) = effects.stream_ids
+        self.assertEqual(self._event_row(event_id)[0], stream_ordering)
+        self.assertGreaterEqual(
+            self.store._stream_id_gen.get_current_token(), stream_ordering
+        )
+
+
+class RustTransactionWorkerTestCase(BaseMultiWorkerStreamTestCase):
+    """`rust_event_persistence: transaction` on a worker event persister:
+    the worker writes in Rust, to the main database."""
+
+    level = "transaction"
+
+    servlets = [
+        admin.register_servlets,
+        room.register_servlets,
+        login.register_servlets,
+    ]
+
+    def default_config(self) -> dict[str, Any]:
+        config = super().default_config()
+        config.setdefault("experimental_features", {})["rust_event_persistence"] = (
+            self.level
         )
         config["stream_writers"] = {"events": ["worker1"]}
         config["instance_map"] = {
@@ -1097,7 +1448,7 @@ class RustTransactionWorkerTestCase(BaseMultiWorkerStreamTestCase):
         return config
 
     def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
-        if RUST_PERSIST_FOR_TESTS not in (None, "transaction"):
+        if RUST_PERSIST_FOR_TESTS not in (None, self.level):
             raise SkipTest("SYNAPSE_TEST_RUST_PERSIST overrides the level")
         if not isinstance(
             hs.get_datastores().main.db_pool._db_pool, RustConnectionPool
@@ -1118,7 +1469,10 @@ class RustTransactionWorkerTestCase(BaseMultiWorkerStreamTestCase):
         room_id = self.helper.create_room_as(user, tok=token)
         event_id = self.helper.send(room_id, "hello", tok=token)["event_id"]
 
-        self.assertGreater(persister.transactions, 0)
+        if self.level == "full":
+            self.assertGreater(persister.chunks, 0)
+        else:
+            self.assertGreater(persister.transactions, 0)
         assert persister.last_effects is not None
         self.assertIn(event_id, persister.last_effects.event_cache_prefills)
         # The main process reads what the worker's Rust pool wrote.
@@ -1127,6 +1481,30 @@ class RustTransactionWorkerTestCase(BaseMultiWorkerStreamTestCase):
         self.assertEqual(position.instance_name, "worker1")
         event = self.get_success(store.get_event(event_id))
         self.assertEqual(event.content["body"], "hello")
+        if self.level == "full":
+            # The worker took the ordering, finished it and moved its
+            # `stream_positions` row on.
+            self.assertEqual(persister.last_effects.stream_ids, [position.stream])
+            worker_id_gen = worker.get_datastores().main._stream_id_gen
+            self.assertGreaterEqual(
+                worker_id_gen.get_current_token_for_writer("worker1"),
+                position.stream,
+            )
+            row = self.get_success(
+                store.db_pool.simple_select_one_onecol(
+                    "stream_positions",
+                    {"stream_name": "events", "instance_name": "worker1"},
+                    "stream_id",
+                )
+            )
+            self.assertEqual(row, position.stream)
+
+
+class RustFullWorkerTestCase(RustTransactionWorkerTestCase):
+    """`rust_event_persistence: full` on a worker event persister: the
+    worker takes the stream orderings in Rust."""
+
+    level = "full"
 
 
 class PersisterOffTestCase(unittest.HomeserverTestCase):