Catch up MSC4354 Sticky Events to remote homeservers that have missed them. (#20165)

Part of: MSC4354 (Sticky Events)

Our locally-sent Sticky Events need to be reliably (within bounds of
possibility) sent to remote homeservers.

This adds the mechanism to make this happen.

Notably omits catching up historical sticky events to homeservers that only just
joined the room.

---------

Signed-off-by: Olivier 'reivilibre <oliverw@matrix.org>
Co-authored-by: Eric Eastwood <erice@element.io>

changelog.d/20165.feature

@@ -0,0 +1 @@
+Catch up [MSC4354 Sticky Events](https://github.com/matrix-org/matrix-spec-proposals/pull/4354) to remote homeservers that have missed them.

synapse/federation/sender/per_destination_queue.py

@@ -20,6 +20,7 @@
 #
 #
 import datetime
+import itertools
 import logging
 from collections import OrderedDict
 from typing import TYPE_CHECKING, Hashable, Iterable
@@ -47,7 +48,8 @@ from synapse.logging import issue9533_logger
 from synapse.logging.context import PreserveLoggingContext
 from synapse.logging.opentracing import SynapseTags, set_tag
 from synapse.metrics import SERVER_NAME_LABEL, sent_transactions_counter
-from synapse.types import JsonDict, ReadReceipt
+from synapse.replication.tcp.streams._base import StickyEventStreamPosition
+from synapse.types import JsonDict, ReadReceipt, RoomID, unwrap
 from synapse.util.retryutils import NotRetryingDestination, get_retry_limiter
 from synapse.visibility import filter_events_for_server
 
@@ -77,6 +79,26 @@ CATCHUP_RETRY_INTERVAL = 60 * 60 * 1000
 # they are bounded in size.
 MAX_PRESENCE_STATES_PER_EDU = 50
 
+MAX_PDUS_PER_TRANSACTION = 50
+"""
+How many PDUs we want to send in a transaction. We use the spec limit of 50 here.
+https://spec.matrix.org/v1.19/server-server-api/#put_matrixfederationv1sendtxnid
+"""
+
+
+@attr.s(slots=True, auto_attribs=True, frozen=True)
+class _StickyEventsTransactionInfo:
+    room_id: RoomID
+    """
+    The room ID from which backlogged sticky events were sent.
+    """
+
+    max_sent_sticky_events_stream_position: StickyEventStreamPosition
+    """
+    The maximum sticky events stream position of the backlogged sticky events
+    sent in this transaction.
+    """
+
 
 @attr.s(slots=True, auto_attribs=True, frozen=True)
 class _PreparedTransaction:
@@ -123,6 +145,18 @@ class _PreparedTransaction:
     When the transaction completes, this should be stored as our position in the events stream.
     """
 
+    sticky_events: _StickyEventsTransactionInfo | None
+    """
+    Information useful for transactions sending backlogged sticky events.
+    """
+
+    def has_anything_to_send(self) -> bool:
+        """
+        Returns true if and only if this contains any
+        data to send to the remote homeserver.
+        """
+        return bool(self.pdus or self.edus)
+
 
 class PerDestinationQueue:
     """
@@ -151,6 +185,16 @@ class PerDestinationQueue:
         self._instance_name = hs.get_instance_name()
         self._federation_shard_config = hs.config.worker.federation_shard_config
         self._state = hs.get_state_handler()
+        self._sticky_event_backlog_tracker = StickyEventBacklogTracker(
+            destination, self._hs, self._hs.config.experimental.msc4354_enabled
+        )
+        self._sticky_backlog_turn = False
+        """
+        Whether the next transaction we attempt to prepare should be a sticky event backlog transaction.
+
+        Alternating between the main real-time queue and the sticky event backlog ensures that neither
+        can starve the other.
+        """
 
         self._should_send_on_this_instance = True
         if not self._federation_shard_config.should_handle(
@@ -192,7 +236,7 @@ class PerDestinationQueue:
         self._last_successful_stream_ordering: int | None = None
 
         # a queue of pending PDUs
-        self._pending_pdus: list[EventBase] = []
+        self._pending_pdus: OrderedDict[str, EventBase] = OrderedDict()
 
         # XXX this is never actually used: see
         # https://github.com/matrix-org/synapse/issues/7549
@@ -257,7 +301,7 @@ class PerDestinationQueue:
         if not self._catching_up or self._last_successful_stream_ordering is None:
             # only enqueue the PDU if we are not catching up (False) or do not
             # yet know if we have anything to catch up (None)
-            self._pending_pdus.append(pdu)
+            self._pending_pdus[pdu.event_id] = pdu
         else:
             assert pdu.internal_metadata.stream_ordering
             self._catchup_last_skipped = pdu.internal_metadata.stream_ordering
@@ -416,18 +460,29 @@ class PerDestinationQueue:
                 transaction = None
                 transaction = await self._prepare_transaction()
 
+                if transaction is not None and not transaction.has_anything_to_send():
+                    # We built a _PreparedTransaction but it has nothing to send!
+                    # This can happen when all the PDUs are filtered out by visibility checks.
+                    # Even though the transaction doesn't contain anything, for sticky event backlog transactions at least,
+                    # the mere act of preparing the transaction has made progress that should be recorded, so we
+                    # complete it here.
+                    await self._complete_transaction(transaction)
+
+                    # Now loop back to try to build another transaction.
+                    continue
+
                 if transaction is None:
                     logger.debug("TX [%s] Nothing to send", self._destination)
 
-                    # If we've gotten told about new things to send during
-                    # checking for things to send, we try looking again.
-                    # Otherwise new PDUs or EDUs might arrive in the meantime,
-                    # but not get sent because we currently have an
-                    # `_active_transmission_loop` running.
                     if self._new_data_to_send:
+                        # If we've gotten told about new things to send during
+                        # checking for things to send, we try looking again.
+                        # Otherwise new PDUs or EDUs might arrive in the meantime,
+                        # but not get sent because we currently have an
+                        # `_active_transmission_loop` running.
                         continue
-                    else:
-                        return
+
+                    return
 
                 if transaction.pdus:
                     logger.debug(
@@ -531,8 +586,7 @@ class PerDestinationQueue:
                 )
             )
 
-        _tmp_last_successful_stream_ordering = self._last_successful_stream_ordering
-        if _tmp_last_successful_stream_ordering is None:
+        if self._last_successful_stream_ordering is None:
             # if it's still None, then this means we don't have the information
             # in our database ­ we haven't successfully sent a PDU to this server
             # (at least since the introduction of the feature tracking
@@ -542,12 +596,10 @@ class PerDestinationQueue:
             self._catching_up = False
             return
 
-        last_successful_stream_ordering: int = _tmp_last_successful_stream_ordering
-
         # get at most 50 catchup room/PDUs
         while self._transmission_loop_enabled:
             event_ids = await self._store.get_catch_up_room_event_ids(
-                self._destination, last_successful_stream_ordering
+                self._destination, self._last_successful_stream_ordering
             )
 
             if not event_ids:
@@ -557,7 +609,8 @@ class PerDestinationQueue:
 
                 if (
                     self._catchup_last_skipped != 0
-                    and self._catchup_last_skipped > last_successful_stream_ordering
+                    and self._catchup_last_skipped
+                    > self._last_successful_stream_ordering
                 ):
                     # another event has been skipped because we were in catch-up mode
                     # As an exception to this case: we can hit this branch if the
@@ -643,7 +696,7 @@ class PerDestinationQueue:
                         # offline
                         if (
                             p.internal_metadata.stream_ordering
-                            < last_successful_stream_ordering
+                            < self._last_successful_stream_ordering
                         ):
                             continue
 
@@ -688,15 +741,30 @@ class PerDestinationQueue:
                 # We pulled this from the DB, so it'll be non-null
                 assert pdu.internal_metadata.stream_ordering
 
+                # When advancing our `last_successful_stream_ordering` position,
+                # there may be unsent sticky events 'in the gap'; note that down as a backlog.
+                await self._store.mark_backlogged_sticky_events_after_catchup_transaction(
+                    self._destination,
+                    old_last_successfully_sent_stream_ordering=self._last_successful_stream_ordering,
+                    new_last_successfully_sent_stream_ordering=pdu.internal_metadata.stream_ordering,
+                    # These are the events we actually sent in this successful catch-up transaction
+                    event_stream_orderings_sent_in_transaction={
+                        # unwrap: these events have been persisted so `stream_ordering` is not None
+                        unwrap(room_catchup_pdu.internal_metadata.stream_ordering)
+                        for room_catchup_pdu in room_catchup_pdus
+                    },
+                )
+                self._sticky_event_backlog_tracker.notify_potential_new_backlog()
+
                 # Note that we mark the last successful stream ordering as that
                 # from the *original* PDU, rather than the PDU(s) we actually
                 # send. This is because we use it to mark our position in the
                 # queue of missed PDUs to process.
-                last_successful_stream_ordering = pdu.internal_metadata.stream_ordering
-
-                self._last_successful_stream_ordering = last_successful_stream_ordering
+                self._last_successful_stream_ordering = (
+                    pdu.internal_metadata.stream_ordering
+                )
                 await self._store.set_destination_last_successful_stream_ordering(
-                    self._destination, last_successful_stream_ordering
+                    self._destination, self._last_successful_stream_ordering
                 )
 
     def _get_receipt_edus(self, limit: int) -> Iterable[Edu]:
@@ -784,13 +852,15 @@ class PerDestinationQueue:
         This throws away the PDU queue.
         """
         self._catching_up = True
-        self._pending_pdus = []
+        self._pending_pdus = OrderedDict()
 
     async def _prepare_transaction(self) -> _PreparedTransaction | None:
         """
-        Work out what should go in the next transaction to this destination, by
-        calculating what we want to send and preparing the information that is
-        useful once we have completed the transaction.
+        Work out what should go in the next transaction to this destination.
+
+        Round-robin alternates between:
+          - the backlog of sticky events; and
+          - the normal real-time queue
 
         Side effects:
             - Dequeues pending EDUs
@@ -811,6 +881,46 @@ class PerDestinationQueue:
         `_complete_transaction` must be called with it.
         """
 
+        if self._sticky_backlog_turn:
+            # Sticky event backlog transaction
+
+            # The normal queue will have the next turn
+            self._sticky_backlog_turn = False
+
+            backlog_transaction = (
+                await self._sticky_event_backlog_tracker.prepare_transaction()
+            )
+            if backlog_transaction is not None:
+                return backlog_transaction
+
+            # Fall back to a main queue transaction
+            return await self._prepare_main_queue_transaction()
+        else:
+            # Main queue (normal) transaction
+            transaction = await self._prepare_main_queue_transaction()
+
+            # If we have a sticky event backlog, the next turn
+            # will be for the sticky event backlog
+            self._sticky_backlog_turn = self._sticky_event_backlog_tracker.is_backlogged
+
+            if transaction is not None:
+                return transaction
+
+            # Fall back to a sticky backlog turn, if it has anything
+            if self._sticky_event_backlog_tracker.is_backlogged:
+                return await self._sticky_event_backlog_tracker.prepare_transaction()
+
+            return None
+
+    async def _prepare_main_queue_transaction(self) -> _PreparedTransaction | None:
+        """
+        Prepare a transaction from the normal real-time queue, by calculating what we
+        want to send and the information that is useful once we have completed the
+        transaction.
+
+        Returns None if the normal queue has nothing to send.
+        """
+
         # First we calculate the EDUs we want to send, if any.
 
         # There's a maximum number of EDUs that can be sent with a transaction,
@@ -899,7 +1009,9 @@ class PerDestinationQueue:
 
         # Now we look for any PDUs to send, by getting up to 50 PDUs from the
         # queue
-        pdus = self._pending_pdus[:50]
+        pdus: list[EventBase] = list(
+            itertools.islice(self._pending_pdus.values(), MAX_PDUS_PER_TRANSACTION)
+        )
 
         if not pdus and not pending_edus:
             # There is nothing to send. There's also nothing to record upon
@@ -919,6 +1031,9 @@ class PerDestinationQueue:
             to_device_message_stream_id=device_stream_id_upon_completion,
             device_list_stream_id=device_list_id_upon_completion,
             last_stream_ordering=last_stream_ordering,
+            # This is not part of the sticky events backlog flow,
+            # so don't advance that
+            sticky_events=None,
         )
 
     async def _complete_transaction(self, transaction: _PreparedTransaction) -> None:
@@ -929,8 +1044,9 @@ class PerDestinationQueue:
         through the various streams we have now got.
         """
         # Successfully sent transactions, so we remove pending PDUs from the queue
-        if transaction.pdus:
-            self._pending_pdus = self._pending_pdus[len(transaction.pdus) :]
+        for pdu in transaction.pdus:
+            # Remove sent events from queue
+            self._pending_pdus.pop(pdu.event_id, None)
 
         # Succeeded to send the transaction so we record where we have sent up
         # to in the various streams
@@ -959,3 +1075,144 @@ class PerDestinationQueue:
             await self._store.set_destination_last_successful_stream_ordering(
                 self._destination, transaction.last_stream_ordering
             )
+
+        if transaction.sticky_events is not None:
+            await self._sticky_event_backlog_tracker.complete_transaction(
+                transaction.sticky_events
+            )
+
+
+class StickyEventBacklogTracker:
+    """
+    Tracks our state with sticky events.
+    """
+
+    def __init__(
+        self, destination: str, hs: "synapse.server.HomeServer", msc4354_enabled: bool
+    ) -> None:
+        # Assume backlogged by default
+        self._backlogged = True
+        """
+        Do we *potentially* have a backlog of sticky events to send out?
+        """
+
+        self._destination = destination
+        """
+        The server name of the destination we are responsible for.
+        """
+
+        self._own_server_name = hs.hostname
+
+        self._storage_controllers = hs.get_storage_controllers()
+
+        self._store = hs.get_datastores().main
+
+        self._msc4354_enabled = msc4354_enabled
+        """
+        MSC4354 Sticky Events
+
+        Specifically in this class, enables catch-up of the backlogged to-be-sent sticky events.
+        """
+
+    @property
+    def is_backlogged(self) -> bool:
+        """
+        Whether we *potentially* have a backlog of sticky events to send out.
+
+        Always false when MSC4354 is disabled, as there is then nothing to send.
+        """
+        return self._msc4354_enabled and self._backlogged
+
+    async def prepare_transaction(self) -> _PreparedTransaction | None:
+        """
+        Try to prepare a transaction based on the sticky event backlog.
+
+        Returns None if there is no backlog to make progress on right now.
+        """
+
+        if not self._msc4354_enabled:
+            # MSC4354 Sticky Events disabled, so nothing to do.
+            return None
+
+        if not self._backlogged:
+            return None
+
+        # Select a room and get up to 50 backlogged sticky events (to fill a transaction)
+        backlog = await self._store.get_backlogged_sticky_events_for_destination(
+            self._destination, limit=MAX_PDUS_PER_TRANSACTION
+        )
+
+        if backlog is None:
+            logger.info(
+                "Caught up on federation sticky event backlog for destination %r",
+                self._destination,
+            )
+            self._backlogged = False
+            return None
+
+        room_id, sticky_event_stream_position, event_ids = backlog
+
+        logger.debug(
+            "Selected %d backlogged sticky events to send to destination %r in room %r (up to stream position %r)",
+            len(event_ids),
+            self._destination,
+            room_id,
+            sticky_event_stream_position,
+        )
+
+        # Fetch the events from the database
+        sticky_events = await self._store.get_events_as_list(event_ids)
+
+        # Filter the sticky events
+        sticky_events = await filter_events_for_server(
+            self._storage_controllers,
+            target_server_name=self._destination,
+            local_server_name=self._own_server_name,
+            events=sticky_events,
+            # Omit filtered events instead of redacting them
+            # (As if the remote isn't allowed to see them, there's no point sending them there.
+            # Not to mention sending a redacted copy would mean the remote sees the event without
+            # any stickiness, as the `msc4354_sticky` field is affected by redaction.)
+            redact=False,
+            # Sticky events sent by erased users no longer need to be sent
+            # as part of catch-up
+            #
+            # > When a sticky event was sent by a [user who has been erased](https://spec.matrix.org/v1.19/client-server-api/#post_matrixclientv3accountdeactivate),
+            # > servers SHOULD NOT send it to other homeservers as part of catch-up.
+            # > — https://github.com/matrix-org/matrix-spec-proposals/blob/4ad14b0cd3b09205dcba59e45cbf1cab1e75edf7/proposals/4354-sticky-events.md#L230-L231
+            filter_out_erased_senders=True,
+            # These are all local events, so no need to do any extra work
+            # only relevant to remote events
+            filter_out_remote_partial_state_events=False,
+        )
+
+        return _PreparedTransaction(
+            pdus=sticky_events,
+            # No EDUs are sent alongside backlogged sticky events.
+            edus=[],
+            device_list_stream_id=None,
+            to_device_message_stream_id=None,
+            last_stream_ordering=None,
+            # Upon completion, advance in the sticky backlog stream
+            sticky_events=_StickyEventsTransactionInfo(
+                room_id=room_id,
+                max_sent_sticky_events_stream_position=sticky_event_stream_position,
+            ),
+        )
+
+    async def complete_transaction(self, info: _StickyEventsTransactionInfo) -> None:
+        """
+        Call upon successfully sending a transaction generated by `prepare_transaction`.
+
+        Will advance the backlogged sticky events stream position in the database.
+        """
+        await self._store.mark_backlogged_sticky_events_sent(
+            self._destination, info.room_id, info.max_sent_sticky_events_stream_position
+        )
+
+    def notify_potential_new_backlog(self) -> None:
+        """
+        Call when something may have added to the sticky event backlog in the database,
+        so that we remember to check it when we next send transactions.
+        """
+        self._backlogged = True

synapse/replication/tcp/streams/_base.py

@@ -26,6 +26,7 @@ from typing import (
     Any,
     Awaitable,
     Callable,
+    NewType,
     Sequence,
     Union,
 )
@@ -866,6 +867,13 @@ class ProfileUpdatesStream(_StreamFromIdGen):
         return rows, rows[-1][0], len(updates) == limit
 
 
+StickyEventStreamPosition = NewType("StickyEventStreamPosition", int)
+"""
+Integer corresponding to the stream position (`stream_id`)
+of a sticky event in the `sticky_events` table.
+"""
+
+
 @attr.s(slots=True, auto_attribs=True)
 class StickyEventsStreamRow:
     """Stream to inform workers about changes to sticky events."""

synapse/storage/database.py

@@ -2757,3 +2757,20 @@ def make_tuple_comparison_clause(keys: list[tuple[str, KV]]) -> tuple[str, list[
         "(%s) > (%s)" % (",".join(k[0] for k in keys), ",".join("?" for _ in keys)),
         [k[1] for k in keys],
     )
+
+
+def user_is_local_like_pattern(hs: "HomeServer") -> str:
+    """
+    Returns a LIKE pattern that matches the User IDs of local users on this
+    homeserver.
+
+    The caller should bind this pattern to a parameter and use it in
+    a `user_id LIKE ?` clause.
+
+    (This function purposefully requires a `HomeServer` object so
+    that it's not accidentally used with untrusted 'server names'
+    that might contain LIKE pattern metacharacters.)
+    """
+    # This is good enough as if you have silly characters in your own
+    # hostname then that's your own fault.
+    return f"@%:{hs.hostname}"

synapse/storage/databases/main/event_push_actions.py

@@ -100,6 +100,7 @@ from synapse.storage.database import (
     LoggingDatabaseConnection,
     LoggingTransaction,
     PostgresEngine,
+    user_is_local_like_pattern,
 )
 from synapse.storage.databases.main.receipts import ReceiptsWorkerStore
 from synapse.storage.databases.main.stream import StreamWorkerStore
@@ -1439,7 +1440,7 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
 
         # We only want local users, so we add a dodgy filter to the above query
         # and recheck it below.
-        user_filter = "%:" + self.hs.hostname
+        user_filter = user_is_local_like_pattern(self.hs)
 
         txn.execute(
             sql,

synapse/storage/databases/main/metrics.py

@@ -30,6 +30,7 @@ from synapse.storage.database import (
     DatabasePool,
     LoggingDatabaseConnection,
     LoggingTransaction,
+    user_is_local_like_pattern,
 )
 from synapse.storage.databases.main.event_push_actions import (
     EventPushActionsWorkerStore,
@@ -133,9 +134,7 @@ class ServerMetricsStore(EventPushActionsWorkerStore, SQLBaseStore):
 
     async def count_daily_sent_e2ee_messages(self) -> int:
         def _count_messages(txn: LoggingTransaction) -> int:
-            # This is good enough as if you have silly characters in your own
-            # hostname then that's your own fault.
-            like_clause = "%:" + self.hs.hostname
+            like_clause = user_is_local_like_pattern(self.hs)
 
             sql = """
                 SELECT COUNT(*) FROM events
@@ -189,9 +188,7 @@ class ServerMetricsStore(EventPushActionsWorkerStore, SQLBaseStore):
 
     async def count_daily_sent_messages(self) -> int:
         def _count_messages(txn: LoggingTransaction) -> int:
-            # This is good enough as if you have silly characters in your own
-            # hostname then that's your own fault.
-            like_clause = "%:" + self.hs.hostname
+            like_clause = user_is_local_like_pattern(self.hs)
 
             sql = """
                 SELECT COUNT(*) FROM events

synapse/storage/databases/main/purge_events.py

@@ -23,7 +23,7 @@ import logging
 from typing import Any, cast
 
 from synapse.api.errors import SynapseError
-from synapse.storage.database import LoggingTransaction
+from synapse.storage.database import LoggingTransaction, user_is_local_like_pattern
 from synapse.storage.databases.main import CacheInvalidationWorkerStore
 from synapse.storage.databases.main.state import StateGroupWorkerStore
 from synapse.storage.engines import PostgresEngine
@@ -53,6 +53,7 @@ Tables which lack an index on `room_id` but have one on `event_id`
 purge_room_tables_with_room_id_column = (
     "current_state_events",
     "destination_rooms",
+    "destination_rooms_sticky_events_backlog",
     "event_backward_extremities",
     "event_forward_extremities",
     "event_push_actions",
@@ -75,6 +76,8 @@ purge_room_tables_with_room_id_column = (
     # so must be deleted first.
     "msc4242_state_dag_forward_extremities",
     "msc4242_state_dag_edges",
+    # Note: `sticky_events` has foreign key to `events` so must be deleted first.
+    "sticky_events",
     "events",
     "federation_inbound_events_staging",
     "receipts_graph",
@@ -213,7 +216,10 @@ class PurgeEventsStore(StateGroupWorkerStore, CacheInvalidationWorkerStore):
             should_delete_expr += " AND sender NOT LIKE ?"
 
             # We include the parameter twice since we use the expression twice
-            should_delete_params += ("%:" + self.hs.hostname, "%:" + self.hs.hostname)
+            should_delete_params += (
+                user_is_local_like_pattern(self.hs),
+                user_is_local_like_pattern(self.hs),
+            )
 
         should_delete_params += (room_id, token.topological)
 

synapse/storage/databases/main/sticky_events.py

@@ -2,6 +2,7 @@
 # This file is licensed under the Affero General Public License (AGPL) version 3.
 #
 # Copyright (C) 2025 New Vector, Ltd
+# 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
@@ -24,19 +25,23 @@ from synapse.api.constants import EventTypes
 from synapse.api.errors import AuthError
 from synapse.events import EventBase
 from synapse.events.snapshot import EventPersistencePair
-from synapse.replication.tcp.streams._base import StickyEventsStream
+from synapse.replication.tcp.streams._base import (
+    StickyEventsStream,
+    StickyEventStreamPosition,
+)
 from synapse.storage.database import (
     DatabasePool,
     LoggingDatabaseConnection,
     LoggingTransaction,
     make_in_list_sql_clause,
+    user_is_local_like_pattern,
 )
 from synapse.storage.databases.main.cache import CacheInvalidationWorkerStore
 from synapse.storage.databases.main.events import DeltaState
 from synapse.storage.databases.main.state import StateGroupWorkerStore
 from synapse.storage.engines import PostgresEngine, Sqlite3Engine
 from synapse.storage.util.id_generators import MultiWriterIdGenerator
-from synapse.types import StateKey
+from synapse.types import RoomID, StateKey
 from synapse.types.state import StateFilter
 from synapse.util.duration import Duration
 from synapse.util.stringutils import shortstr
@@ -759,3 +764,366 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor
             "delete_expired_sticky_events",
             self._delete_expired_sticky_events,
         )
+
+    async def get_backlogged_sticky_events_for_destination(
+        self, destination: str, *, limit: int = 50
+    ) -> tuple[RoomID, StickyEventStreamPosition, list[str]] | None:
+        """
+        From the `destination_rooms_sticky_events_backlog` table, if there are backlogged
+        sticky events to send to the given destination, returns up to the `limit` of sticky
+        event IDs from one room.
+
+        Only processing one room at once is an arbitrary choice, motivated by regular catch-up transactions
+        only processing one room at once.
+        It is also slightly simpler for us to keep track of.
+        We have comments related to old catch-up transaction behaviour, explaining why we send only forward extremities
+        (because it is relatively intensive for a server to receive events from several places in the DAG at once).
+        As sticky event catch-up transactions seem to meet this 'relatively intensive' category, it seems to make sense
+        to keep the transaction complexity down (which may be a reason to reduce the limit in the future).
+        Sending events, no more than 1 hour old, from only one room _probably_ means that they can benefit from cache
+        warmth at the receiving server.
+        With that said, we can still consider changing this later. It was just the ambient thought process going into this.
+
+        The sticky events are constrained to originating from this server:
+
+        > Attempt to **push** their own[^origin] sticky events to all joined servers
+        > — https://github.com/matrix-org/matrix-spec-proposals/blame/74fc75e1dc1301230cc3fcb7435205bf4f567ef8/proposals/4354-sticky-events.md#L88
+        >
+        > [^origin]: That is, the domain of the sender of the sticky event is the sending server.
+        > — https://github.com/matrix-org/matrix-spec-proposals/blame/74fc75e1dc1301230cc3fcb7435205bf4f567ef8/proposals/4354-sticky-events.md#L491
+
+        The sticky events are ordered by oldest `sticky_events.stream_id` first,
+        which corresponds to `stream_ordering` first for locally-originating events.
+
+        Returns
+            - `None` if no backlog exists
+            - if a backlog exists, a tuple of
+                1. room ID
+                2. The sticky event stream position that should be advanced to upon
+                   successful sending of this batch.
+                   (currently: the highest sticky event stream position of the returned sticky events)
+                3. event IDs of backlogged sticky events (between 1 and `limit` of them)
+        """
+
+        now_millis = self.clock.time_msec()
+
+        def _get_backlogged_sticky_events_for_destination_txn(
+            txn: LoggingTransaction,
+        ) -> tuple[RoomID, StickyEventStreamPosition, list[str]] | None:
+            first_try = _try_get_backlogged_sticky_events_for_destination_txn(txn)
+            if first_try is None:
+                return None
+
+            room_id, advance_sticky_event_stream_pos, sticky_event_ids = first_try
+            if sticky_event_ids:
+                assert advance_sticky_event_stream_pos is not None
+                return room_id, advance_sticky_event_stream_pos, sticky_event_ids
+
+            # A room is considered backlogged but doesn't have any
+            # sticky events to send
+            # This can happen when the sticky events expire, for instance.
+            # Trigger a cleanup of the table for this destination and try round again.
+            _clean_backlog_txn(txn)
+
+            # After having cleaned the backlog, try again
+            second_try = _try_get_backlogged_sticky_events_for_destination_txn(txn)
+            if not second_try:
+                return None
+            room_id, max_sticky_events_stream_position, event_ids = second_try
+
+            assert len(event_ids) > 0
+            assert max_sticky_events_stream_position is not None
+
+            return room_id, max_sticky_events_stream_position, event_ids
+
+        def _try_get_backlogged_sticky_events_for_destination_txn(
+            txn: LoggingTransaction,
+        ) -> tuple[RoomID, StickyEventStreamPosition | None, list[str]] | None:
+            """
+            Attempt to pull out backlogged sticky events for the destination
+            from any room.
+
+            Returns
+                - `None` if no backlog exists
+                - if a backlog exists, a tuple of
+                    1. room ID
+                    2. The sticky event stream position that should be advanced to upon
+                       successful sending of this batch, or `None` if no events.
+                       (currently: the highest sticky event stream position of the returned sticky events)
+                    3. event IDs of backlogged sticky events (between 0 and `limit` of them)
+
+                  It is possible for a room ID to be returned with zero sticky events,
+                  for example if all the backlogged sticky events for that room expired.
+
+                  In that case, clean-up should be triggered on the table and then
+                  try again.
+            """
+
+            txn.execute(
+                """
+                SELECT room_id, sticky_events_stream_position
+                FROM destination_rooms_sticky_events_backlog
+                WHERE destination = ?
+                LIMIT 1
+                """,
+                (destination,),
+            )
+            row = txn.fetchone()
+            if not row:
+                return None
+
+            room_id, next_to_send_sticky_event_stream_position = cast(
+                tuple[str, int], row
+            )
+
+            txn.execute(
+                """
+                SELECT event_id, stream_id
+                FROM sticky_events
+                WHERE room_id = ?
+                    AND ? <= stream_id
+                    AND ? < expires_at
+                    -- filter to locally-originating sticky events
+                    AND sender LIKE ?
+                ORDER BY stream_id ASC
+                LIMIT ?
+                """,
+                (
+                    room_id,
+                    next_to_send_sticky_event_stream_position,
+                    now_millis,
+                    user_is_local_like_pattern(self.hs),
+                    limit,
+                ),
+            )
+
+            # -1 and below aren't used as stream positions
+            max_stream_position = -1
+            event_ids = []
+            for event_id, stream_position in txn:
+                event_ids.append(event_id)
+                max_stream_position = max(max_stream_position, stream_position)
+
+            max_stream_position_return = (
+                None
+                if max_stream_position == -1
+                else StickyEventStreamPosition(max_stream_position)
+            )
+
+            return RoomID.from_string(room_id), max_stream_position_return, event_ids
+
+        def _clean_backlog_txn(txn: LoggingTransaction) -> None:
+            """
+            Clean up `destination_rooms_sticky_events_backlog` rows that no longer apply,
+            because there are no longer active sticky events in that range in that room.
+
+            Invoked when we try to process a room and find that it has no sticky events
+            to send to this destination.
+            """
+            txn.execute(
+                """
+                WITH to_clean_up AS (
+                    SELECT backlog.room_id FROM destination_rooms_sticky_events_backlog AS backlog
+                    -- This is an anti-join: we want to find backlog rows where no sticky events match
+                    LEFT JOIN sticky_events AS se
+                        ON se.room_id = backlog.room_id
+                        -- filter to locally-originating sticky events
+                        AND se.sender LIKE ?
+                        AND ? < se.expires_at
+                        AND backlog.sticky_events_stream_position <= se.stream_id
+                    WHERE se.event_id IS NULL
+                        AND backlog.destination = ?
+                )
+                DELETE FROM destination_rooms_sticky_events_backlog
+                WHERE destination = ? AND room_id IN (SELECT room_id FROM to_clean_up)
+                """,
+                (
+                    user_is_local_like_pattern(self.hs),
+                    now_millis,
+                    destination,
+                    destination,
+                ),
+            )
+
+        return await self.db_pool.runInteraction(
+            "get_backlogged_sticky_events_for_destination",
+            _get_backlogged_sticky_events_for_destination_txn,
+        )
+
+    async def mark_backlogged_sticky_events_after_catchup_transaction(
+        self,
+        destination: str,
+        *,
+        old_last_successfully_sent_stream_ordering: int,
+        new_last_successfully_sent_stream_ordering: int,
+        event_stream_orderings_sent_in_transaction: Collection[int],
+    ) -> None:
+        """
+        For the given `destination`, update the `destination_rooms_sticky_events_backlog`
+        table to potentially mark rooms as backlogged, following the successful
+        transmission of PDUs in a catch-up (federation) transaction.
+
+        Only catch-up transactions skip over PDUs in the 'outbox' (so to speak),
+        or in other words: they produce a 'gap' of unsent events (PDUs).
+        This implies that they can produce a gap of unsent *sticky* events,
+        which we need to carefully track and ensure we make a best-effort attempt
+        to send them later.
+
+        As a brief reminder: a catch-up transaction sends a subset of one room's
+        forward extremities, then advances `last_successfully_sent_stream_ordering`
+        for the destination.
+
+
+        Let's imagine this situation, with 3 rooms containing events that have not
+        yet been sent to the destination:
+
+        ```
+                legend: . = event
+                        S = sticky event
+
+                    -----------> event stream_ordering
+
+                   |
+            room1  |  .    .    .    S   .   .
+            room2  |   .     S    .    S   .    S
+            room3  |    .      .   S    .   .    .
+                   |
+                   |
+                   ^
+                   last_successfully_sent_stream_ordering
+        ```
+
+        A catch-up transaction then happens, which selects room1 as it has the oldest
+        (in stream_ordering terms) forward extremity.
+        After the transaction is sent successfully, the `last_successfully_sent_stream_ordering`
+        is advanced in kind.
+
+        ```
+                    -----------> event stream_ordering
+
+                                             |
+            room1     .    .    .    S   .   .
+            room2      .     S    .    S   . |  S
+            room3       .      .   S    .   .|   .
+                                             |
+                                             |
+                                             ^
+                   last_successfully_sent_stream_ordering
+        ```
+
+        In the gap left by this advancement of the `last_successfully_sent_stream_ordering`
+        position, there are 4 sticky events.
+
+        These are the sticky events that this function tracks in the
+        `destination_rooms_sticky_events_backlog` table.
+        Without us doing this, no other mechanism would provide a way of knowing
+        that those 4 sticky events hadn't yet been sent to the destination.
+
+        Arguments:
+            old_last_successfully_sent_stream_ordering:
+                The old position of `last_successfully_sent_stream_ordering`
+            new_last_successfully_sent_stream_ordering:
+                The new position of `last_successfully_sent_stream_ordering`
+            event_stream_orderings_sent_in_transaction:
+                event `stream_ordering`s of events that were actually sent in this transaction.
+                These events will not be considered eligible for triggering a backlog.
+        """
+
+        def _txn(txn: LoggingTransaction) -> None:
+            not_event_stream_ordering_in_clause, not_event_stream_ordering_in_args = (
+                make_in_list_sql_clause(
+                    self.database_engine,
+                    "se.event_stream_ordering",
+                    event_stream_orderings_sent_in_transaction,
+                    negative=True,
+                )
+            )
+
+            # This is a pipeline:
+            # 1. In `destination_rooms`, find all rooms associated with this destination,
+            #    unless the room didn't have any events after `old_last_successfully_sent_stream_ordering`
+            # 2. For each room, consider all sticky events with `stream_ordering` within the range
+            #    `old_last_successfully_sent_stream_ordering` < x < `new_last_successfully_sent_stream_ordering`
+            #   3. Except those that were just sent (according to `event_stream_orderings_sent_in_transaction`).
+            #   4. Get the least `sticky_events.stream_id` out of all of those events for the room.
+            # 5. Insert those positions into the backlog, unless the backlog already exists with a smaller position.
+            txn.execute(
+                f"""
+                INSERT INTO destination_rooms_sticky_events_backlog AS backlog
+                (destination, room_id, sticky_events_stream_position)
+
+                    SELECT ?, dr.room_id, MIN(se.stream_id)
+                    FROM destination_rooms AS dr
+                    INNER JOIN sticky_events se USING (room_id)
+                    WHERE
+                        -- Only consider rooms associated with this destination (1)
+                        dr.destination = ?
+
+                        -- Only consider rooms that could possibly have events in the gap (1)
+                        AND ? < dr.stream_ordering
+
+                        -- Only consider events in the gap (2):
+                        AND ? < se.event_stream_ordering
+                        AND se.event_stream_ordering < ?
+
+                        -- Only consider locally-sent events
+                        -- as we're not responsible for sending other servers' events
+                        AND se.sender LIKE ?
+
+                        -- Exclude sticky events that we in fact did just send (3)
+                        -- se.event_stream_ordering NOT IN event_stream_orderings_sent_in_transaction
+                        AND {not_event_stream_ordering_in_clause}
+
+                    GROUP BY dr.room_id
+
+                ON CONFLICT (destination, room_id)
+                DO
+                    -- Insert backlogs, unless already exists with smaller position (5)
+                    UPDATE SET sticky_events_stream_position = EXCLUDED.sticky_events_stream_position
+                    -- Only move the position *backwards*; this also prevents no-op row
+                    -- updates, avoiding needless dead tuples.
+                    WHERE EXCLUDED.sticky_events_stream_position < backlog.sticky_events_stream_position
+                """,
+                (
+                    destination,
+                    destination,
+                    old_last_successfully_sent_stream_ordering,
+                    old_last_successfully_sent_stream_ordering,
+                    new_last_successfully_sent_stream_ordering,
+                    user_is_local_like_pattern(self.hs),
+                    *not_event_stream_ordering_in_args,
+                ),
+            )
+
+        return await self.db_pool.runInteraction(
+            "mark_backlogged_sticky_events_after_catchup_transaction",
+            _txn,
+        )
+
+    async def mark_backlogged_sticky_events_sent(
+        self,
+        destination: str,
+        room_id: RoomID,
+        max_sent_sticky_events_stream_position: StickyEventStreamPosition,
+    ) -> None:
+        """
+        Marks some backlogged sticky events as sent.
+
+        The specific sticky events so marked are those in the given room,
+        with sticky event stream positions <= `max_sent_sticky_events_stream_position`.
+        """
+
+        await self.db_pool.simple_upsert(
+            desc="mark_backlogged_sticky_events_sent",
+            table="destination_rooms_sticky_events_backlog",
+            keyvalues={
+                "destination": destination,
+                "room_id": room_id.to_string(),
+            },
+            values={
+                # Add one because this is an inclusive lower bound on what's left to be sent.
+                "sticky_events_stream_position": (
+                    max_sent_sticky_events_stream_position + 1
+                ),
+            },
+        )

synapse/storage/databases/main/transactions.py

@@ -452,16 +452,19 @@ class TransactionWorkerStore(CacheInvalidationWorkerStore):
     def _get_catch_up_outstanding_destinations_txn(
         txn: LoggingTransaction, now_time_ms: int, after_destination: str | None
     ) -> list[str]:
-        # We're looking for destinations which satisfy either of the following
+        # We're looking for destinations which satisfy one of the following
         # conditions:
         #
         #   * There is at least one room where we have an event that we have not yet
         #     sent to them, indicated by a row in `destination_rooms` with a
         #     `stream_ordering` older than the `last_successful_stream_ordering`
-        #     (if any) in `destinations`, or:
+        #     (if any) in `destinations`,
         #
         #   * There is at least one to-device message outstanding for the destination,
-        #     indicated by a row in `device_federation_outbox`.
+        #     indicated by a row in `device_federation_outbox`, or
+        #
+        #   * There is at least one room where we have a sticky event backlog,
+        #     denoted by a row in `destination_rooms_sticky_events_backlog`
         #
         # Of course, that may produce destinations where we are already busy sending
         # the relevant PDU or to-device message, but in that case, waking up the
@@ -514,10 +517,23 @@ class TransactionWorkerStore(CacheInvalidationWorkerStore):
                )
             ORDER BY destination
             LIMIT 25
+        ), sticky_event_destinations AS (
+            SELECT DISTINCT destination
+            FROM destination_rooms_sticky_events_backlog
+            LEFT JOIN destinations USING (destination)
+            WHERE
+               destination > ?
+               AND (
+                    destinations.retry_last_ts IS NULL OR
+                    destinations.retry_last_ts + destinations.retry_interval < ?
+               )
+            ORDER BY destination
+            LIMIT 25
         )
 
         SELECT destination FROM pdu_destinations
         UNION SELECT destination FROM to_device_destinations
+        UNION SELECT destination FROM sticky_event_destinations
             ORDER BY destination
             LIMIT 25
         """
@@ -528,7 +544,14 @@ class TransactionWorkerStore(CacheInvalidationWorkerStore):
 
         txn.execute(
             q,
-            (after_destination, now_time_ms, after_destination, now_time_ms),
+            (
+                after_destination,
+                now_time_ms,
+                after_destination,
+                now_time_ms,
+                after_destination,
+                now_time_ms,
+            ),
         )
         destinations = [row[0] for row in txn]
 

synapse/storage/schema/main/delta/94/11_destination_room_sticky_events.sql

@@ -0,0 +1,55 @@
+--
+-- 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>.
+
+
+-- Tracks rooms that have outstanding sticky events that still need to be
+-- sent to a destination (remote homeserver).
+--
+-- Essentially a queue of unsent sticky events, but with a compact storage
+-- representation that only needs one row per (destination, room) pair.
+--
+-- Each row means: in the given room and for the given destination,
+-- we still need to send sticky events with `stream_id` at or higher than
+-- `sticky_events_stream_position`.
+--
+-- Due to sticky event expiration and event deletion, rows in this table are
+-- hints that there *may* be sticky events to send, but not a guarantee that
+-- there actually are.
+CREATE TABLE destination_rooms_sticky_events_backlog (
+    -- Server name of the remote homeserver.
+    destination TEXT NOT NULL,
+
+    -- Room ID in which sticky events have been missed.
+    room_id TEXT NOT NULL
+        -- Only track this information for rooms we know about;
+        -- if we delete a room locally then also delete the tracking info.
+        REFERENCES rooms(room_id) ON DELETE CASCADE,
+
+    -- Position in the sticky events stream, corresponding to the
+    -- `sticky_events.stream_id` of the first sticky event that
+    -- has yet to be sent.
+    --
+    -- Because sticky events must be sent in order (as per MSC4354),
+    -- all subsequent sticky events for the same room with higher
+    -- `stream_id`s are also unsent.
+    --
+    -- Not a foreign key because we must still support expiration of sticky events.
+    sticky_events_stream_position INTEGER NOT NULL,
+
+    -- It's enough to track one position per (destination, room_id) pair.
+    PRIMARY KEY (destination, room_id)
+);
+
+-- Should have this index to make `room_id` foreign key constraint efficient,
+-- as well as for cleanup per room.
+CREATE INDEX destination_rooms_sticky_events_backlog_room_id ON destination_rooms_sticky_events_backlog (room_id);

synapse/types/__init__.py

@@ -1785,3 +1785,17 @@ class EventOrderings:
         stream = event.internal_metadata.stream_ordering
         assert stream is not None
         return EventOrderings(stream, event.depth)
+
+
+def unwrap(val: T | None) -> T:
+    """
+    Assert that a value is non-None.
+
+    Analogous to `Option<T>.unwrap` in Rust.
+
+    Useful as an inline alternative to `assert val is not None`,
+    e.g. in set comprehensions.
+    """
+
+    assert val is not None
+    return val

tests/federation/test_federation_catch_up.py

@@ -1,6 +1,7 @@
+import sqlite3
 from typing import Callable, Collection
 from unittest import mock
-from unittest.mock import AsyncMock, Mock
+from unittest.mock import Mock
 
 from twisted.internet.testing import MemoryReactor
 
@@ -11,24 +12,26 @@ from synapse.federation.sender import (
     PerDestinationQueue,
     TransactionManager,
 )
+from synapse.federation.sender.per_destination_queue import _PreparedTransaction
 from synapse.federation.units import Edu, Transaction
 from synapse.rest import admin
 from synapse.rest.client import login, room
 from synapse.server import HomeServer
 from synapse.types import JsonDict
 from synapse.util.clock import Clock
+from synapse.util.duration import Duration
 from synapse.util.retryutils import NotRetryingDestination
 
 from tests.test_utils import event_injection
 from tests.unittest import FederatingHomeserverTestCase
+from tests.utils import USE_POSTGRES_FOR_TESTS
 
 
-class FederationCatchUpTestCases(FederatingHomeserverTestCase):
+class _FederationCatchUpTestCaseBase(FederatingHomeserverTestCase):
     """
-    Tests cases of catching up over federation.
-
-    By default for test cases federation sending is disabled. This Test class has it
-    re-enabled for the main process.
+    Scaffolding for a homeserver with federation sending enabled
+    and a mocked-out federation transport, so that
+    outbound transactions can be recorded (or made to fail).
     """
 
     servlets = [
@@ -44,16 +47,6 @@ class FederationCatchUpTestCases(FederatingHomeserverTestCase):
         )
 
     def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
-        # stub out get_current_hosts_in_room
-        state_storage_controller = hs.get_storage_controllers().state
-
-        # This mock is crucial for destination_rooms to be populated.
-        # TODO: this seems to no longer be the case---tests pass with this mock
-        # commented out.
-        state_storage_controller.get_current_hosts_in_room = AsyncMock(  # type: ignore[method-assign]
-            return_value={"test", "host2"}
-        )
-
         # whenever send_transaction is called, record the pdu data
         self.pdus: list[JsonDict] = []
         self.failed_pdus: list[JsonDict] = []
@@ -112,6 +105,55 @@ class FederationCatchUpTestCases(FederatingHomeserverTestCase):
         )[0]
         return {"event_id": event_id, "stream_ordering": stream_ordering}
 
+    def make_fake_destination_queue(
+        self, destination: str = "host2"
+    ) -> tuple[PerDestinationQueue, list[EventBase]]:
+        """
+        Makes a fake per-destination queue.
+        """
+        transaction_manager = TransactionManager(self.hs)
+        per_dest_queue = PerDestinationQueue(self.hs, transaction_manager, destination)
+        results_list = []
+
+        async def fake_send(
+            destination_tm: str,
+            pending_pdus: list[EventBase],
+            _pending_edus: list[Edu],
+        ) -> None:
+            assert destination == destination_tm
+            results_list.extend(pending_pdus)
+
+        transaction_manager.send_new_transaction = fake_send  # type: ignore[assignment]
+
+        return per_dest_queue, results_list
+
+    def run_transaction(
+        self, per_dest_queue: PerDestinationQueue
+    ) -> _PreparedTransaction | None:
+        """
+        Prepares and completes one transaction, returning it.
+
+        Returns None if there was no transaction to be made.
+        """
+
+        async def run() -> _PreparedTransaction | None:
+            transaction = await per_dest_queue._prepare_transaction()
+            if transaction is not None:
+                # As the transmission loop does once the transaction has been sent.
+                await per_dest_queue._complete_transaction(transaction)
+            return transaction
+
+        return self.get_success(run())
+
+
+class FederationCatchUpTestCases(_FederationCatchUpTestCaseBase):
+    """
+    Tests cases of catching up over federation.
+
+    By default, federation is disabled for test cases. `_FederationCatchUpTestCaseBase` has it
+    re-enabled for the main process.
+    """
+
     def test_catch_up_destination_rooms_tracking(self) -> None:
         """
         Tests that we populate the `destination_rooms` table as needed.
@@ -267,28 +309,6 @@ class FederationCatchUpTestCases(FederatingHomeserverTestCase):
         self.assertEqual(self.pdus[0]["content"]["body"], "hi user!")
         self.assertEqual(self.pdus[1]["content"]["body"], "wombats!")
 
-    def make_fake_destination_queue(
-        self, destination: str = "host2"
-    ) -> tuple[PerDestinationQueue, list[EventBase]]:
-        """
-        Makes a fake per-destination queue.
-        """
-        transaction_manager = TransactionManager(self.hs)
-        per_dest_queue = PerDestinationQueue(self.hs, transaction_manager, destination)
-        results_list = []
-
-        async def fake_send(
-            destination_tm: str,
-            pending_pdus: list[EventBase],
-            _pending_edus: list[Edu],
-        ) -> None:
-            assert destination == destination_tm
-            results_list.extend(pending_pdus)
-
-        transaction_manager.send_new_transaction = fake_send  # type: ignore[assignment]
-
-        return per_dest_queue, results_list
-
     def test_catch_up_loop(self) -> None:
         """
         Tests the behaviour of _catch_up_transmission_loop.
@@ -583,3 +603,346 @@ class FederationCatchUpTestCases(FederatingHomeserverTestCase):
             per_dest_queue._last_successful_stream_ordering,
             event_2.internal_metadata.stream_ordering,
         )
+
+
+class FederationStickyEventCatchUpTestCase(_FederationCatchUpTestCaseBase):
+    """
+    Tests for the catch-up of backlogged sticky events over federation.
+    """
+
+    if not USE_POSTGRES_FOR_TESTS and sqlite3.sqlite_version_info < (3, 40, 0):
+        skip = f"SQLite version is too old to support sticky events: {sqlite3.sqlite_version_info} (See https://github.com/element-hq/synapse/issues/19428)"
+
+    def default_config(self) -> JsonDict:
+        config = super().default_config()
+        config["experimental_features"] = {"msc4354_enabled": True}
+        return config
+
+    def _send_sticky(self, room_id: str, body: str, tok: str) -> str:
+        """
+        Send a sticky event.
+        """
+        return self.helper.send_sticky_event(
+            room_id,
+            EventTypes.Message,
+            duration=Duration(minutes=1),
+            content={"body": body, "msgtype": "m.text"},
+            tok=tok,
+        )["event_id"]
+
+    def _stream_ordering_for(self, event_id: str) -> int:
+        """
+        Get the `stream_ordering` for the given event.
+        """
+        event = self.get_success(self.hs.get_datastores().main.get_event(event_id))
+        stream_ordering = event.internal_metadata.stream_ordering
+        assert stream_ordering is not None
+        return stream_ordering
+
+    def _sticky_stream_id_for(self, event_id: str) -> int:
+        """
+        Get the `sticky_events` `stream_id` for the given event.
+        """
+        return self.get_success(
+            self.hs.get_datastores().main.db_pool.simple_select_one_onecol(
+                table="sticky_events",
+                keyvalues={"event_id": event_id},
+                retcol="stream_id",
+                desc="test:get_sticky_stream_id",
+            )
+        )
+
+    def _sticky_backlog_rows(self) -> list[tuple[str, str, int]]:
+        """
+        All rows of the `destination_rooms_sticky_events_backlog` table.
+        """
+        rows = self.get_success(
+            self.hs.get_datastores().main.db_pool.simple_select_list(
+                table="destination_rooms_sticky_events_backlog",
+                keyvalues=None,
+                retcols=("destination", "room_id", "sticky_events_stream_position"),
+            )
+        )
+        return sorted(rows)
+
+    def _assert_remote_host_was_last_sent_event_id(self, event_id: str) -> None:
+        """
+        Sanity-checks that `host2` was last successfully sent `event_id`, i.e. that
+        catch-up will resume from there.
+        """
+        self.assertEqual(
+            self.get_success(
+                self.hs.get_datastores().main.get_destination_last_successful_stream_ordering(
+                    "host2"
+                )
+            ),
+            self._stream_ordering_for(event_id),
+            "test fault: host2 is not up to date as of the expected event",
+        )
+
+    def test_catch_up_records_skipped_sticky_events_as_backlogged(self) -> None:
+        """
+        Tests that 'regular' federation catch-up marks the sticky events in the gaps
+        that it skips.
+
+        See the docstring on `mark_backlogged_sticky_events_after_catchup_transaction`
+        for a diagrammatical explanation.
+        """
+
+        per_dest_queue, sent_pdus = self.make_fake_destination_queue()
+        # The queue starts up in 'sticky event backlogged' mode, but we want our test to cover
+        # the steady-state. So clear the backlogged flag and allow it to be set naturally.
+        per_dest_queue._sticky_event_backlog_tracker._backlogged = False
+
+        # Create a local user and 3 rooms
+        self.register_user("u1", "you the one")
+        u1_token = self.login("u1", "you the one")
+        room_1 = self.helper.create_room_as("u1", tok=u1_token)
+        room_2 = self.helper.create_room_as("u1", tok=u1_token)
+        room_3 = self.helper.create_room_as("u1", tok=u1_token)
+        # Make a remote user join each of the rooms, so that we consider them
+        # for sending events over federation
+        for room_id in (room_1, room_2, room_3):
+            self.get_success(
+                event_injection.inject_member_event(
+                    self.hs, room_id, "@user:host2", "join"
+                )
+            )
+
+        # Set up some events according to this sequence:
+        #
+        #         stream_ordering ------------------------------->
+        # room 1    •                  •
+        #      event_id_1        sticky_id_5
+        # room 2      •          •
+        #        event_id_2  event_id_4
+        # room 3        •                  •
+        #           sticky_id_3        event_id_6
+
+        (_event_id_1,) = self.helper.send_messages(room_1, num_events=1, tok=u1_token)
+        (event_id_2,) = self.helper.send_messages(room_2, num_events=1, tok=u1_token)
+        # Let the federation sender act on the events above.
+        self.reactor.advance(0)
+
+        # Put host2 down, to prevent the federation sender from succeeding in sending the
+        # next events out (and to trigger catch-up mode).
+        self.is_online = False
+        self._assert_remote_host_was_last_sent_event_id(event_id_2)
+
+        # This sticky event will fall in the gap that the first catch-up transaction
+        # skips over, so it must end up backlogged.
+        sticky_id_3 = self._send_sticky(room_3, "sticky in the gap", u1_token)
+        (event_id_4,) = self.helper.send_messages(room_2, num_events=1, tok=u1_token)
+        # This event will be a forward extremity itself at the time of catch-up,
+        # so it will get sent as part of regular catch-up and must _not_ be backlogged.
+        sticky_id_5 = self._send_sticky(room_1, "sticky extremity", u1_token)
+        (event_id_6,) = self.helper.send_messages(room_3, num_events=1, tok=u1_token)
+        # Advance for the federation sender to trigger on those newly-sent events
+        self.reactor.advance(0)
+
+        # Now trigger a catch-up loop manually
+        # We do this explicitly instead of `attempt_new_transaction` because we don't want a
+        # regular transaction to take place afterwards, which might muddy our test.
+        self.get_success(per_dest_queue._catch_up_transmission_loop())
+
+        # First sanity-check what got sent and what the state of the 'regular'
+        # catch-up is.
+        # Each room's latest local event (forward extremity) was sent, oldest first.
+        # (This is because catch-up transactions only send forward extremities where possible.)
+        self.assertEqual(
+            [pdu.event_id for pdu in sent_pdus],
+            [event_id_4, sticky_id_5, event_id_6],
+        )
+        self.assertFalse(per_dest_queue._catching_up, "should have completed catch-up")
+        self._assert_remote_host_was_last_sent_event_id(event_id_6)
+
+        # Check the state of the sticky event backlog
+        # - `sticky_id_3` fell in the gap that got skipped over by the catch-up
+        #   transaction for room 2 (as room 2 had the oldest forward extremity
+        #   of all the rooms.)
+        #   So room 3 is backlogged.
+        # - `sticky_id_5` got sent as a forward extremity, so room 1 is _not_
+        #   backlogged.
+        self.assertEqual(
+            self._sticky_backlog_rows(),
+            [("host2", room_3, self._sticky_stream_id_for(sticky_id_3))],
+        )
+        self.assertTrue(
+            per_dest_queue._sticky_event_backlog_tracker.is_backlogged,
+            "the queue must notice the backlog that it has just recorded",
+        )
+
+    def test_backlogged_sticky_events_are_sent_in_dedicated_transactions(self) -> None:
+        """
+        Tests the dedicated transactions that send backlogged sticky events.
+        """
+        per_dest_queue, _sent_pdus = self.make_fake_destination_queue()
+
+        # Create a local user and a room
+        self.register_user("u1", "you the one")
+        u1_token = self.login("u1", "you the one")
+        room_id = self.helper.create_room_as("u1", tok=u1_token)
+        self.get_success(
+            event_injection.inject_member_event(self.hs, room_id, "@user:host2", "join")
+        )
+
+        # Send whilst host2 is up, so this is where catch-up will resume from later on
+        (already_sent_id,) = self.helper.send_messages(
+            room_id, num_events=1, tok=u1_token
+        )
+        # Trigger federation sender
+        self.reactor.advance(0)
+
+        # Put host2 down, to prevent the federation sender from succeeding in sending the
+        # next events out (and to trigger catch-up mode).
+        self.is_online = False
+        self._assert_remote_host_was_last_sent_event_id(already_sent_id)
+
+        sticky_id_1 = self._send_sticky(room_id, "sticky 1", u1_token)
+        sticky_id_2 = self._send_sticky(room_id, "sticky 2", u1_token)
+        # Send an event to be the room's forward extremity
+        # This will make catch-up skip over the 2 sticky events so we can use those 2
+        # to test sticky event backlog catch-up
+        self.helper.send_messages(room_id, num_events=1, tok=u1_token)
+        # Trigger federation sender
+        self.reactor.advance(0)
+
+        # Do a 'regular' catch-up transaction.
+        # (As a reminder, regular catch-up transactions only send the forward extremities.)
+        # This is also what records the sticky event backlog
+        self.get_success(per_dest_queue._catch_up_transmission_loop())
+        self.assertEqual(
+            self._sticky_backlog_rows(),
+            [("host2", room_id, self._sticky_stream_id_for(sticky_id_1))],
+        )
+
+        # Build a transaction. Since the main queue has nothing to send,
+        # this is a sticky event backlog transaction.
+        transaction = self.run_transaction(per_dest_queue)
+        assert transaction is not None
+        self.assertEqual(
+            [pdu.event_id for pdu in transaction.pdus], [sticky_id_1, sticky_id_2]
+        )
+        self.assertEqual(transaction.edus, [])
+
+        # The position should have moved to just past the last one.
+        self.assertEqual(
+            self._sticky_backlog_rows(),
+            [("host2", room_id, self._sticky_stream_id_for(sticky_id_2) + 1)],
+        )
+
+        # Try to build a transaction, but get None as neither the main queue
+        # nor the sticky event backlog should have anything to send
+        self.assertIsNone(self.run_transaction(per_dest_queue))
+        self.assertEqual(self._sticky_backlog_rows(), [])
+        self.assertFalse(per_dest_queue._sticky_event_backlog_tracker.is_backlogged)
+
+    def test_backlogged_sticky_events_are_drained_by_the_same_loop(self) -> None:
+        """
+        Tests that a sticky event backlog recorded by catch-up is drained by the same
+        transmission loop, rather than sitting there until unrelated traffic to the
+        destination happens to start a new one.
+        """
+        per_dest_queue, sent_pdus = self.make_fake_destination_queue()
+        # The queue starts up in 'sticky event backlogged' mode, but we want our test to cover
+        # the steady-state. So clear the backlogged flag and allow it to be set naturally.
+        per_dest_queue._sticky_event_backlog_tracker._backlogged = False
+
+        self.register_user("u1", "you the one")
+        u1_token = self.login("u1", "you the one")
+        room_id = self.helper.create_room_as("u1", tok=u1_token)
+        self.get_success(
+            event_injection.inject_member_event(self.hs, room_id, "@user:host2", "join")
+        )
+
+        # Send an event whilst host2 is up, so this is where catch-up will resume from.
+        (already_sent_id,) = self.helper.send_messages(
+            room_id, num_events=1, tok=u1_token
+        )
+        self.reactor.advance(0)
+
+        # Put host2 down
+        self.is_online = False
+        self._assert_remote_host_was_last_sent_event_id(already_sent_id)
+
+        sticky_id = self._send_sticky(room_id, "backlogged sticky", u1_token)
+        # The room's latest event, so catch-up sends this one and skips over the sticky
+        # event, leaving it backlogged.
+        (latest_id,) = self.helper.send_messages(room_id, num_events=1, tok=u1_token)
+        # Trigger the federation sender to fail to send the events and then
+        # to go into catch-up mode
+        self.reactor.advance(0)
+
+        # Trigger regular federation catch-up and then the sticky event catch-up
+        # (2 transactions, without having to retrigger the loop by calling
+        # `attempt_new_transaction` again)
+        per_dest_queue.attempt_new_transaction()
+        assert per_dest_queue.active_transmission_loop is not None
+        self.get_success(per_dest_queue.active_transmission_loop)
+
+        # Catch-up sent the room's latest event, then the backlog transaction sent the
+        # sticky event that catch-up had skipped.
+        self.assertEqual([pdu.event_id for pdu in sent_pdus], [latest_id, sticky_id])
+        self.assertEqual(self._sticky_backlog_rows(), [])
+        self.assertFalse(per_dest_queue._sticky_event_backlog_tracker.is_backlogged)
+
+    def test_backlogged_sticky_events_do_not_delay_pending_pdus(self) -> None:
+        """
+        Tests that the normal federation transmission queue gets the opportunity
+        to send main queue (normal) transactions in between sticky event backlog transactions.
+        """
+        per_dest_queue, _sent_pdus = self.make_fake_destination_queue()
+
+        self.register_user("u1", "you the one")
+        u1_token = self.login("u1", "you the one")
+        room_id = self.helper.create_room_as("u1", tok=u1_token)
+        self.get_success(
+            event_injection.inject_member_event(self.hs, room_id, "@user:host2", "join")
+        )
+
+        # Sent whilst host2 is up, so this is where catch-up will resume from.
+        (already_sent_id,) = self.helper.send_messages(
+            room_id, num_events=1, tok=u1_token
+        )
+        # Let the federation sender act on the events above.
+        self.reactor.advance(0)
+        self.is_online = False
+        self._assert_remote_host_was_last_sent_event_id(already_sent_id)
+
+        # Stands in for a PDU still awaiting real-time delivery; queued by hand below.
+        (realtime_id,) = self.helper.send_messages(room_id, num_events=1, tok=u1_token)
+        sticky_id = self._send_sticky(room_id, "backlogged sticky", u1_token)
+        # The room's latest event, so catch-up sends this one and skips over the
+        # sticky event, leaving it backlogged.
+        self.helper.send_messages(room_id, num_events=1, tok=u1_token)
+        # Let the federation sender act on the events above.
+        self.reactor.advance(0)
+
+        # Catching up is what records the backlog.
+        self.get_success(per_dest_queue._catch_up_transmission_loop())
+        self.assertEqual(
+            self._sticky_backlog_rows(),
+            [("host2", room_id, self._sticky_stream_id_for(sticky_id))],
+        )
+
+        realtime_event = self.get_success(
+            self.hs.get_datastores().main.get_event(realtime_id)
+        )
+
+        # Manually append to `_pending_pdus` rather than calling `send_pdu`, because
+        # `send_pdu` will trigger `attempt_new_transaction()`,
+        # but we want to be in control here so we can inspect each transaction
+        # individually.
+        per_dest_queue._pending_pdus[realtime_event.event_id] = realtime_event
+
+        # Transaction 1 (normal) sends the real-time queued PDU...
+        transaction = self.run_transaction(per_dest_queue)
+        assert transaction is not None
+        self.assertEqual([pdu.event_id for pdu in transaction.pdus], [realtime_id])
+        # (...and it is removed from the queue once the transaction completes)
+        self.assertEqual(list(per_dest_queue._pending_pdus), [])
+
+        # Transaction 2 sends the sticky event backlog
+        transaction = self.run_transaction(per_dest_queue)
+        assert transaction is not None
+        self.assertEqual([pdu.event_id for pdu in transaction.pdus], [sticky_id])

tests/storage/test_sticky_events.py

@@ -23,15 +23,20 @@ from synapse.api.constants import (
     StickyEventField,
 )
 from synapse.api.room_versions import RoomVersions
+from synapse.replication.tcp.streams._base import StickyEventStreamPosition
 from synapse.rest import admin
 from synapse.rest.client import login, register, room, sync
 from synapse.server import HomeServer
-from synapse.types import JsonDict, create_requester
+from synapse.types import JsonDict, RoomID, create_requester
 from synapse.util.clock import Clock
 from synapse.util.duration import Duration
 
 from tests import unittest
-from tests.test_utils.event_injection import create_event, inject_event
+from tests.test_utils.event_injection import (
+    create_event,
+    inject_event,
+    inject_member_event,
+)
 from tests.utils import USE_POSTGRES_FOR_TESTS
 
 
@@ -636,3 +641,570 @@ class StickyEventsTestCase(unittest.HomeserverTestCase):
         # ...and the soft-failure has been cleared from the event itself.
         event = self.get_success(self.store.get_event(event_id))
         self.assertFalse(event.internal_metadata.is_soft_failed())
+
+
+class StickyEventsFederationBacklogTestCase(unittest.HomeserverTestCase):
+    """
+    Storage-level tests for the federation sticky event backlog mechanism.
+
+    This mechanism is used to catch a destination up on sticky events that were skipped over by a
+    federation catch-up transaction.
+    """
+
+    if not USE_POSTGRES_FOR_TESTS and sqlite3.sqlite_version_info < (3, 40, 0):
+        skip = f"SQLite version is too old to support sticky events: {sqlite3.sqlite_version_info} (See https://github.com/element-hq/synapse/issues/19428)"
+
+    servlets = [
+        room.register_servlets,
+        login.register_servlets,
+        register.register_servlets,
+        admin.register_servlets,
+    ]
+
+    def default_config(self) -> JsonDict:
+        config = super().default_config()
+        config["experimental_features"] = {"msc4354_enabled": True}
+        return config
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        self.store = self.hs.get_datastores().main
+
+        # Register an account and create a room
+        self.user_id = self.register_user("user", "pass")
+        self.token = self.login(self.user_id, "pass")
+
+    def _send_sticky(self, room_id: str, body: str) -> str:
+        """
+        Send a sticky event.
+        """
+        return self.helper.send_sticky_event(
+            room_id,
+            EventTypes.Message,
+            duration=Duration(minutes=1),
+            content={"body": body, "msgtype": "m.text"},
+            tok=self.token,
+        )["event_id"]
+
+    def _stream_ordering_for(self, event_id: str) -> int:
+        """
+        Get the `stream_ordering` for the given event.
+        """
+        event = self.get_success(self.hs.get_datastores().main.get_event(event_id))
+        stream_ordering = event.internal_metadata.stream_ordering
+        assert stream_ordering is not None
+        return stream_ordering
+
+    def _sticky_stream_id_for(self, event_id: str) -> int:
+        """
+        Get the `sticky_events` `stream_id` for the given event.
+        """
+        return self.get_success(
+            self.hs.get_datastores().main.db_pool.simple_select_one_onecol(
+                table="sticky_events",
+                keyvalues={"event_id": event_id},
+                retcol="stream_id",
+                desc="test:get_sticky_stream_id",
+            )
+        )
+
+    def _sticky_backlog_rows(self) -> set[tuple[str, str, int]]:
+        """
+        All rows of the `destination_rooms_sticky_events_backlog` table,
+        as a set: row order is arbitrary, as the query has no ORDER BY.
+        """
+        rows = self.get_success(
+            self.hs.get_datastores().main.db_pool.simple_select_list(
+                table="destination_rooms_sticky_events_backlog",
+                keyvalues=None,
+                retcols=("destination", "room_id", "sticky_events_stream_position"),
+            )
+        )
+        return set(rows)
+
+    def test_mark_backlogged_after_catchup_records_earliest_unsent_per_room(
+        self,
+    ) -> None:
+        """
+        Tests that a catch-up transaction that advances over a gap records,
+        per room and per destination, one row for every earliest sticky event
+        left unsent in the gap.
+
+        See the docstring on `mark_backlogged_sticky_events_after_catchup_transaction`
+        for a diagrammatical description.
+        """
+        room1 = self.helper.create_room_as(self.user_id, tok=self.token)
+        room2 = self.helper.create_room_as(self.user_id, tok=self.token)
+
+        # The event immediately before the gap.
+        # Suppose that this is where the destination had
+        # successfully been caught up to.
+        before_gap_event_id = self.helper.send(room1, "before the gap", tok=self.token)[
+            "event_id"
+        ]
+
+        # Send 2 sticky events into room1
+        room1_sticky1_event_id = self._send_sticky(room1, "sticky 1")
+        _room1_sticky2_event_id = self._send_sticky(room1, "sticky 2")
+
+        # Send a sticky event into room2
+        room2_sticky3_event_id = self._send_sticky(room2, "sticky 3")
+
+        # Send a couple of events for the the catch-up transaction to advance us to.
+        (room1_after_gap_event_id,) = self.helper.send_messages(
+            room1, 1, tok=self.token
+        )
+        (room2_after_gap_event_id,) = self.helper.send_messages(
+            room2, 1, tok=self.token
+        )
+
+        # We store destination rooms entries for those:
+        # this is how the outstanding events to be sent are tracked.
+        # It's also a necessary prerequisite for the backlog marking calculation.
+        self.get_success(
+            self.store.store_destination_rooms_entries(
+                {"host2"}, room1, self._stream_ordering_for(room1_after_gap_event_id)
+            )
+        )
+        self.get_success(
+            self.store.store_destination_rooms_entries(
+                {"host2"}, room2, self._stream_ordering_for(room2_after_gap_event_id)
+            )
+        )
+
+        # Now suppose we sent the first catch-up transaction (for room1, since the forward extremity
+        # of room1 is the oldest catch-up forward extremity in our database).
+        # This creates a gap of unsent sticky events that need to be caught up.
+        self.get_success(
+            self.store.mark_backlogged_sticky_events_after_catchup_transaction(
+                "host2",
+                old_last_successfully_sent_stream_ordering=self._stream_ordering_for(
+                    before_gap_event_id
+                ),
+                new_last_successfully_sent_stream_ordering=self._stream_ordering_for(
+                    room1_after_gap_event_id
+                ),
+                event_stream_orderings_sent_in_transaction={
+                    self._stream_ordering_for(room1_after_gap_event_id)
+                },
+            )
+        )
+
+        self.assertEqual(
+            self._sticky_backlog_rows(),
+            {
+                # In room1: we need to catch up from the first sticky event
+                ("host2", room1, self._sticky_stream_id_for(room1_sticky1_event_id)),
+                # In room2: we need to catch up from the first sticky event in that room
+                ("host2", room2, self._sticky_stream_id_for(room2_sticky3_event_id)),
+            },
+        )
+
+    def test_mark_backlogged_after_catchup_ignores_events_outside_the_gap(self) -> None:
+        """
+        Tests that when marking a backlog, we ignore:
+            - sticky events outside the gap
+            - non-sticky events inside the gap
+        """
+        room_id = self.helper.create_room_as(self.user_id, tok=self.token)
+
+        # Send a sticky event. Suppose we had already delivered this one to the destination.
+        event1_id_sticky = self._send_sticky(room_id, "already sent")
+
+        # Send 2 regular events that we suppose we had _not_ delivered to the destination yet.
+        (_event2_id_nonsticky, _event3_id_nonsticky) = self.helper.send_messages(
+            room_id, 2, tok=self.token
+        )
+        # Send a sticky event. This is the one we'll treat as the forward extremity
+        event4_id_sticky = self._send_sticky(room_id, "not yet due to be sent")
+
+        # Note down that we have events up to `event4_sticky` that need to be
+        # sent out.
+        self.get_success(
+            self.store.store_destination_rooms_entries(
+                {"host2"}, room_id, self._stream_ordering_for(event4_id_sticky)
+            )
+        )
+
+        # Suppose we sent a catch-up transaction with `event4_sticky`,
+        self.get_success(
+            self.store.mark_backlogged_sticky_events_after_catchup_transaction(
+                "host2",
+                old_last_successfully_sent_stream_ordering=self._stream_ordering_for(
+                    event1_id_sticky
+                ),
+                new_last_successfully_sent_stream_ordering=self._stream_ordering_for(
+                    event4_id_sticky
+                ),
+                # Really we 'should' put `event4_sticky` here to match reality
+                # However we're interested in testing the range being correct without
+                # the set difference operation covering up any mistakes.
+                event_stream_orderings_sent_in_transaction=set(),
+            )
+        )
+
+        # There should be no backlog of unsent sticky events tracked,
+        # because there were none in the gap.
+        self.assertEqual(self._sticky_backlog_rows(), set())
+
+    def test_mark_backlogged_after_catchup_keeps_earliest_position(self) -> None:
+        """
+        Tests that repeated catch-up transactions do not advance the backlog position
+        (as that would lose sticky events in the gap).
+        """
+        room_id = self.helper.create_room_as(self.user_id, tok=self.token)
+
+        # First of all, suppose we had already sent out an event
+        event1_start_id = self.helper.send(room_id, "gap start", tok=self.token)[
+            "event_id"
+        ]
+        # then send a sticky event that we will lose in the gap
+        event2_sticky_id = self._send_sticky(room_id, "early sticky")
+        # Send a 'middle' event that we will send out in a catch-up transaction
+        event3_middle_id = self.helper.send(room_id, "middle", tok=self.token)[
+            "event_id"
+        ]
+
+        self.get_success(
+            self.store.store_destination_rooms_entries(
+                {"host2"}, room_id, self._stream_ordering_for(event3_middle_id)
+            )
+        )
+
+        # We get a catch-up transaction sent out with `middle` in it.
+        # This creates a gap of unsent sticky events.
+        self.get_success(
+            self.store.mark_backlogged_sticky_events_after_catchup_transaction(
+                "host2",
+                old_last_successfully_sent_stream_ordering=self._stream_ordering_for(
+                    event1_start_id
+                ),
+                new_last_successfully_sent_stream_ordering=self._stream_ordering_for(
+                    event3_middle_id
+                ),
+                event_stream_orderings_sent_in_transaction={
+                    self._stream_ordering_for(event3_middle_id)
+                },
+            )
+        )
+        self.assertEqual(
+            self._sticky_backlog_rows(),
+            {("host2", room_id, self._sticky_stream_id_for(event2_sticky_id))},
+        )
+
+        # Now send another sticky event and 'lose' it in a gap again.
+        # Send a final event that we will send out in a catch-up transaction
+        # (in order to create a gap for this event to sit in)
+        event5_end_id = self.helper.send(room_id, "gap end", tok=self.token)["event_id"]
+
+        self.get_success(
+            self.store.store_destination_rooms_entries(
+                {"host2"}, room_id, self._stream_ordering_for(event5_end_id)
+            )
+        )
+
+        self.get_success(
+            self.store.mark_backlogged_sticky_events_after_catchup_transaction(
+                "host2",
+                old_last_successfully_sent_stream_ordering=self._stream_ordering_for(
+                    event3_middle_id
+                ),
+                new_last_successfully_sent_stream_ordering=self._stream_ordering_for(
+                    event5_end_id
+                ),
+                event_stream_orderings_sent_in_transaction={
+                    self._stream_ordering_for(event5_end_id)
+                },
+            )
+        )
+
+        # We should find that the backlog still starts at `event2_sticky`,
+        # because it's the earliest sticky event that needs catching up.
+        self.assertEqual(
+            self._sticky_backlog_rows(),
+            {("host2", room_id, self._sticky_stream_id_for(event2_sticky_id))},
+        )
+
+    def test_get_backlogged_sticky_events_returns_none_when_no_backlog(self) -> None:
+        """
+        Tests that `get_backlogged_sticky_events` returns `None` when
+        there is nothing to catch up on (empty backlog table).
+        """
+        # Make a room with a sticky event that we intend to send to
+        # the destination
+        room_id = self.helper.create_room_as(self.user_id, tok=self.token)
+        event_id = self._send_sticky(room_id, "sticky")
+
+        # This notes our intention to send the event to the destination
+        # But it's not a sticky event backlog yet, just part of the regular
+        # event flow
+        self.get_success(
+            self.store.store_destination_rooms_entries(
+                {"host2"}, room_id, self._stream_ordering_for(event_id)
+            )
+        )
+
+        # So there should be no backlog
+        self.assertIsNone(
+            self.get_success(
+                self.store.get_backlogged_sticky_events_for_destination("host2")
+            )
+        )
+
+    def test_get_backlogged_sticky_events_returns_local_events_in_stream_order(
+        self,
+    ) -> None:
+        """
+        Tests that `get_backlogged_sticky_events` returns sticky events in stream order.
+        """
+        room_id = self.helper.create_room_as(self.user_id, tok=self.token)
+
+        _sticky_1_event_id = self._send_sticky(room_id, "sticky 1")
+        sticky_2_event_id = self._send_sticky(room_id, "sticky 2")
+        sticky_3_event_id = self._send_sticky(room_id, "sticky 3")
+        sticky_4_event_id = self._send_sticky(room_id, "sticky 4")
+
+        # Pretend a catch-up transaction left a gap of unsent sticky events,
+        # starting from sticky_2 onwards.
+        self.get_success(
+            self.store.db_pool.simple_insert(
+                table="destination_rooms_sticky_events_backlog",
+                values={
+                    "destination": "host2",
+                    "room_id": room_id,
+                    "sticky_events_stream_position": self._sticky_stream_id_for(
+                        sticky_2_event_id
+                    ),
+                },
+                desc="test:insert_backlog",
+            )
+        )
+
+        result = self.get_success(
+            self.store.get_backlogged_sticky_events_for_destination("host2")
+        )
+
+        self.assertEqual(
+            result,
+            (
+                RoomID.from_string(room_id),
+                self._sticky_stream_id_for(sticky_4_event_id),
+                # We see sticky 2 events up to and including 4, in that order
+                [sticky_2_event_id, sticky_3_event_id, sticky_4_event_id],
+            ),
+        )
+
+    def test_get_backlogged_sticky_events_respects_limit(self) -> None:
+        """
+        Tests that `get_backlogged_sticky_events` respects the limit.
+        """
+        room_id = self.helper.create_room_as(self.user_id, tok=self.token)
+
+        sticky_1_event_id = self._send_sticky(room_id, "sticky 1")
+        sticky_2_event_id = self._send_sticky(room_id, "sticky 2")
+        _sticky_3_event_id = self._send_sticky(room_id, "sticky 3")
+
+        # Pretend a catch-up transaction left a gap of unsent sticky events,
+        # starting from sticky_1 onwards.
+        self.get_success(
+            self.store.db_pool.simple_insert(
+                table="destination_rooms_sticky_events_backlog",
+                values={
+                    "destination": "host2",
+                    "room_id": room_id,
+                    "sticky_events_stream_position": self._sticky_stream_id_for(
+                        sticky_1_event_id
+                    ),
+                },
+                desc="test:insert_backlog",
+            )
+        )
+
+        self.assertEqual(
+            self.get_success(
+                self.store.get_backlogged_sticky_events_for_destination(
+                    "host2", limit=2
+                )
+            ),
+            (
+                RoomID.from_string(room_id),
+                # We get the sticky event stream ID of sticky_2 as that's the last one we received
+                # in this window
+                self._sticky_stream_id_for(sticky_2_event_id),
+                # We limit to 2 so we don't see sticky_3 here
+                [sticky_1_event_id, sticky_2_event_id],
+            ),
+        )
+
+    def test_get_backlogged_sticky_events_excludes_remote_senders(self) -> None:
+        """
+        Tests that we only consider our own (locally-sent) sticky events as
+        eligible for backlog catch-up.
+
+        > As with regular events, servers are only responsible for sending sticky events originating from their own server.
+        > — https://github.com/matrix-org/matrix-spec-proposals/blob/kegan/persist-edu/proposals/4354-sticky-events.md?plain=1#L195C1-L195C114
+        """
+        room_id = self.helper.create_room_as(self.user_id, tok=self.token)
+        self.get_success(
+            inject_member_event(self.hs, room_id, "@remote:host3", Membership.JOIN)
+        )
+
+        remote_sticky_event_id = self.get_success(
+            inject_event(
+                self.hs,
+                room_id=room_id,
+                sender="@remote:host3",
+                type=EventTypes.Message,
+                content={"body": "remote sticky", "msgtype": "m.text"},
+                # Corresponds to StickyEvent.EVENT_FIELD_NAME
+                msc4354_sticky=StickyEventField(
+                    duration_ms=Duration(minutes=1).as_millis()
+                ),
+            )
+        ).event_id
+        local_sticky_event_id = self._send_sticky(room_id, "local sticky")
+
+        # Sanity check our test: the remote event _is_ in the sticky events table.
+        self.assertEqual(
+            set(
+                self.get_success(
+                    self.store.db_pool.simple_select_onecol(
+                        table="sticky_events",
+                        keyvalues={"room_id": room_id},
+                        retcol="event_id",
+                        desc="test:all_sticky_event_ids",
+                    )
+                )
+            ),
+            {remote_sticky_event_id, local_sticky_event_id},
+        )
+
+        assert self._sticky_stream_id_for(
+            remote_sticky_event_id
+        ) < self._sticky_stream_id_for(local_sticky_event_id)
+
+        self.get_success(
+            self.store.db_pool.simple_insert(
+                table="destination_rooms_sticky_events_backlog",
+                values={
+                    "destination": "host2",
+                    "room_id": room_id,
+                    "sticky_events_stream_position": self._sticky_stream_id_for(
+                        remote_sticky_event_id
+                    ),
+                },
+                desc="test:insert_backlog",
+            )
+        )
+
+        self.assertEqual(
+            self.get_success(
+                self.store.get_backlogged_sticky_events_for_destination("host2")
+            ),
+            (
+                RoomID.from_string(room_id),
+                self._sticky_stream_id_for(local_sticky_event_id),
+                [local_sticky_event_id],
+            ),
+        )
+
+    def test_get_backlogged_sticky_events_cleans_up_stale_backlog(self) -> None:
+        """
+        Tests that backlog rows are removed on-demand when it turns out there are
+        no unexpired sticky events remaining.
+
+        The backlog row is essentially just a 'hint' that there might be sticky events
+        left to send, but not a guarantee (due to expiry).
+        """
+        room_id = self.helper.create_room_as(self.user_id, tok=self.token)
+
+        # A sticky event that expires almost immediately.
+        short_lived_event_id = self.helper.send_sticky_event(
+            room_id,
+            EventTypes.Message,
+            duration=Duration(milliseconds=1),
+            content={"body": "short lived", "msgtype": "m.text"},
+            tok=self.token,
+        )["event_id"]
+
+        self.get_success(
+            self.store.db_pool.simple_insert(
+                table="destination_rooms_sticky_events_backlog",
+                values={
+                    "destination": "host2",
+                    "room_id": room_id,
+                    "sticky_events_stream_position": self._sticky_stream_id_for(
+                        short_lived_event_id
+                    ),
+                },
+                desc="test:insert_backlog",
+            )
+        )
+
+        # Advance the reactor and trigger the deletion of expired sticky events
+        self.reactor.advance(0.002)
+        self.get_success(self.store._delete_expired_sticky_events())
+
+        self.assertIsNone(
+            self.get_success(
+                self.store.get_backlogged_sticky_events_for_destination("host2")
+            )
+        )
+
+        # Also note that the `destination_rooms_sticky_events_backlog` has been cleared
+        # so that we don't keep reconsidering this room that no longer has any
+        # unexpired sticky events to be sent.
+        self.assertEqual(self._sticky_backlog_rows(), set())
+
+    def test_mark_backlogged_sticky_events_sent_advances_position(self) -> None:
+        """
+        Marking a batch as sent moves the recorded position to just *after* the
+        highest sent position, so the next batch starts with the first unsent event.
+        """
+        room_id = self.helper.create_room_as(self.user_id, tok=self.token)
+
+        sticky_1_event_id = self._send_sticky(room_id, "sticky 1")
+        sticky_2_event_id = self._send_sticky(room_id, "sticky 2")
+        sticky_3_event_id = self._send_sticky(room_id, "sticky 3")
+
+        self.get_success(
+            self.store.db_pool.simple_insert(
+                table="destination_rooms_sticky_events_backlog",
+                values={
+                    "destination": "host2",
+                    "room_id": room_id,
+                    "sticky_events_stream_position": self._sticky_stream_id_for(
+                        sticky_1_event_id
+                    ),
+                },
+                desc="test:insert_backlog",
+            )
+        )
+
+        self.get_success(
+            self.store.mark_backlogged_sticky_events_sent(
+                "host2",
+                RoomID.from_string(room_id),
+                StickyEventStreamPosition(
+                    self._sticky_stream_id_for(sticky_2_event_id)
+                ),
+            )
+        )
+
+        # The stored position is an *inclusive lower bound on what is left*, hence
+        # exactly one past the highest event we sent.
+        self.assertEqual(
+            self._sticky_backlog_rows(),
+            {("host2", room_id, self._sticky_stream_id_for(sticky_2_event_id) + 1)},
+        )
+
+        # And the next batch is just the remaining event.
+        self.assertEqual(
+            self.get_success(
+                self.store.get_backlogged_sticky_events_for_destination("host2")
+            ),
+            (
+                RoomID.from_string(room_id),
+                self._sticky_stream_id_for(sticky_3_event_id),
+                [sticky_3_event_id],
+            ),
+        )