Update wait_for_stream_token(...) patterns and fix sync fetching with unbounded token

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.

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.

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.

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
 

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,

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)(

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()],

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:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+
+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)