diff --git a/changelog.d/20165.feature b/changelog.d/20165.feature new file mode 100644 index 0000000000..85763e9af0 --- /dev/null +++ b/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. \ No newline at end of file diff --git a/synapse/federation/sender/per_destination_queue.py b/synapse/federation/sender/per_destination_queue.py index 4c0918e0ac..02e69bf778 100644 --- a/synapse/federation/sender/per_destination_queue.py +++ b/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 diff --git a/synapse/replication/tcp/streams/_base.py b/synapse/replication/tcp/streams/_base.py index f9c7821a39..442be70daa 100644 --- a/synapse/replication/tcp/streams/_base.py +++ b/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.""" diff --git a/synapse/storage/database.py b/synapse/storage/database.py index 469639fbf3..3db4070463 100644 --- a/synapse/storage/database.py +++ b/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}" diff --git a/synapse/storage/databases/main/event_push_actions.py b/synapse/storage/databases/main/event_push_actions.py index afbbe6b61a..7cecbcb65c 100644 --- a/synapse/storage/databases/main/event_push_actions.py +++ b/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, diff --git a/synapse/storage/databases/main/metrics.py b/synapse/storage/databases/main/metrics.py index b2b4561247..90b018073e 100644 --- a/synapse/storage/databases/main/metrics.py +++ b/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 diff --git a/synapse/storage/databases/main/purge_events.py b/synapse/storage/databases/main/purge_events.py index 1accf207be..f564d3322a 100644 --- a/synapse/storage/databases/main/purge_events.py +++ b/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) diff --git a/synapse/storage/databases/main/sticky_events.py b/synapse/storage/databases/main/sticky_events.py index bc2c2e67e4..1ff584a4a7 100644 --- a/synapse/storage/databases/main/sticky_events.py +++ b/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 + ), + }, + ) diff --git a/synapse/storage/databases/main/transactions.py b/synapse/storage/databases/main/transactions.py index a3d75002b9..023b338d24 100644 --- a/synapse/storage/databases/main/transactions.py +++ b/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] diff --git a/synapse/storage/schema/main/delta/94/11_destination_room_sticky_events.sql b/synapse/storage/schema/main/delta/94/11_destination_room_sticky_events.sql new file mode 100644 index 0000000000..3b51b41a3c --- /dev/null +++ b/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: +-- . + + +-- 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); diff --git a/synapse/types/__init__.py b/synapse/types/__init__.py index 90d6c9f32f..599ed5e3e4 100644 --- a/synapse/types/__init__.py +++ b/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.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 diff --git a/tests/federation/test_federation_catch_up.py b/tests/federation/test_federation_catch_up.py index fd1ef043bb..95a4fa57c1 100644 --- a/tests/federation/test_federation_catch_up.py +++ b/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]) diff --git a/tests/storage/test_sticky_events.py b/tests/storage/test_sticky_events.py index c73da479f4..7058344123 100644 --- a/tests/storage/test_sticky_events.py +++ b/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], + ), + )