diff --git a/changelog.d/20165.feature b/changelog.d/20165.feature new file mode 100644 index 00000000000..85763e9af01 --- /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/changelog.d/20166.misc b/changelog.d/20166.misc new file mode 100644 index 00000000000..716e910f8ff --- /dev/null +++ b/changelog.d/20166.misc @@ -0,0 +1 @@ +Refactor the federation transmission code to delineate transaction preparation and completion. \ 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 32f8630c9d1..a33579b97b5 100644 --- a/synapse/federation/sender/per_destination_queue.py +++ b/synapse/federation/sender/per_destination_queue.py @@ -22,7 +22,6 @@ import datetime import logging from collections import OrderedDict -from types import TracebackType from typing import TYPE_CHECKING, Hashable, Iterable import attr @@ -48,7 +47,8 @@ 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 @@ -79,6 +79,76 @@ MAX_PRESENCE_STATES_PER_EDU = 50 +@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: + """ + A transaction that has been prepared for sending: what to send, along with the + information that is useful for marking the transaction as complete once it has + been successfully sent. + + Produced by `PerDestinationQueue._prepare_transaction` and consumed by + `PerDestinationQueue._complete_transaction`. + """ + + pdus: list[EventBase] + """ + The PDUs to send in this transaction. + """ + + edus: list[Edu] + """ + The EDUs to send in this transaction. + """ + + to_device_message_stream_id: int | None + """ + This is the stream ID of the latest to-device message (`device_federation_outbox`) to be + sent (or None if none sent). + + When the transaction completes, to-device messages up to this point will be deleted from + the outbox. + """ + + device_list_stream_id: int | None + """ + This is the stream ID of the latest device list to be sent (or None if none sent). + + When the transaction completes, we will mark device lists up to this point as having been + sent. + """ + + last_stream_ordering: int | None + """ + This is the stream ordering of the last PDU that was sent (or None if none sent). + + When the transaction completes, this should be stored as our position in the events stream. + """ + + pdu_count_from_main_queue: int + """ + The number of PDUs that were sent from the main queue. + """ + + sticky_events: _StickyEventsTransactionInfo | None + """ + Information useful for transactions sending backlogged sticky events. + """ + + class PerDestinationQueue: """ Manages the per-destination transmission queues. @@ -106,6 +176,16 @@ def __init__( 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( @@ -343,7 +423,7 @@ def attempt_new_transaction(self) -> None: ) async def _transaction_transmission_loop(self) -> None: - pending_pdus: list[EventBase] = [] + transaction: _PreparedTransaction | None = None try: self.transmission_loop_running = True # This will throw if we wouldn't retry. We do this here so we fail @@ -367,45 +447,63 @@ async def _transaction_transmission_loop(self) -> None: while self._transmission_loop_enabled: self._new_data_to_send = False - async with _TransactionQueueManager(self) as ( - pending_pdus, # noqa: F811 - pending_edus, + transaction = await self._prepare_transaction() + + if ( + transaction is not None + and not transaction.pdus + and not transaction.edus ): - if not pending_pdus and not pending_edus: - 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: - continue - else: - return - - if pending_pdus: - logger.debug( - "TX [%s] len(pending_pdus_by_dest[dest]) = %d", - self._destination, - len(pending_pdus), - ) - - await self._transaction_manager.send_new_transaction( - self._destination, pending_pdus, pending_edus + # There is nothing to send, but preparing the transaction has + # made progress that needs recording: the backlogged sticky + # events we selected must have all gotten filtered out. + await self._complete_transaction(transaction) + transaction = None + + 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. + # + # We also keep going whilst there are backlogged sticky events + # left to send, as returning here would leave the backlog + # waiting for unrelated traffic to start a new transmission loop. + if ( + self._new_data_to_send + or self._sticky_event_backlog_tracker.is_backlogged + ): + continue + else: + return + + if transaction.pdus: + logger.debug( + "TX [%s] len(pending_pdus_by_dest[dest]) = %d", + self._destination, + len(transaction.pdus), ) - sent_transactions_counter.labels( - **{SERVER_NAME_LABEL: self.server_name} + await self._transaction_manager.send_new_transaction( + self._destination, transaction.pdus, transaction.edus + ) + + sent_transactions_counter.labels( + **{SERVER_NAME_LABEL: self.server_name} + ).inc() + sent_edus_counter.labels(**{SERVER_NAME_LABEL: self.server_name}).inc( + len(transaction.edus) + ) + for edu in transaction.edus: + sent_edus_by_type.labels( + type=edu.edu_type, + **{SERVER_NAME_LABEL: self.server_name}, ).inc() - sent_edus_counter.labels( - **{SERVER_NAME_LABEL: self.server_name} - ).inc(len(pending_edus)) - for edu in pending_edus: - sent_edus_by_type.labels( - type=edu.edu_type, - **{SERVER_NAME_LABEL: self.server_name}, - ).inc() + + await self._complete_transaction(transaction) except NotRetryingDestination as e: logger.debug( @@ -455,16 +553,19 @@ async def _transaction_transmission_loop(self) -> None: "TX [%s] Failed to send transaction: %s", self._destination, e ) - for p in pending_pdus: - logger.info( - "Failed to send event %s to %s", p.event_id, self._destination - ) + if transaction is not None: + for p in transaction.pdus: + logger.info( + "Failed to send event %s to %s", p.event_id, self._destination + ) except Exception: logger.exception("TX [%s] Failed to send transaction", self._destination) - for p in pending_pdus: - logger.info( - "Failed to send event %s to %s", p.event_id, self._destination - ) + + if transaction is not None: + for p in transaction.pdus: + logger.info( + "Failed to send event %s to %s", p.event_id, self._destination + ) finally: # We want to be *very* sure we clear this after we stop processing self.active_transmission_loop = None @@ -491,6 +592,8 @@ async def _catch_up_transmission_loop(self) -> None: # needs catching up — so catching up is futile; let's stop. self._catching_up = False return + # (We just proved above that this is not None) + assert self._last_successful_stream_ordering is not None last_successful_stream_ordering: int = _tmp_last_successful_stream_ordering @@ -638,6 +741,20 @@ async def _catch_up_transmission_loop(self) -> None: # 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(pdu.internal_metadata.stream_ordering) + for 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 @@ -736,21 +853,73 @@ def _start_catching_up(self) -> None: self._catching_up = True self._pending_pdus = [] + async def _prepare_transaction(self) -> _PreparedTransaction | None: + """ + 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 -@attr.s(slots=True, auto_attribs=True) -class _TransactionQueueManager: - """A helper async context manager for pulling stuff off the queues and - tracking what was last successfully sent, etc. - """ + Side effects: + - Dequeues pending EDUs + - `_pending_presence` + - `_pending_receipt_edus` + - `_pending_edus` (currently unused in practice) + - `_pending_keyed_edus` + - Advances our devices stream position (but only if there is nothing to + send, so this is harmless) + + PDUs are not dequeued until acknowledged by `_complete_transaction`. - queue: PerDestinationQueue + Returns: + - the prepared transaction; or + - None if there is nothing to send and no progress to record + + Once the prepared transaction has been sent successfully, + `_complete_transaction` must be called with it. + """ - _device_stream_id: int | None = None - _device_list_id: int | None = None - _last_stream_ordering: int | None = None - _pdus: list[EventBase] = attr.Factory(list) + 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. + """ - async def __aenter__(self) -> tuple[list[EventBase], list[Edu]]: # 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, @@ -767,30 +936,30 @@ async def __aenter__(self) -> tuple[list[EventBase], list[Edu]]: pending_edus = [] # Add presence EDU. - if self.queue._pending_presence: + if self._pending_presence: # Only send max 50 presence entries in the EDU, to bound the amount # of data we're sending. presence_to_add: list[JsonDict] = [] while ( - self.queue._pending_presence + self._pending_presence and len(presence_to_add) < MAX_PRESENCE_STATES_PER_EDU ): - _, presence = self.queue._pending_presence.popitem(last=False) + _, presence = self._pending_presence.popitem(last=False) presence_to_add.append( - format_user_presence_state(presence, self.queue._clock.time_msec()) + format_user_presence_state(presence, self._clock.time_msec()) ) pending_edus.append( Edu( - origin=self.queue.server_name, - destination=self.queue._destination, + origin=self.server_name, + destination=self._destination, edu_type=EduTypes.PRESENCE, content={"push": presence_to_add}, ) ) # Add read receipt EDUs. - pending_edus.extend(self.queue._get_receipt_edus(limit=5)) + pending_edus.extend(self._get_receipt_edus(limit=5)) edu_limit = MAX_EDUS_PER_TRANSACTION - len(pending_edus) # Next, prioritize to-device messages so that existing encryption channels @@ -799,91 +968,240 @@ async def __aenter__(self) -> tuple[list[EventBase], list[Edu]]: ( to_device_edus, device_stream_id, - ) = await self.queue._get_to_device_message_edus( + ) = await self._get_to_device_message_edus( edu_limit - NUMBER_OF_RESERVED_EDUS_PER_TRANSACTION ) + device_stream_id_upon_completion: int | None = None if to_device_edus: - self._device_stream_id = device_stream_id + # We can advance our position in the device stream after the transaction completes. + device_stream_id_upon_completion = device_stream_id else: - self.queue._last_device_stream_id = device_stream_id + # We can advance our position in the device stream immediately, as there's nothing to send. + self._last_device_stream_id = device_stream_id pending_edus.extend(to_device_edus) edu_limit -= len(to_device_edus) # Add device list update EDUs. - device_update_edus, dev_list_id = await self.queue._get_device_update_edus( - edu_limit - ) + device_update_edus, dev_list_id = await self._get_device_update_edus(edu_limit) + device_list_id_upon_completion: int | None = None if device_update_edus: - self._device_list_id = dev_list_id + # We can advance our position in the device list stream after the transaction completes. + device_list_id_upon_completion = dev_list_id else: - self.queue._last_device_list_stream_id = dev_list_id + # We can advance our position in the device list stream immediately, as there's nothing to send. + self._last_device_list_stream_id = dev_list_id pending_edus.extend(device_update_edus) edu_limit -= len(device_update_edus) # Finally add any other types of EDUs if there is room. - other_edus = self.queue._pop_pending_edus(edu_limit) + other_edus = self._pop_pending_edus(edu_limit) pending_edus.extend(other_edus) edu_limit -= len(other_edus) - while edu_limit > 0 and self.queue._pending_edus_keyed: - _, val = self.queue._pending_edus_keyed.popitem() + while edu_limit > 0 and self._pending_edus_keyed: + _, val = self._pending_edus_keyed.popitem() pending_edus.append(val) edu_limit -= 1 # Now we look for any PDUs to send, by getting up to 50 PDUs from the # queue - self._pdus = self.queue._pending_pdus[:50] - - if not self._pdus and not pending_edus: - return [], [] - - if self._pdus: - self._last_stream_ordering = self._pdus[ - -1 - ].internal_metadata.stream_ordering - assert self._last_stream_ordering - - return self._pdus, pending_edus + pdus = self._pending_pdus[:50] + + if not pdus and not pending_edus: + # There is nothing to send. There's also nothing to record upon + # completion: the only progress we could have made without sending + # anything is advancing our positions in the device streams, and that + # has already been done above. + return None + + last_stream_ordering: int | None = None + if pdus: + last_stream_ordering = pdus[-1].internal_metadata.stream_ordering + assert last_stream_ordering + + return _PreparedTransaction( + pdus=pdus, + edus=pending_edus, + 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, + pdu_count_from_main_queue=len(pdus), + # This is not part of the sticky events backlog flow, + # so don't advance that + sticky_events=None, + ) - async def __aexit__( - self, - exc_type: type[BaseException] | None, - exc: BaseException | None, - tb: TracebackType | None, - ) -> None: - if exc_type is not None: - # Failed to send transaction, so we bail out. - return + async def _complete_transaction(self, transaction: _PreparedTransaction) -> None: + """ + Handle the fact that a transaction has been successfully completed. + Must not be called if sending the transaction failed, as it records how far + through the various streams we have now got. + """ # Successfully sent transactions, so we remove pending PDUs from the queue - if self._pdus: - self.queue._pending_pdus = self.queue._pending_pdus[len(self._pdus) :] + if transaction.pdu_count_from_main_queue: + self._pending_pdus = self._pending_pdus[ + transaction.pdu_count_from_main_queue : + ] # Succeeded to send the transaction so we record where we have sent up # to in the various streams - if self._device_stream_id: - await self.queue._store.delete_device_msgs_for_remote( - self.queue._destination, self._device_stream_id + if transaction.to_device_message_stream_id: + await self._store.delete_device_msgs_for_remote( + self._destination, transaction.to_device_message_stream_id ) - self.queue._last_device_stream_id = self._device_stream_id + self._last_device_stream_id = transaction.to_device_message_stream_id # also mark the device updates as sent - if self._device_list_id: + if transaction.device_list_stream_id: logger.info( - "Marking as sent %r %r", self.queue._destination, self._device_list_id + "Marking as sent %r %r", + self._destination, + transaction.device_list_stream_id, ) - await self.queue._store.mark_as_sent_devices_by_remote( - self.queue._destination, self._device_list_id + await self._store.mark_as_sent_devices_by_remote( + self._destination, transaction.device_list_stream_id ) - self.queue._last_device_list_stream_id = self._device_list_id + self._last_device_list_stream_id = transaction.device_list_stream_id - if self._last_stream_ordering: + if transaction.last_stream_ordering: # we sent some PDUs and it was successful, so update our # last_successful_stream_ordering in the destinations table. - await self.queue._store.set_destination_last_successful_stream_ordering( - self.queue._destination, self._last_stream_ordering + 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 + + @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 + backlog = await self._store.get_backlogged_sticky_events_for_destination( + self._destination + ) + + if backlog is None: + logger.info( + "Completed 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 %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, + self._destination, + self._own_server_name, + sticky_events, + # Omit filtered events + redact=False, + # Sticky events sent by erased users no longer need to be sent + # as part of catch-up + 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_id=None, + device_stream_id=None, + last_stream_ordering=None, + # These events are not from the main queue, so don't advance the main queue + pdu_count_from_main_queue=0, + # 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 f9c7821a396..774779caf44 100644 --- a/synapse/replication/tcp/streams/_base.py +++ b/synapse/replication/tcp/streams/_base.py @@ -28,6 +28,7 @@ Callable, Sequence, Union, + NewType, ) import attr @@ -866,6 +867,13 @@ async def _update_function( 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 023014276bf..e6f21364727 100644 --- a/synapse/storage/database.py +++ b/synapse/storage/database.py @@ -2748,3 +2748,16 @@ 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 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 9c5fd359064..d272efda1c8 100644 --- a/synapse/storage/databases/main/event_push_actions.py +++ b/synapse/storage/databases/main/event_push_actions.py @@ -100,6 +100,7 @@ LoggingDatabaseConnection, LoggingTransaction, PostgresEngine, + user_is_local_like_pattern, ) from synapse.storage.databases.main.receipts import ReceiptsWorkerStore from synapse.storage.databases.main.stream import StreamWorkerStore @@ -1442,7 +1443,7 @@ def _handle_new_receipts_for_notifs_txn(self, txn: LoggingTransaction) -> bool: # 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 b2b45612478..90b018073ed 100644 --- a/synapse/storage/databases/main/metrics.py +++ b/synapse/storage/databases/main/metrics.py @@ -30,6 +30,7 @@ DatabasePool, LoggingDatabaseConnection, LoggingTransaction, + user_is_local_like_pattern, ) from synapse.storage.databases.main.event_push_actions import ( EventPushActionsWorkerStore, @@ -133,9 +134,7 @@ def _count_messages(txn: LoggingTransaction) -> int: 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 @@ def _count_messages(txn: LoggingTransaction) -> int: 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 1accf207be0..a4fa318715a 100644 --- a/synapse/storage/databases/main/purge_events.py +++ b/synapse/storage/databases/main/purge_events.py @@ -23,7 +23,7 @@ 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 @@ -213,7 +213,10 @@ def _purge_history_txn( 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 eee6b924159..6e713d89afc 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 @@ -18,17 +19,22 @@ from twisted.internet.defer import Deferred from synapse.events import EventBase -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.state import StateGroupWorkerStore from synapse.storage.engines import PostgresEngine, Sqlite3Engine from synapse.storage.util.id_generators import MultiWriterIdGenerator +from synapse.types import RoomID from synapse.util.duration import Duration if TYPE_CHECKING: @@ -420,3 +426,340 @@ def _run_background_cleanup(self) -> Deferred: "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_room_sticky_events_backlog` table, if there are backlogged + sticky events to send to the given destination, returns up to 50 IDs of sticky + events from one room. + + 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) + """ + + 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_room_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 + -- filter to locally-originating sticky events + AND sender LIKE ? + ORDER BY stream_id ASC + LIMIT ? + """, + ( + room_id, + next_to_send_sticky_event_stream_position, + 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_room_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_room_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 backlog.sticky_events_stream_position <= se.stream_id + WHERE se.event_id IS NULL + AND backlog.destination = ? + ) + DELETE FROM destination_room_sticky_events_backlog + WHERE destination = ? AND room_id IN (SELECT room_id FROM to_clean_up) + """, + (user_is_local_like_pattern(self.hs), 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_room_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_room_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_room_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 < ? + + -- 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, + *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_room_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/schema/main/delta/93/05_destination_room_sticky_events.sql b/synapse/storage/schema/main/delta/93/05_destination_room_sticky_events.sql new file mode 100644 index 00000000000..3cf7d030960 --- /dev/null +++ b/synapse/storage/schema/main/delta/93/05_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_room_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_room_sticky_events_backlog_room_id ON destination_room_sticky_events_backlog (room_id); diff --git a/synapse/types/__init__.py b/synapse/types/__init__.py index 7516847303b..d92871a745c 100644 --- a/synapse/types/__init__.py +++ b/synapse/types/__init__.py @@ -1765,3 +1765,17 @@ def from_event(event: "EventBase") -> "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 fd1ef043bb8..c7350de851d 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,27 @@ PerDestinationQueue, TransactionManager, ) +from synapse.federation.sender.per_destination_queue import _PreparedTransaction from synapse.federation.units import Edu, Transaction +from synapse.replication.tcp.streams._base import StickyEventStreamPosition from synapse.rest import admin from synapse.rest.client import login, room from synapse.server import HomeServer -from synapse.types import JsonDict +from synapse.types import JsonDict, RoomID 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 +48,6 @@ def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer: ) 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 +106,55 @@ def get_destination_room(self, room: str, destination: str = "host2") -> dict: )[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 for test cases federation sending is disabled. This Test class 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 +310,6 @@ def test_catch_up_from_blank_state(self) -> None: 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 +604,359 @@ async def mock_await_full_state(event_ids: Collection[str]) -> None: 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 _backlog_rows(self) -> list[tuple[str, str, int]]: + """ + All rows of the `destination_room_sticky_events_backlog` table. + """ + rows = self.get_success( + self.hs.get_datastores().main.db_pool.simple_select_list( + table="destination_room_sticky_events_backlog", + keyvalues=None, + retcols=("destination", "room_id", "sticky_events_stream_position"), + ) + ) + return sorted(rows) + + def _assert_up_to_date_as_of(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) + 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 • • + # (message) sticky_id_5 + # room 2 • • + # event_id_2 event_id_4 + # room 3 • • + # sticky_id_3 event_id_6 + + self.helper.send_messages(room_1, 1, tok=u1_token) + (event_id_2,) = self.helper.send_messages(room_2, 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_up_to_date_as_of(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, 1, tok=u1_token) + # This one is itself a forward extremity, so it will get sent as 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, 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 + 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 was sent, oldest first. + 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.assertEqual( + per_dest_queue._last_successful_stream_ordering, + self._stream_ordering_for(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._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, 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_up_to_date_as_of(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, 1, tok=u1_token) + # Trigger federation sender + self.reactor.advance(0) + + # Do a 'regular' catch-up transaction. + # This is also what records the sticky event backlog + self.get_success(per_dest_queue._catch_up_transmission_loop()) + self.assertEqual( + self._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._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._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, 1, tok=u1_token) + self.reactor.advance(0) + + # Put host2 down + self.is_online = False + self._assert_up_to_date_as_of(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, 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 federation catch-up + 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._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, 1, tok=u1_token) + # Let the federation sender act on the events above. + self.reactor.advance(0) + self.is_online = False + self._assert_up_to_date_as_of(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, 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, 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._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) + ) + + # 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.append(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(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]) + + +class FederationStickyEventBacklogDisabledTestCase(_FederationCatchUpTestCaseBase): + def test_no_backlog_transaction_when_msc4354_disabled(self) -> None: + """ + Tests that when MSC4354 switched off, the sticky backlog is not serviced. + """ + per_dest_queue, _sent_pdus = self.make_fake_destination_queue() + + async def must_not_be_called( + destination: str, *, limit: int = 50 + ) -> tuple[RoomID, StickyEventStreamPosition, list[str]] | None: + raise AssertionError( + "Consulted the sticky event backlog despite MSC4354 being disabled" + ) + + with mock.patch.object( + self.hs.get_datastores().main, + "get_backlogged_sticky_events_for_destination", + must_not_be_called, + ): + # Check twice to make sure it's not down to round-robin + self.assertIsNone(self.run_transaction(per_dest_queue)) + self.assertIsNone(self.run_transaction(per_dest_queue)) diff --git a/tests/storage/test_sticky_events.py b/tests/storage/test_sticky_events.py index e77b362f528..0787336defe 100644 --- a/tests/storage/test_sticky_events.py +++ b/tests/storage/test_sticky_events.py @@ -22,15 +22,16 @@ 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 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 inject_event +from tests.test_utils.event_injection import inject_event, inject_member_event from tests.utils import USE_POSTGRES_FOR_TESTS @@ -428,3 +429,562 @@ def test_spam_checker_spammy_events_are_not_tracked(self) -> None: self.assertEqual(len(updates), 1) self.assertEqual(updates[0].event_id, valid_sticky_event.event_id) + + +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 _backlog_rows(self) -> list[tuple[str, str, int]]: + """ + All rows of the `destination_room_sticky_events_backlog` table. + """ + rows = self.get_success( + self.hs.get_datastores().main.db_pool.simple_select_list( + table="destination_room_sticky_events_backlog", + keyvalues=None, + retcols=("destination", "room_id", "sticky_events_stream_position"), + ) + ) + return sorted(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 = self.helper.send(room1, "before the gap", tok=self.token)[ + "event_id" + ] + + # Send 2 sticky events into room1 + room1_sticky1 = self._send_sticky(room1, "sticky 1") + _room1_sticky2 = self._send_sticky(room1, "sticky 2") + + # Send a sticky event into room2 + room2_sticky3 = self._send_sticky(room2, "sticky 3") + + # Send a couple of events for the the catch-up transaction to advance us to. + (room1_after_gap,) = self.helper.send_messages(room1, 1, tok=self.token) + (room2_after_gap,) = 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) + ) + ) + self.get_success( + self.store.store_destination_rooms_entries( + {"host2"}, room2, self._stream_ordering_for(room2_after_gap) + ) + ) + + # 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 + ), + new_last_successfully_sent_stream_ordering=self._stream_ordering_for( + room1_after_gap + ), + event_stream_orderings_sent_in_transaction={ + self._stream_ordering_for(room1_after_gap) + }, + ) + ) + + self.assertEqual( + self._backlog_rows(), + [ + # In room1: we need to catch up from the first sticky event + ("host2", room1, self._sticky_stream_id_for(room1_sticky1)), + # In room2: we need to catch up from the first sticky event in that room + ("host2", room2, self._sticky_stream_id_for(room2_sticky3)), + ], + ) + + 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_sticky = self._send_sticky(room_id, "already sent") + + # Send 2 regular events that we suppose we had _not_ delivered to the destination yet. + (_event2_nonsticky, _event3_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_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_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_sticky + ), + new_last_successfully_sent_stream_ordering=self._stream_ordering_for( + event4_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._backlog_rows(), []) + + 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 = 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 = self._send_sticky(room_id, "early sticky") + # Send a 'middle' event that we will send out in a catch-up transaction + event3_middle = 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) + ) + ) + + # 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 + ), + new_last_successfully_sent_stream_ordering=self._stream_ordering_for( + event3_middle + ), + event_stream_orderings_sent_in_transaction={ + self._stream_ordering_for(event3_middle) + }, + ) + ) + self.assertEqual( + self._backlog_rows(), + [("host2", room_id, self._sticky_stream_id_for(event2_sticky))], + ) + + # Now send another sticky event and 'lose' it in a gap again. + event4_sticky = self._send_sticky(room_id, "late sticky") + # Send a final event that we will send out in a catch-up transaction + # (in order to create a gap for `event4_sticky` to sit in) + event5_end = 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) + ) + ) + + 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 + ), + new_last_successfully_sent_stream_ordering=self._stream_ordering_for( + event5_end + ), + event_stream_orderings_sent_in_transaction={ + self._stream_ordering_for(event5_end) + }, + ) + ) + + # 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._backlog_rows(), + [("host2", room_id, self._sticky_stream_id_for(event2_sticky))], + ) + + 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 = self._send_sticky(room_id, "sticky 1") + sticky_2 = self._send_sticky(room_id, "sticky 2") + sticky_3 = self._send_sticky(room_id, "sticky 3") + sticky_4 = 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_room_sticky_events_backlog", + values={ + "destination": "host2", + "room_id": room_id, + "sticky_events_stream_position": self._sticky_stream_id_for( + sticky_2 + ), + }, + 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), + # We see sticky 2 events up to and including 4, in that order + [sticky_2, sticky_3, sticky_4], + ), + ) + + 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 = self._send_sticky(room_id, "sticky 1") + sticky_2 = self._send_sticky(room_id, "sticky 2") + _sticky_3 = 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_room_sticky_events_backlog", + values={ + "destination": "host2", + "room_id": room_id, + "sticky_events_stream_position": self._sticky_stream_id_for( + sticky_1 + ), + }, + 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), + # We limit to 2 so we don't see sticky_3 here + [sticky_1, sticky_2], + ), + ) + + 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 = 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 = 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, local_sticky}, + ) + + assert self._sticky_stream_id_for(remote_sticky) < self._sticky_stream_id_for( + local_sticky + ) + + self.get_success( + self.store.db_pool.simple_insert( + table="destination_room_sticky_events_backlog", + values={ + "destination": "host2", + "room_id": room_id, + "sticky_events_stream_position": self._sticky_stream_id_for( + remote_sticky + ), + }, + 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), + [local_sticky], + ), + ) + + 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 = 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_room_sticky_events_backlog", + values={ + "destination": "host2", + "room_id": room_id, + "sticky_events_stream_position": self._sticky_stream_id_for( + short_lived + ), + }, + 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_room_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._backlog_rows(), []) + + 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 = self._send_sticky(room_id, "sticky 1") + sticky_2 = self._send_sticky(room_id, "sticky 2") + sticky_3 = self._send_sticky(room_id, "sticky 3") + + self.get_success( + self.store.db_pool.simple_insert( + table="destination_room_sticky_events_backlog", + values={ + "destination": "host2", + "room_id": room_id, + "sticky_events_stream_position": self._sticky_stream_id_for( + sticky_1 + ), + }, + 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)), + ) + ) + + # The stored position is an *inclusive lower bound on what is left*, hence + # exactly one past the highest event we sent. + self.assertEqual( + self._backlog_rows(), + [("host2", room_id, self._sticky_stream_id_for(sticky_2) + 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), + [sticky_3], + ), + )