From 2bdafecfeae23cf1aeddd7cfcd9d519739f59584 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 2 Apr 2026 12:34:59 -0500 Subject: [PATCH 01/19] Start of `wait_for_multi_writer_stream_token` --- synapse/notifier.py | 52 +++++++++++++++++++++++++- synapse/storage/databases/main/room.py | 7 +++- synapse/types/__init__.py | 17 +++++++++ 3 files changed, 74 insertions(+), 2 deletions(-) diff --git a/synapse/notifier.py b/synapse/notifier.py index 93d438def71..020b3416893 100644 --- a/synapse/notifier.py +++ b/synapse/notifier.py @@ -46,6 +46,9 @@ from synapse.logging.context import PreserveLoggingContext from synapse.logging.opentracing import log_kv, start_active_span from synapse.metrics import SERVER_NAME_LABEL, LaterGauge +from synapse.storage.util.id_generators import ( + MultiWriterIdGenerator, +) from synapse.streams.config import PaginationConfig from synapse.types import ( ISynapseReactor, @@ -831,8 +834,15 @@ 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. + + 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 @@ -840,6 +850,7 @@ async def wait_for_stream_token(self, stream_token: StreamToken) -> bool: # 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: @@ -849,6 +860,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 @@ -863,6 +875,44 @@ async def wait_for_stream_token(self, stream_token: StreamToken) -> bool: # TODO: be better await self.clock.sleep(Duration(milliseconds=500)) + async def wait_for_multi_writer_stream_token( + self, token: MultiWriterStreamToken, id_gen: MultiWriterIdGenerator + ) -> bool: + """Wait for this worker to catch up with the given stream token.""" + current_token = id_gen.get_current_token() + # Return early if we are already caught up + if 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. + token = token.bound_future_token(TODO) + + # Start waiting until we've caught up to the `stream_token` + start = self.clock.time_msec() + logged = False + while True: + current_token = self.event_sources.get_current_token() + if token.is_before_or_eq(current_token): + return True + + now = self.clock.time_msec() + + # Timed out + if now - start > 10_000: + return False + + if not logged: + logger.info( + "Waiting for current token to reach %s; currently at %s", + token, + current_token, + ) + logged = True + + # TODO: be better + await self.clock.sleep(Duration(milliseconds=500)) + async def _get_room_ids( self, user: UserID, explicit_room_id: str | None ) -> tuple[StrCollection, bool]: diff --git a/synapse/storage/databases/main/room.py b/synapse/storage/databases/main/room.py index 7ac88e4c2aa..5e6e971f9fb 100644 --- a/synapse/storage/databases/main/room.py +++ b/synapse/storage/databases/main/room.py @@ -58,7 +58,12 @@ from synapse.storage.databases.main.cache import CacheInvalidationWorkerStore from synapse.storage.types import Cursor from synapse.storage.util.id_generators import IdGenerator, MultiWriterIdGenerator -from synapse.types import JsonDict, RetentionPolicy, StrCollection, ThirdPartyInstanceID +from synapse.types import ( + JsonDict, + RetentionPolicy, + StrCollection, + ThirdPartyInstanceID, +) from synapse.util.caches.descriptors import cached, cachedList from synapse.util.json import json_encoder from synapse.util.stringutils import MXC_REGEX diff --git a/synapse/types/__init__.py b/synapse/types/__init__.py index 02889795bbd..6e68123e629 100644 --- a/synapse/types/__init__.py +++ b/synapse/types/__init__.py @@ -685,6 +685,23 @@ 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 max_stream <= self.get_max_stream_pos(): + return self + + # Log *something* as we consider this a programming error + # + # 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: token: %s, bound: %s", + self, + max_stream, + ) min_pos = min(self.stream, max_stream) return type(self)( From dc3d205ffc55b3703871395067e3a2df148cfe38 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 2 Apr 2026 12:52:15 -0500 Subject: [PATCH 02/19] Better `wait_for_multi_writer_stream_token` --- synapse/notifier.py | 20 +++++++++++++++----- 1 file changed, 15 insertions(+), 5 deletions(-) diff --git a/synapse/notifier.py b/synapse/notifier.py index 020b3416893..df1b87c945d 100644 --- a/synapse/notifier.py +++ b/synapse/notifier.py @@ -51,6 +51,7 @@ ) from synapse.streams.config import PaginationConfig from synapse.types import ( + AbstractMultiWriterStreamToken, ISynapseReactor, JsonDict, MultiWriterStreamToken, @@ -876,23 +877,32 @@ async def wait_for_stream_token(self, stream_token: StreamToken) -> bool: await self.clock.sleep(Duration(milliseconds=500)) async def wait_for_multi_writer_stream_token( - self, token: MultiWriterStreamToken, id_gen: MultiWriterIdGenerator + self, + token: AbstractMultiWriterStreamToken, + id_gen: MultiWriterIdGenerator, ) -> bool: - """Wait for this worker to catch up with the given stream token.""" - current_token = id_gen.get_current_token() + """ + Wait for this worker to catch up with the given stream token. + + Returns: + True when this worker has caught up + False when we timed out waiting + """ + current_token = AbstractMultiWriterStreamToken.from_generator(id_gen) # Return early if we are already caught up if 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. - token = token.bound_future_token(TODO) + max_token = await id_gen.get_max_allocated_token() + token = token.bound_stream_token(max_token) # Start waiting until we've caught up to the `stream_token` start = self.clock.time_msec() logged = False while True: - current_token = self.event_sources.get_current_token() + current_token = AbstractMultiWriterStreamToken.from_generator(id_gen) if token.is_before_or_eq(current_token): return True From 6eda7661cd26217c70ecd8e59028af0959ca3640 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 2 Apr 2026 13:12:58 -0500 Subject: [PATCH 03/19] More refinement --- synapse/notifier.py | 25 ++++++++++++++++++++++++- synapse/streams/events.py | 8 ++++++++ synapse/types/__init__.py | 14 -------------- 3 files changed, 32 insertions(+), 15 deletions(-) diff --git a/synapse/notifier.py b/synapse/notifier.py index df1b87c945d..961ea92e0d6 100644 --- a/synapse/notifier.py +++ b/synapse/notifier.py @@ -884,6 +884,17 @@ async def wait_for_multi_writer_stream_token( """ Wait for this worker to catch up with the given stream token. + If the token has positions ahead of our persisted positions in the database + (invalid), then we simply wait until we catch-up to our max persisted position + (recover gracefully); instead of waiting for a position that may never come + around. We have a 10 second timeout here but upstream usage will just return the + same token as the `next_batch` and the user will try again which could cause no + progress to be made. + + This can happen due to older versions of Synapse giving out stream positions + without persisting them in the DB, and so on restart the stream would get reset + back to an older position. + Returns: True when this worker has caught up False when we timed out waiting @@ -896,7 +907,19 @@ async def wait_for_multi_writer_stream_token( # 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. max_token = await id_gen.get_max_allocated_token() - token = token.bound_stream_token(max_token) + if max_token < token.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. + logger.error( + "Bounding token from the future: token: %s, bound: %s", + token, + max_token, + ) + token = token.bound_stream_token(max_token) # Start waiting until we've caught up to the `stream_token` start = self.clock.time_msec() diff --git a/synapse/streams/events.py b/synapse/streams/events.py index d2720fb9592..a5b7ded74fc 100644 --- a/synapse/streams/events.py +++ b/synapse/streams/events.py @@ -149,6 +149,14 @@ 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 a Synapse programming error + # + # 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 6e68123e629..d127be98eda 100644 --- a/synapse/types/__init__.py +++ b/synapse/types/__init__.py @@ -689,20 +689,6 @@ def bound_stream_token(self, max_stream: int) -> "Self": if max_stream <= self.get_max_stream_pos(): return self - # Log *something* as we consider this a programming error - # - # 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: token: %s, bound: %s", - self, - max_stream, - ) - min_pos = min(self.stream, max_stream) return type(self)( stream=min_pos, From f61d795c3dc369f08660e41be6664714458f2ff6 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 2 Apr 2026 14:02:15 -0500 Subject: [PATCH 04/19] Separate out bounding Previously, we would wait for the bounded token and then still use the unbounded `since_token` for all of the queries (flawed). --- synapse/handlers/sliding_sync/__init__.py | 12 ++++ synapse/handlers/sync.py | 7 +++ synapse/notifier.py | 69 +++++++++++++---------- 3 files changed, 59 insertions(+), 29 deletions(-) diff --git a/synapse/handlers/sliding_sync/__init__.py b/synapse/handlers/sliding_sync/__init__.py index 6feb6c292e9..142e7dbd35d 100644 --- a/synapse/handlers/sliding_sync/__init__.py +++ b/synapse/handlers/sliding_sync/__init__.py @@ -150,6 +150,18 @@ 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. + # + # 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. + 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 c8ef5e2aa6c..90b5ebf8ac0 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -413,6 +413,13 @@ 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. + # + # 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 961ea92e0d6..d266684a742 100644 --- a/synapse/notifier.py +++ b/synapse/notifier.py @@ -838,6 +838,12 @@ async def wait_for_stream_token(self, stream_token: StreamToken) -> bool: """ 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. + Returns: True when this worker has caught up False when we timed out waiting @@ -847,9 +853,21 @@ async def wait_for_stream_token(self, stream_token: StreamToken) -> bool: 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) + # Assert as we consider this a Synapse programming error. We shouldn't be + # handing out invalid future tokens and tokens should be validated before it + # reaches this point. + # + # We consider a token invalid, if the token has positions ahead of our persisted + # positions in the database + # + # Previously, we would bound the tokens within this function but that leads to + # bad patterns upstream where people can continue to use the unbounded token. + original_stream_token = stream_token + max_token = await self.event_sources.bound_future_token(stream_token) + assert stream_token.is_before_or_eq(max_token), ( + f"Unable to wait for invalid future stream token (token={original_stream_token} has positions " + "ahead of our max persisted position {max_token})" + ) # Start waiting until we've caught up to the `stream_token` start = self.clock.time_msec() @@ -884,16 +902,11 @@ async def wait_for_multi_writer_stream_token( """ Wait for this worker to catch up with the given stream token. - If the token has positions ahead of our persisted positions in the database - (invalid), then we simply wait until we catch-up to our max persisted position - (recover gracefully); instead of waiting for a position that may never come - around. We have a 10 second timeout here but upstream usage will just return the - same token as the `next_batch` and the user will try again which could cause no - progress to be made. - - This can happen due to older versions of Synapse giving out stream positions - without persisting them in the DB, and so on restart the stream would get reset - back to an older position. + 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. Returns: True when this worker has caught up @@ -904,22 +917,20 @@ async def wait_for_multi_writer_stream_token( if 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. - max_token = await id_gen.get_max_allocated_token() - if max_token < token.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. - logger.error( - "Bounding token from the future: token: %s, bound: %s", - token, - max_token, - ) - token = token.bound_stream_token(max_token) + # Assert as we consider this a Synapse programming error. We shouldn't be + # handing out invalid future tokens and tokens should be validated before it + # reaches this point. + # + # We consider a token invalid, if the token has positions ahead of our persisted + # positions in the database + # + # Previously, we would bound the tokens within this function but that leads to + # bad patterns upstream where people can continue to use the unbounded token. + max_persisted_position = await id_gen.get_max_allocated_token() + assert max_persisted_position >= token.get_max_stream_pos(), ( + f"Unable to wait for invalid future token (token={token} has positions " + "ahead of our max persisted position {max_persisted_position})" + ) # Start waiting until we've caught up to the `stream_token` start = self.clock.time_msec() From 03e5f47bf4eac7177d4bf13376835003fac1689d Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 2 Apr 2026 14:11:07 -0500 Subject: [PATCH 05/19] Revert stray changes --- synapse/storage/databases/main/room.py | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/synapse/storage/databases/main/room.py b/synapse/storage/databases/main/room.py index 5e6e971f9fb..7ac88e4c2aa 100644 --- a/synapse/storage/databases/main/room.py +++ b/synapse/storage/databases/main/room.py @@ -58,12 +58,7 @@ from synapse.storage.databases.main.cache import CacheInvalidationWorkerStore from synapse.storage.types import Cursor from synapse.storage.util.id_generators import IdGenerator, MultiWriterIdGenerator -from synapse.types import ( - JsonDict, - RetentionPolicy, - StrCollection, - ThirdPartyInstanceID, -) +from synapse.types import JsonDict, RetentionPolicy, StrCollection, ThirdPartyInstanceID from synapse.util.caches.descriptors import cached, cachedList from synapse.util.json import json_encoder from synapse.util.stringutils import MXC_REGEX From 2945a020061c08071664fe8b79c880846a29029b Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 2 Apr 2026 14:17:48 -0500 Subject: [PATCH 06/19] Better explain --- synapse/streams/events.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/synapse/streams/events.py b/synapse/streams/events.py index a5b7ded74fc..f5677a20828 100644 --- a/synapse/streams/events.py +++ b/synapse/streams/events.py @@ -149,7 +149,9 @@ 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 a Synapse programming error + # 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. From 22d879c404a7969398299e1a13741cc7e262cdca Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 2 Apr 2026 14:26:07 -0500 Subject: [PATCH 07/19] Add changelog --- changelog.d/19644.bugfix | 1 + changelog.d/19644.misc | 1 + 2 files changed, 2 insertions(+) create mode 100644 changelog.d/19644.bugfix create mode 100644 changelog.d/19644.misc 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/changelog.d/19644.misc b/changelog.d/19644.misc new file mode 100644 index 00000000000..d3a7a12fc35 --- /dev/null +++ b/changelog.d/19644.misc @@ -0,0 +1 @@ +Add new `Notifier.wait_for_multi_writer_stream_token(...)` utility to make it easy wait for the worker to catch up to any kind of token. From 4ba64fc35ad0252b462de48960a61b961e3067a3 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 2 Apr 2026 14:30:47 -0500 Subject: [PATCH 08/19] Explain intentional scenario --- synapse/handlers/sliding_sync/__init__.py | 6 ++++-- synapse/handlers/sync.py | 6 ++++-- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/synapse/handlers/sliding_sync/__init__.py b/synapse/handlers/sliding_sync/__init__.py index 142e7dbd35d..e23a0ee1ff2 100644 --- a/synapse/handlers/sliding_sync/__init__.py +++ b/synapse/handlers/sliding_sync/__init__.py @@ -150,8 +150,10 @@ 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. + # 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 diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index 90b5ebf8ac0..e33d6f82925 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -413,8 +413,10 @@ 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. + # 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 From ff5402f414d8a4d868af92cd1556dc741346e1e1 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 2 Apr 2026 15:53:51 -0500 Subject: [PATCH 09/19] Fix condition See https://github.com/element-hq/synapse/pull/19644#discussion_r3030326143 Example test failure: ```shell $ SYNAPSE_TEST_LOG_LEVEL=INFO poetry run trial tests.handlers.test_sync.SyncTestCase_state.test_wait_for_invalid_future_sync_token_ROOM tests.handlers.test_sync SyncTestCase_state test_wait_for_invalid_future_sync_token_ROOM ... [FAIL] =============================================================================== [FAIL] Traceback (most recent call last): File "/home/eric/.cache/pypoetry/virtualenvs/matrix-synapse-xCtC9ulO-py3.14/lib/python3.14/site-packages/parameterized/parameterized.py", line 620, in standalone_func return func(*(a + p.args), **p.kwargs, **kw) File "/home/eric/Documents/github/element/synapse/tests/handlers/test_sync.py", line 1115, in test_wait_for_invalid_future_sync_token self.get_success(sync_d) File "/home/eric/Documents/github/element/synapse/tests/unittest.py", line 742, in get_success return self.successResultOf(deferred) File "/home/eric/.cache/pypoetry/virtualenvs/matrix-synapse-xCtC9ulO-py3.14/lib/python3.14/site-packages/twisted/trial/_synctest.py", line 723, in successResultOf self.fail( twisted.trial.unittest.FailTest: Success result expected on , found no result instead tests.handlers.test_sync.SyncTestCase_state.test_wait_for_invalid_future_sync_token_ROOM ------------------------------------------------------------------------------- Ran 1 tests in 0.163s FAILED (failures=1) ``` --- synapse/types/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/synapse/types/__init__.py b/synapse/types/__init__.py index d127be98eda..8b005ef84d5 100644 --- a/synapse/types/__init__.py +++ b/synapse/types/__init__.py @@ -686,7 +686,7 @@ 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 max_stream <= self.get_max_stream_pos(): + if self.get_max_stream_pos() <= max_stream: return self min_pos = min(self.stream, max_stream) From 154b41d0fedc5cdab8d032de8064d87a26b26b7a Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 2 Apr 2026 22:22:42 -0500 Subject: [PATCH 10/19] Add tests --- synapse/notifier.py | 7 +- tests/test_notifier.py | 163 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 166 insertions(+), 4 deletions(-) create mode 100644 tests/test_notifier.py diff --git a/synapse/notifier.py b/synapse/notifier.py index d266684a742..6fe66797beb 100644 --- a/synapse/notifier.py +++ b/synapse/notifier.py @@ -51,7 +51,6 @@ ) from synapse.streams.config import PaginationConfig from synapse.types import ( - AbstractMultiWriterStreamToken, ISynapseReactor, JsonDict, MultiWriterStreamToken, @@ -896,7 +895,7 @@ async def wait_for_stream_token(self, stream_token: StreamToken) -> bool: async def wait_for_multi_writer_stream_token( self, - token: AbstractMultiWriterStreamToken, + token: MultiWriterStreamToken, id_gen: MultiWriterIdGenerator, ) -> bool: """ @@ -912,7 +911,7 @@ async def wait_for_multi_writer_stream_token( True when this worker has caught up False when we timed out waiting """ - current_token = AbstractMultiWriterStreamToken.from_generator(id_gen) + current_token = MultiWriterStreamToken.from_generator(id_gen) # Return early if we are already caught up if token.is_before_or_eq(current_token): return True @@ -936,7 +935,7 @@ async def wait_for_multi_writer_stream_token( start = self.clock.time_msec() logged = False while True: - current_token = AbstractMultiWriterStreamToken.from_generator(id_gen) + current_token = MultiWriterStreamToken.from_generator(id_gen) if token.is_before_or_eq(current_token): return True diff --git a/tests/test_notifier.py b/tests/test_notifier.py new file mode 100644 index 00000000000..42ff76eab95 --- /dev/null +++ b/tests/test_notifier.py @@ -0,0 +1,163 @@ +# 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 +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_multi_writer_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 + presence_id_gen = self.store.get_presence_stream_id_gen() + token = MultiWriterStreamToken.from_generator(presence_id_gen) + + # Function under test + wait_d = defer.ensureDeferred( + self.notifier.wait_for_multi_writer_stream_token(token, presence_id_gen) + ) + + # Done waiting and caught-up (True) + wait_result = self.get_success(wait_d) + self.assertEqual(wait_result, True) + + def test_wait_for_multi_writer_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". + presence_id_gen = self.store.get_presence_stream_id_gen() + ctx_mgr = presence_id_gen.get_next() + stream_id = self.get_success(ctx_mgr.__aenter__()) + + # Create the new token based on the stream ID above. + current_token = MultiWriterStreamToken.from_generator(presence_id_gen) + token = current_token.copy_and_advance(MultiWriterStreamToken(stream=stream_id)) + + # Function under test + wait_d = defer.ensureDeferred( + self.notifier.wait_for_multi_writer_stream_token(token, presence_id_gen) + ) + + # This should block waiting for the stream to update + # + # Advance time a little bit to make the + # `wait_for_multi_writer_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_multi_writer_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_multi_writer_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". + presence_id_gen = self.store.get_presence_stream_id_gen() + ctx_mgr = presence_id_gen.get_next() + stream_id = self.get_success(ctx_mgr.__aenter__()) + + # Create the new token based on the stream ID above. + current_token = MultiWriterStreamToken.from_generator(presence_id_gen) + token = current_token.copy_and_advance(MultiWriterStreamToken(stream=stream_id)) + + # Function under test + wait_d = defer.ensureDeferred( + self.notifier.wait_for_multi_writer_stream_token(token, presence_id_gen) + ) + # Advance time a little bit to make the + # `wait_for_multi_writer_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_multi_writer_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_multi_writer_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) + + def test_wait_for_multi_writer_stream_token_with_invalid_future_sync_token( + self, + ) -> None: + """Like the previous test, except we give a token that has a stream + position ahead of what is in the DB, i.e. its invalid and we shouldn't + wait for the stream to advance (as it may never do so). + + This can happen due to older versions of Synapse giving out stream + positions without persisting them in the DB, and so on restart the + stream would get reset back to an older position. + """ + presence_id_gen = self.store.get_presence_stream_id_gen() + + # Create a token and advance one of the streams. + current_token = MultiWriterStreamToken.from_generator(presence_id_gen) + token = current_token.copy_and_advance( + MultiWriterStreamToken(stream=current_token.get_max_stream_pos() + 1) + ) + + # Function under test + wait_d = defer.ensureDeferred( + self.notifier.wait_for_multi_writer_stream_token(token, presence_id_gen) + ) + + # Expect to fail. We expect callers to sanitize/validate the tokens they give to + # `wait_for_multi_writer_stream_token` to ensure they aren't in the future. + self.get_failure(wait_d, exc=AssertionError) From edfcb539cac06ce5cd732fdcf26ea385fd932169 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 2 Apr 2026 22:27:12 -0500 Subject: [PATCH 11/19] Align existing test --- tests/handlers/test_sync.py | 14 ++++++++++++-- tests/test_notifier.py | 7 ++++--- 2 files changed, 16 insertions(+), 5 deletions(-) diff --git a/tests/handlers/test_sync.py b/tests/handlers/test_sync.py index 18ec2ca6b6d..21525c4a98a 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 index 42ff76eab95..63ec6404644 100644 --- a/tests/test_notifier.py +++ b/tests/test_notifier.py @@ -137,9 +137,10 @@ def test_wait_for_multi_writer_stream_token_with_future_sync_token_timeout( def test_wait_for_multi_writer_stream_token_with_invalid_future_sync_token( self, ) -> None: - """Like the previous test, except we give a token that has a stream - position ahead of what is in the DB, i.e. its invalid and we shouldn't - wait for the stream to advance (as it may never do so). + """ + Like `test_wait_for_multi_writer_stream_token_with_future_sync_token`, except we + give a token that has a stream position ahead of what is in the DB, i.e. its + invalid and we shouldn't wait for the stream to advance (as it may never do so). This can happen due to older versions of Synapse giving out stream positions without persisting them in the DB, and so on restart the From 5c7f96ecfd645eeebf9a9769f8562645c7b9a49d Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Wed, 8 Apr 2026 11:07:35 -0500 Subject: [PATCH 12/19] Note future about `M_UNKNOWN_POS` See https://github.com/element-hq/synapse/pull/19644#discussion_r3051619047 --- synapse/handlers/sliding_sync/__init__.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/synapse/handlers/sliding_sync/__init__.py b/synapse/handlers/sliding_sync/__init__.py index e23a0ee1ff2..3ef1409e2fd 100644 --- a/synapse/handlers/sliding_sync/__init__.py +++ b/synapse/handlers/sliding_sync/__init__.py @@ -158,6 +158,9 @@ async def wait_for_sync_for_user( # 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 From 9c9a1d233cec8733e621d835571184ce560492b7 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Wed, 8 Apr 2026 19:14:32 -0500 Subject: [PATCH 13/19] Update error language: unable -> refuse --- synapse/notifier.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/synapse/notifier.py b/synapse/notifier.py index 6fe66797beb..fe3e4a3d668 100644 --- a/synapse/notifier.py +++ b/synapse/notifier.py @@ -864,8 +864,9 @@ async def wait_for_stream_token(self, stream_token: StreamToken) -> bool: original_stream_token = stream_token max_token = await self.event_sources.bound_future_token(stream_token) assert stream_token.is_before_or_eq(max_token), ( - f"Unable to wait for invalid future stream token (token={original_stream_token} has positions " - "ahead of our max persisted position {max_token})" + f"Refusing to wait for invalid future stream token (token={original_stream_token} " + "that has positions ahead of our max persisted position {max_token}) " + "(Synapse programming error)" ) # Start waiting until we've caught up to the `stream_token` @@ -927,8 +928,9 @@ async def wait_for_multi_writer_stream_token( # bad patterns upstream where people can continue to use the unbounded token. max_persisted_position = await id_gen.get_max_allocated_token() assert max_persisted_position >= token.get_max_stream_pos(), ( - f"Unable to wait for invalid future token (token={token} has positions " - "ahead of our max persisted position {max_persisted_position})" + f"Refusing to wait for invalid future token (token={token} " + "that has positions ahead of our max persisted position {max_persisted_position}) " + "(Synapse programming error)" ) # Start waiting until we've caught up to the `stream_token` From 83d6bdbd77baab48570462b1fe0f9a68ca7145d8 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 7 May 2026 14:35:48 -0500 Subject: [PATCH 14/19] Remove `wait_for_multi_writer_stream_token` See https://github.com/element-hq/synapse/pull/19644#discussion_r3029861226 --- rust/src/canonical_json.rs | 2 +- synapse/notifier.py | 67 ------------------------------ synapse/types/__init__.py | 7 ++++ tests/test_notifier.py | 83 +++++++++++++++++++------------------- 4 files changed, 50 insertions(+), 109 deletions(-) diff --git a/rust/src/canonical_json.rs b/rust/src/canonical_json.rs index ff1fcd3ee41..4abd0018479 100644 --- a/rust/src/canonical_json.rs +++ b/rust/src/canonical_json.rs @@ -824,7 +824,7 @@ mod tests { // Serialize the keys in the reverse order. for (c, _) in ascii_order.iter().rev() { - map_serializer.serialize_entry(c.into(), &1).unwrap(); + map_serializer.serialize_entry(c, &1).unwrap(); } SerializeMap::end(map_serializer).unwrap(); diff --git a/synapse/notifier.py b/synapse/notifier.py index ef53625aae5..656c18f6d1a 100644 --- a/synapse/notifier.py +++ b/synapse/notifier.py @@ -47,9 +47,6 @@ from synapse.logging.context import PreserveLoggingContext from synapse.logging.opentracing import log_kv, start_active_span from synapse.metrics import SERVER_NAME_LABEL, LaterGauge -from synapse.storage.util.id_generators import ( - MultiWriterIdGenerator, -) from synapse.streams.config import PaginationConfig from synapse.types import ( ISynapseReactor, @@ -895,70 +892,6 @@ async def wait_for_stream_token(self, stream_token: StreamToken) -> bool: # TODO: be better await self.clock.sleep(Duration(milliseconds=500)) - async def wait_for_multi_writer_stream_token( - self, - token: MultiWriterStreamToken, - id_gen: MultiWriterIdGenerator, - ) -> bool: - """ - 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. - - Returns: - True when this worker has caught up - False when we timed out waiting - """ - current_token = MultiWriterStreamToken.from_generator(id_gen) - # Return early if we are already caught up - if token.is_before_or_eq(current_token): - return True - - # Assert as we consider this a Synapse programming error. We shouldn't be - # handing out invalid future tokens and tokens should be validated before it - # reaches this point. - # - # We consider a token invalid, if the token has positions ahead of our persisted - # positions in the database - # - # Previously, we would bound the tokens within this function but that leads to - # bad patterns upstream where people can continue to use the unbounded token. - max_persisted_position = await id_gen.get_max_allocated_token() - assert max_persisted_position >= token.get_max_stream_pos(), ( - f"Refusing to wait for invalid future token (token={token} " - "that has positions ahead of our max persisted position {max_persisted_position}) " - "(Synapse programming error)" - ) - - # Start waiting until we've caught up to the `stream_token` - start = self.clock.time_msec() - logged = False - while True: - current_token = MultiWriterStreamToken.from_generator(id_gen) - if token.is_before_or_eq(current_token): - return True - - now = self.clock.time_msec() - - # Timed out - if now - start > 10_000: - return False - - if not logged: - logger.info( - "Waiting for current token to reach %s; currently at %s", - token, - current_token, - ) - logged = True - - # TODO: be better - await self.clock.sleep(Duration(milliseconds=500)) - async def _get_room_ids( self, user: UserID, explicit_room_id: str | None ) -> tuple[StrCollection, bool]: diff --git a/synapse/types/__init__.py b/synapse/types/__init__.py index 8b005ef84d5..df63da874a6 100644 --- a/synapse/types/__init__.py +++ b/synapse/types/__init__.py @@ -1312,6 +1312,13 @@ def is_before_or_eq(self, other_token: "StreamToken") -> bool: self_value = self.get_field(key) other_value = other_token.get_field(key) + logger.info( + "asdf key=%s self_value=%s, other_value=%s", + key, + self_value, + other_value, + ) + if isinstance(self_value, RoomStreamToken): assert isinstance(other_value, RoomStreamToken) if not self_value.is_before_or_eq(other_value): diff --git a/tests/test_notifier.py b/tests/test_notifier.py index 63ec6404644..fdd44d7b6be 100644 --- a/tests/test_notifier.py +++ b/tests/test_notifier.py @@ -16,7 +16,7 @@ from twisted.internet.testing import MemoryReactor from synapse.server import HomeServer -from synapse.types import MultiWriterStreamToken +from synapse.types import MultiWriterStreamToken, StreamKeyType, StreamToken from synapse.util.clock import Clock from synapse.util.duration import Duration @@ -30,24 +30,23 @@ 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_multi_writer_stream_token_with_caught_up_token(self) -> None: + 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 - presence_id_gen = self.store.get_presence_stream_id_gen() - token = MultiWriterStreamToken.from_generator(presence_id_gen) + 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_multi_writer_stream_token(token, presence_id_gen) - ) + 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_multi_writer_stream_token_with_future_sync_token(self) -> None: + 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. @@ -57,23 +56,24 @@ def test_wait_for_multi_writer_stream_token_with_future_sync_token(self) -> None """ # We simulate a lagging stream by getting a stream ID from the ID gen # and then waiting to mark it as "persisted". - presence_id_gen = self.store.get_presence_stream_id_gen() - ctx_mgr = presence_id_gen.get_next() - stream_id = self.get_success(ctx_mgr.__aenter__()) + 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_token = MultiWriterStreamToken.from_generator(presence_id_gen) - token = current_token.copy_and_advance(MultiWriterStreamToken(stream=stream_id)) + 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_multi_writer_stream_token(token, presence_id_gen) - ) + 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_multi_writer_stream_token(...)` sleep loop iterate. + # `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) @@ -82,7 +82,7 @@ def test_wait_for_multi_writer_stream_token_with_future_sync_token(self) -> None self.get_success(ctx_mgr.__aexit__(None, None, None)) # Advance time to make another iteration of - # `wait_for_multi_writer_stream_token(...)` sleep loop so it sees that we're + # `wait_for_stream_token(...)` sleep loop so it sees that we're # finally caught up now. self.reactor.advance(Duration(seconds=1).as_secs()) @@ -90,7 +90,7 @@ def test_wait_for_multi_writer_stream_token_with_future_sync_token(self) -> None wait_result = self.get_success(wait_d) self.assertEqual(wait_result, True) - def test_wait_for_multi_writer_stream_token_with_future_sync_token_timeout( + def test_wait_for_stream_token_with_future_sync_token_timeout( self, ) -> None: """ @@ -103,42 +103,43 @@ def test_wait_for_multi_writer_stream_token_with_future_sync_token_timeout( """ # We simulate a lagging stream by getting a stream ID from the ID gen # and then waiting to mark it as "persisted". - presence_id_gen = self.store.get_presence_stream_id_gen() - ctx_mgr = presence_id_gen.get_next() - stream_id = self.get_success(ctx_mgr.__aenter__()) + 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_token = MultiWriterStreamToken.from_generator(presence_id_gen) - token = current_token.copy_and_advance(MultiWriterStreamToken(stream=stream_id)) + 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_multi_writer_stream_token(token, presence_id_gen) - ) + wait_d = defer.ensureDeferred(self.notifier.wait_for_stream_token(token)) # Advance time a little bit to make the - # `wait_for_multi_writer_stream_token(...)` sleep loop record 0 as the `start` time. + # `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_multi_writer_stream_token(...)` sleep loop iterate. + # `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_multi_writer_stream_token(...)` sleep loop give up. + # `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) - def test_wait_for_multi_writer_stream_token_with_invalid_future_sync_token( + def test_wait_for_stream_token_with_invalid_future_sync_token( self, ) -> None: """ - Like `test_wait_for_multi_writer_stream_token_with_future_sync_token`, except we + Like `test_wait_for_stream_token_with_future_sync_token`, except we give a token that has a stream position ahead of what is in the DB, i.e. its invalid and we shouldn't wait for the stream to advance (as it may never do so). @@ -146,19 +147,19 @@ def test_wait_for_multi_writer_stream_token_with_invalid_future_sync_token( positions without persisting them in the DB, and so on restart the stream would get reset back to an older position. """ - presence_id_gen = self.store.get_presence_stream_id_gen() - # Create a token and advance one of the streams. - current_token = MultiWriterStreamToken.from_generator(presence_id_gen) - token = current_token.copy_and_advance( - MultiWriterStreamToken(stream=current_token.get_max_stream_pos() + 1) + receipt_id_gen = self.store.get_receipts_stream_id_gen() + current_receipt_token = MultiWriterStreamToken.from_generator(receipt_id_gen) + receipt_token = current_receipt_token.copy_and_advance( + MultiWriterStreamToken( + stream=current_receipt_token.get_max_stream_pos() + 1 + ) ) + token = StreamToken.START.copy_and_advance(StreamKeyType.RECEIPT, receipt_token) # Function under test - wait_d = defer.ensureDeferred( - self.notifier.wait_for_multi_writer_stream_token(token, presence_id_gen) - ) + wait_d = defer.ensureDeferred(self.notifier.wait_for_stream_token(token)) # Expect to fail. We expect callers to sanitize/validate the tokens they give to - # `wait_for_multi_writer_stream_token` to ensure they aren't in the future. + # `wait_for_stream_token` to ensure they aren't in the future. self.get_failure(wait_d, exc=AssertionError) From a2a9d4215a34134824f2ff94afa43482b4cabe62 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 7 May 2026 14:50:06 -0500 Subject: [PATCH 15/19] Don't `assert` invalid future token in `wait_for_stream_token(...)` See https://github.com/element-hq/synapse/pull/19644#discussion_r3029874002 --- synapse/notifier.py | 34 +++++++++++++++++----------------- tests/test_notifier.py | 29 ----------------------------- 2 files changed, 17 insertions(+), 46 deletions(-) diff --git a/synapse/notifier.py b/synapse/notifier.py index 656c18f6d1a..dfd2178f933 100644 --- a/synapse/notifier.py +++ b/synapse/notifier.py @@ -841,6 +841,23 @@ async def wait_for_stream_token(self, stream_token: StreamToken) -> bool: 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). + + Previously, we would `bound_future_token(...)` within this function but that + leads to bad patterns upstream where people can continue to use the unbounded + token. + + 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 @@ -850,23 +867,6 @@ async def wait_for_stream_token(self, stream_token: StreamToken) -> bool: if stream_token.is_before_or_eq(current_token): return True - # Assert as we consider this a Synapse programming error. We shouldn't be - # handing out invalid future tokens and tokens should be validated before it - # reaches this point. - # - # We consider a token invalid, if the token has positions ahead of our persisted - # positions in the database - # - # Previously, we would bound the tokens within this function but that leads to - # bad patterns upstream where people can continue to use the unbounded token. - original_stream_token = stream_token - max_token = await self.event_sources.bound_future_token(stream_token) - assert stream_token.is_before_or_eq(max_token), ( - f"Refusing to wait for invalid future stream token (token={original_stream_token} " - "that has positions ahead of our max persisted position {max_token}) " - "(Synapse programming error)" - ) - # Start waiting until we've caught up to the `stream_token` start = self.clock.time_msec() logged = False diff --git a/tests/test_notifier.py b/tests/test_notifier.py index fdd44d7b6be..c65134e8325 100644 --- a/tests/test_notifier.py +++ b/tests/test_notifier.py @@ -134,32 +134,3 @@ def test_wait_for_stream_token_with_future_sync_token_timeout( # Make sure we gave up waiting and not caught-up (False) wait_result = self.get_success(wait_d) self.assertEqual(wait_result, False) - - def test_wait_for_stream_token_with_invalid_future_sync_token( - self, - ) -> None: - """ - Like `test_wait_for_stream_token_with_future_sync_token`, except we - give a token that has a stream position ahead of what is in the DB, i.e. its - invalid and we shouldn't wait for the stream to advance (as it may never do so). - - This can happen due to older versions of Synapse giving out stream - positions without persisting them in the DB, and so on restart the - stream would get reset back to an older position. - """ - # Create a token and advance one of the streams. - receipt_id_gen = self.store.get_receipts_stream_id_gen() - current_receipt_token = MultiWriterStreamToken.from_generator(receipt_id_gen) - receipt_token = current_receipt_token.copy_and_advance( - MultiWriterStreamToken( - stream=current_receipt_token.get_max_stream_pos() + 1 - ) - ) - token = StreamToken.START.copy_and_advance(StreamKeyType.RECEIPT, receipt_token) - - # Function under test - wait_d = defer.ensureDeferred(self.notifier.wait_for_stream_token(token)) - - # Expect to fail. We expect callers to sanitize/validate the tokens they give to - # `wait_for_stream_token` to ensure they aren't in the future. - self.get_failure(wait_d, exc=AssertionError) From 1b3ac3992575eaec864e1259ba508740cf4ab2ea Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 7 May 2026 14:55:31 -0500 Subject: [PATCH 16/19] Remove changelog which mentions `wait_for_multi_writer_stream_token` --- changelog.d/19644.misc | 1 - 1 file changed, 1 deletion(-) delete mode 100644 changelog.d/19644.misc diff --git a/changelog.d/19644.misc b/changelog.d/19644.misc deleted file mode 100644 index d3a7a12fc35..00000000000 --- a/changelog.d/19644.misc +++ /dev/null @@ -1 +0,0 @@ -Add new `Notifier.wait_for_multi_writer_stream_token(...)` utility to make it easy wait for the worker to catch up to any kind of token. From d9d5f54e1d35216b029154e82ade8c953ab6e889 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 7 May 2026 14:56:06 -0500 Subject: [PATCH 17/19] Revert stray automatic Rust change --- rust/src/canonical_json.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rust/src/canonical_json.rs b/rust/src/canonical_json.rs index 4abd0018479..ff1fcd3ee41 100644 --- a/rust/src/canonical_json.rs +++ b/rust/src/canonical_json.rs @@ -824,7 +824,7 @@ mod tests { // Serialize the keys in the reverse order. for (c, _) in ascii_order.iter().rev() { - map_serializer.serialize_entry(c, &1).unwrap(); + map_serializer.serialize_entry(c.into(), &1).unwrap(); } SerializeMap::end(map_serializer).unwrap(); From a8e81d69f2f61d0a66c9b7cc9aa6fdc9d5cbf30f Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 7 May 2026 15:58:21 -0500 Subject: [PATCH 18/19] Add more context --- synapse/notifier.py | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) diff --git a/synapse/notifier.py b/synapse/notifier.py index dfd2178f933..6a057ac09fa 100644 --- a/synapse/notifier.py +++ b/synapse/notifier.py @@ -847,11 +847,18 @@ async def wait_for_stream_token(self, stream_token: StreamToken) -> bool: 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). - - Previously, we would `bound_future_token(...)` within this function but that - leads to bad patterns upstream where people can continue to use the unbounded - token. + 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 From 6acc2a9eea942710fac2e11d430435028cb302e1 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Wed, 13 May 2026 12:21:27 -0500 Subject: [PATCH 19/19] Remove leftover debug logging As spotted in https://github.com/element-hq/synapse/pull/19644#discussion_r3233096398 --- synapse/types/__init__.py | 7 ------- 1 file changed, 7 deletions(-) diff --git a/synapse/types/__init__.py b/synapse/types/__init__.py index df63da874a6..8b005ef84d5 100644 --- a/synapse/types/__init__.py +++ b/synapse/types/__init__.py @@ -1312,13 +1312,6 @@ def is_before_or_eq(self, other_token: "StreamToken") -> bool: self_value = self.get_field(key) other_value = other_token.get_field(key) - logger.info( - "asdf key=%s self_value=%s, other_value=%s", - key, - self_value, - other_value, - ) - if isinstance(self_value, RoomStreamToken): assert isinstance(other_value, RoomStreamToken) if not self_value.is_before_or_eq(other_value):