diff --git a/changelog.d/19644.bugfix b/changelog.d/19644.bugfix new file mode 100644 index 00000000000..73ab4bc63e9 --- /dev/null +++ b/changelog.d/19644.bugfix @@ -0,0 +1 @@ +Fix long-standing but niche bug with sync where it could attempt to fetch data with flawed invalid future tokens. diff --git a/synapse/handlers/sliding_sync/__init__.py b/synapse/handlers/sliding_sync/__init__.py index a3443b300cc..2e26bfc43a2 100644 --- a/synapse/handlers/sliding_sync/__init__.py +++ b/synapse/handlers/sliding_sync/__init__.py @@ -150,6 +150,23 @@ async def wait_for_sync_for_user( # events or future events if the user is nefariously, manually modifying the # token. if from_token is not None: + # Work around a bug where older Synapse versions gave out tokens "from the + # future", i.e. that are ahead of the tokens persisted in the DB. This could + # also happen if a user is intentionally messing with the token so this also + # acts as sanitization/validation. + # + # If the token has positions ahead of our persisted positions in the + # database (invalid), then we simply use our max persisted position (recover + # gracefully); instead of waiting for a position that may never come around. + # + # FIXME: For Sliding Sync, instead of bounding the token, we should detect + # the invalid future position and raise a `M_UNKNOWN_POS` error. + from_token = SlidingSyncStreamToken( + stream_token=await self.event_sources.bound_future_token( + from_token.stream_token + ), + connection_position=from_token.connection_position, + ) # We need to make sure this worker has caught up with the token. If # this returns false, it means we timed out waiting, and we should # just return an empty response. diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index c88f703ae98..bca51b64b71 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -414,6 +414,15 @@ async def _wait_for_sync_for_user( context.tag = sync_label if since_token is not None: + # Work around a bug where older Synapse versions gave out tokens "from the + # future", i.e. that are ahead of the tokens persisted in the DB. This could + # also happen if a user is intentionally messing with the token so this also + # acts as sanitization/validation. + # + # If the token has positions ahead of our persisted positions in the + # database (invalid), then we simply use our max persisted position (recover + # gracefully); instead of waiting for a position that may never come around. + since_token = await self.event_sources.bound_future_token(since_token) # We need to make sure this worker has caught up with the token. If # this returns false it means we timed out waiting, and we should # just return an empty response. diff --git a/synapse/notifier.py b/synapse/notifier.py index f1cec74462f..6a057ac09fa 100644 --- a/synapse/notifier.py +++ b/synapse/notifier.py @@ -832,15 +832,49 @@ async def check_for_updates( return result async def wait_for_stream_token(self, stream_token: StreamToken) -> bool: - """Wait for this worker to catch up with the given stream token.""" + """ + Wait for this worker to catch up with the given stream token. + + This is important to ensure that the worker has a proper view of the world + before trying to serve a request. For example, one worker can return a response + with some `next_batch` token, but then the next request goes to another worker + which is behind; if the worker assembles a response up to the token, it could be + missing data in the gap between where it's behind and the requested token. + + ### Inavlid future tokens + + We assume the token has already been validated/sanitized before being passed to + this function to ensure it's not some invalid future token. We consider a token + invalid, if the token has positions ahead of our persisted positions in the + database. This is important as we we don't want to wait for the stream to + advance in those cases (as it may never do so) (it's a waste of time for the + user and server). + + Previously, we would sanitize and `bound_future_token(...)` within this function + but that leads to bad patterns upstream where people can continue to use the + unbounded token. + + While it was possible for older Synapse versions to erroneously give out invalid + future tokens, this is no longer the case and its considered a Synapse + programming error if this ever happens. Validation/sanitization is still + necessary as a user can intentionally mess with numbers in the tokens being + provided. + + Args: + stream_token: The token to wait for. We assume the token has already been + validated/sanitized to ensure it's not some invalid future token (has a + stream position ahead of what is in the DB). (see details above) + + Returns: + True when this worker has caught up + False when we timed out waiting + """ current_token = self.event_sources.get_current_token() + # Return early if we are already caught up if stream_token.is_before_or_eq(current_token): return True - # Work around a bug where older Synapse versions gave out tokens "from - # the future", i.e. that are ahead of the tokens persisted in the DB. - stream_token = await self.event_sources.bound_future_token(stream_token) - + # Start waiting until we've caught up to the `stream_token` start = self.clock.time_msec() logged = False while True: @@ -850,6 +884,7 @@ async def wait_for_stream_token(self, stream_token: StreamToken) -> bool: now = self.clock.time_msec() + # Timed out if now - start > 10_000: return False diff --git a/synapse/streams/events.py b/synapse/streams/events.py index d2720fb9592..f5677a20828 100644 --- a/synapse/streams/events.py +++ b/synapse/streams/events.py @@ -149,6 +149,16 @@ async def bound_future_token(self, token: StreamToken) -> StreamToken: ].get_max_allocated_token() if max_token < token_value.get_max_stream_pos(): + # Log *something* as we consider this as a Synapse programming error + # (assuming no malicious user manipulation of the token) (we shouldn't be + # handing out future tokens). + # + # We don't assert as the whole point of bounding is so that we can recover + # gracefully. + # + # Old versions of Synapse could advance streams without persisting anything in + # the DB (fixed in https://github.com/element-hq/synapse/pull/17229) and on + # restart, those updates would be lost. logger.error( "Bounding token from the future '%s': token: %s, bound: %s", key, diff --git a/synapse/types/__init__.py b/synapse/types/__init__.py index 02889795bbd..8b005ef84d5 100644 --- a/synapse/types/__init__.py +++ b/synapse/types/__init__.py @@ -685,6 +685,9 @@ def is_before_or_eq(self, other_token: Self) -> bool: def bound_stream_token(self, max_stream: int) -> "Self": """Bound the stream positions to a maximum value""" + # Shortcut if we're already under the bound + if self.get_max_stream_pos() <= max_stream: + return self min_pos = min(self.stream, max_stream) return type(self)( diff --git a/tests/handlers/test_sync.py b/tests/handlers/test_sync.py index b9dee1c9547..e4a4d59aeb3 100644 --- a/tests/handlers/test_sync.py +++ b/tests/handlers/test_sync.py @@ -51,6 +51,7 @@ create_requester, ) from synapse.util.clock import Clock +from synapse.util.duration import Duration import tests.unittest import tests.utils @@ -1058,13 +1059,22 @@ def test_wait_for_future_sync_token(self) -> None: ) # This should block waiting for the presence stream to update - self.pump() + # + # Advance time a little bit to make the + # `wait_for_stream_token(...)` sleep loop iterate. + self.reactor.advance(Duration(seconds=2).as_secs()) + # It should still not be done yet self.assertFalse(sync_d.called) # Marking the stream ID as persisted should unblock the request. self.get_success(ctx_mgr.__aexit__(None, None, None)) - self.get_success(sync_d, by=1.0) + # Advance time to make another iteration of `wait_for_stream_token(...)` sleep + # loop so it sees that we're finally caught up now. + self.reactor.advance(Duration(seconds=1).as_secs()) + + # Done waiting + self.get_success(sync_d) @parameterized.expand( [(key,) for key in StreamKeyType.__members__.values()], diff --git a/tests/test_notifier.py b/tests/test_notifier.py new file mode 100644 index 00000000000..c65134e8325 --- /dev/null +++ b/tests/test_notifier.py @@ -0,0 +1,136 @@ +# 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: +# . + +import logging + +from twisted.internet import defer +from twisted.internet.testing import MemoryReactor + +from synapse.server import HomeServer +from synapse.types import MultiWriterStreamToken, StreamKeyType, StreamToken +from synapse.util.clock import Clock +from synapse.util.duration import Duration + +import tests.unittest + +logger = logging.getLogger(__name__) + + +class NotifierTestCase(tests.unittest.HomeserverTestCase): + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + self.store = self.hs.get_datastores().main + self.notifier = self.hs.get_notifier() + + def test_wait_for_stream_token_with_caught_up_token(self) -> None: + """ + Test `wait_for_stream_token` when we receive a token that we are caught up to. + """ + # Create a token + receipt_id_gen = self.store.get_receipts_stream_id_gen() + receipt_token = MultiWriterStreamToken.from_generator(receipt_id_gen) + token = StreamToken.START.copy_and_replace(StreamKeyType.RECEIPT, receipt_token) + + # Function under test + wait_d = defer.ensureDeferred(self.notifier.wait_for_stream_token(token)) + + # Done waiting and caught-up (True) + wait_result = self.get_success(wait_d) + self.assertEqual(wait_result, True) + + def test_wait_for_stream_token_with_future_sync_token(self) -> None: + """ + Test `wait_for_stream_token` when we receive a token that is ahead of our + current token, we'll wait until the stream position advances. + + This can happen if replication streams start lagging, and the client's + previous sync request was serviced by a worker ahead of ours. + """ + # We simulate a lagging stream by getting a stream ID from the ID gen + # and then waiting to mark it as "persisted". + receipt_id_gen = self.store.get_receipts_stream_id_gen() + ctx_mgr = receipt_id_gen.get_next() + receipt_stream_id = self.get_success(ctx_mgr.__aenter__()) + + # Create the new token based on the stream ID above. + current_receipt_token = MultiWriterStreamToken.from_generator(receipt_id_gen) + receipt_token = current_receipt_token.copy_and_advance( + MultiWriterStreamToken(stream=receipt_stream_id) + ) + token = StreamToken.START.copy_and_advance(StreamKeyType.RECEIPT, receipt_token) + + # Function under test + wait_d = defer.ensureDeferred(self.notifier.wait_for_stream_token(token)) + + # This should block waiting for the stream to update + # + # Advance time a little bit to make the + # `wait_for_stream_token(...)` sleep loop iterate. + self.reactor.advance(Duration(seconds=2).as_secs()) + # It should still not be done yet + self.assertFalse(wait_d.called) + + # Marking the stream ID as persisted should unblock the request. + self.get_success(ctx_mgr.__aexit__(None, None, None)) + + # Advance time to make another iteration of + # `wait_for_stream_token(...)` sleep loop so it sees that we're + # finally caught up now. + self.reactor.advance(Duration(seconds=1).as_secs()) + + # Done waiting and caught-up (True) + wait_result = self.get_success(wait_d) + self.assertEqual(wait_result, True) + + def test_wait_for_stream_token_with_future_sync_token_timeout( + self, + ) -> None: + """ + Test `wait_for_stream_token` when we receive a token that is ahead of our + current token, we'll wait until the stream position advances *until* we hit the + timeout. + + This can happen if replication streams start lagging, and the client's + previous sync request was serviced by a worker ahead of ours. + """ + # We simulate a lagging stream by getting a stream ID from the ID gen + # and then waiting to mark it as "persisted". + receipt_id_gen = self.store.get_receipts_stream_id_gen() + ctx_mgr = receipt_id_gen.get_next() + receipt_stream_id = self.get_success(ctx_mgr.__aenter__()) + + # Create the new token based on the stream ID above. + current_receipt_token = MultiWriterStreamToken.from_generator(receipt_id_gen) + receipt_token = current_receipt_token.copy_and_advance( + MultiWriterStreamToken(stream=receipt_stream_id) + ) + token = StreamToken.START.copy_and_advance(StreamKeyType.RECEIPT, receipt_token) + + # Function under test + wait_d = defer.ensureDeferred(self.notifier.wait_for_stream_token(token)) + # Advance time a little bit to make the + # `wait_for_stream_token(...)` sleep loop record 0 as the `start` time. + self.reactor.advance(Duration(seconds=0).as_secs()) + + # This should block waiting for the stream to update + # + # Advance time a little bit to make the + # `wait_for_stream_token(...)` sleep loop iterate. + self.reactor.advance(Duration(seconds=5).as_secs()) + # It should still not be done yet (not enough time to hit the timeout) + self.assertFalse(wait_d.called) + # Advance time past the 10 second timeout (5 + 6 = 11 seconds) to make the + # `wait_for_stream_token(...)` sleep loop give up. + self.reactor.advance(Duration(seconds=6).as_secs()) + + # Make sure we gave up waiting and not caught-up (False) + wait_result = self.get_success(wait_d) + self.assertEqual(wait_result, False)