From e8f39a3f0e9d594d94e534317dc1093ad2c1aebb Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Wed, 12 Aug 2026 10:18:06 +0000 Subject: [PATCH] Count `wait_for_stream_token` timeouts per lagging stream When a worker's replication of a stream stalls, clients whose previous sync was served by a worker ahead of it hand us a token we will never catch up to, `wait_for_stream_token` times out after 10s and `/sync` returns an empty response indefinitely. Nothing metric-side pointed at which stream had stalled (see #20080, where it was `quarantined_media`). Count the timeouts against each stream the token is still ahead of us on, and name those streams (and their positions) in the logs instead of dumping two whole tokens. Co-Authored-By: Claude Opus 5 (1M context) --- changelog.d/20095.misc | 1 + synapse/notifier.py | 46 +++++++++++++++++++++++++--- synapse/types/__init__.py | 41 +++++++++++++++++-------- tests/test_notifier.py | 64 +++++++++++++++++++++++++++++++++++++++ 4 files changed, 135 insertions(+), 17 deletions(-) create mode 100644 changelog.d/20095.misc diff --git a/changelog.d/20095.misc b/changelog.d/20095.misc new file mode 100644 index 0000000000..7e0d0c58c7 --- /dev/null +++ b/changelog.d/20095.misc @@ -0,0 +1 @@ +Add a `synapse_notifier_wait_for_stream_token_timeouts` metric, counting the times a request gave up waiting for a worker to catch up to a stream token, labelled by the lagging stream. diff --git a/synapse/notifier.py b/synapse/notifier.py index e24d0ef5a2..8811be1a07 100644 --- a/synapse/notifier.py +++ b/synapse/notifier.py @@ -82,6 +82,14 @@ users_woken_by_stream_counter = Counter( labelnames=["stream", SERVER_NAME_LABEL], ) +wait_for_stream_token_timeout_counter = Counter( + "synapse_notifier_wait_for_stream_token_timeouts", + "Number of times we gave up waiting to catch up to a stream token, counted " + "once per lagging stream. `stream_key` is a `StreamToken` field name, which " + "is not always the replication stream name", + labelnames=["stream_key", SERVER_NAME_LABEL], +) + notifier_listeners_gauge = LaterGauge( name="synapse_notifier_listeners", @@ -103,6 +111,22 @@ notifier_users_gauge = LaterGauge( T = TypeVar("T") +def _describe_lagging_streams( + target_token: StreamToken, + current_token: StreamToken, + lagging_stream_keys: Collection[StreamKeyType], +) -> str: + """Describe how far each lagging stream has to go, for logging. + + e.g. `quarantined_media_key (at 4, waiting for 6)` + """ + return ", ".join( + f"{key.value} (at {current_token.get_field(key)}, " + f"waiting for {target_token.get_field(key)})" + for key in lagging_stream_keys + ) + + # TODO(paul): Should be shared somewhere def count(func: Callable[[T], bool], it: Iterable[T]) -> int: """Return the number of items in it for which func returns true.""" @@ -880,20 +904,34 @@ class Notifier: logged = False while True: current_token = self.event_sources.get_current_token() - if stream_token.is_before_or_eq(current_token): + lagging_stream_keys = stream_token.fields_behind(current_token) + if not lagging_stream_keys: return True now = self.clock.time_msec() # Timed out if now - start > 10_000: + for stream_key in lagging_stream_keys: + wait_for_stream_token_timeout_counter.labels( + stream_key=stream_key.value, + **{SERVER_NAME_LABEL: self.server_name}, + ).inc() + + logger.warning( + "Timed out waiting for current token to catch up on %s", + _describe_lagging_streams( + stream_token, current_token, lagging_stream_keys + ), + ) return False if not logged: logger.info( - "Waiting for current token to reach %s; currently at %s", - stream_token, - current_token, + "Waiting for current token to catch up on %s", + _describe_lagging_streams( + stream_token, current_token, lagging_stream_keys + ), ) logged = True diff --git a/synapse/types/__init__.py b/synapse/types/__init__.py index 7516847303..b3f7a7c2c6 100644 --- a/synapse/types/__init__.py +++ b/synapse/types/__init__.py @@ -31,6 +31,7 @@ from typing import ( Any, ClassVar, Final, + Iterator, Literal, Mapping, Match, @@ -1392,21 +1393,20 @@ class StreamToken: """Returns the stream ID for the given key.""" return getattr(self, key.value) - def is_before_or_eq(self, other_token: "StreamToken") -> bool: - """Wether this token is before the other token, i.e. every constituent - part is before the other. + def _iter_fields_behind( + self, other_token: "StreamToken" + ) -> Iterator[StreamKeyType]: + """The keys where `other_token` is behind this token. - Essentially it is `self <= other`. + A generator so that `is_before_or_eq` can stop at the first one. - Note: if `self.is_before_or_eq(other_token) is False` then that does not - imply that the reverse is True. + The typing key is never yielded. That stream is allowed to "reset", and + so comparisons don't really make sense as is. + TODO: Figure out a better way of tracking resets. """ for _, key in StreamKeyType.__members__.items(): if key == StreamKeyType.TYPING: - # Typing stream is allowed to "reset", and so comparisons don't - # really make sense as is. - # TODO: Figure out a better way of tracking resets. continue self_value = self.get_field(key) @@ -1415,17 +1415,32 @@ class StreamToken: if isinstance(self_value, RoomStreamToken): assert isinstance(other_value, RoomStreamToken) if not self_value.is_before_or_eq(other_value): - return False + yield key elif isinstance(self_value, MultiWriterStreamToken): assert isinstance(other_value, MultiWriterStreamToken) if not self_value.is_before_or_eq(other_value): - return False + yield key else: assert isinstance(other_value, int) if self_value > other_value: - return False + yield key - return True + def fields_behind(self, other_token: "StreamToken") -> list[StreamKeyType]: + """The keys where `other_token` is behind this token, i.e. the keys that + stop `self.is_before_or_eq(other_token)` from being True. + """ + return list(self._iter_fields_behind(other_token)) + + def is_before_or_eq(self, other_token: "StreamToken") -> bool: + """Whether this token is before the other token, i.e. every constituent + part is before the other. + + Essentially it is `self <= other`. + + Note: if `self.is_before_or_eq(other_token) is False` then that does not + imply that the reverse is True. + """ + return next(self._iter_fields_behind(other_token), None) is None def __str__(self) -> str: return ( diff --git a/tests/test_notifier.py b/tests/test_notifier.py index c65134e832..24dcf3890a 100644 --- a/tests/test_notifier.py +++ b/tests/test_notifier.py @@ -11,10 +11,13 @@ # . import logging +from collections import Counter from twisted.internet import defer from twisted.internet.testing import MemoryReactor +from synapse.metrics import SERVER_NAME_LABEL +from synapse.notifier import wait_for_stream_token_timeout_counter from synapse.server import HomeServer from synapse.types import MultiWriterStreamToken, StreamKeyType, StreamToken from synapse.util.clock import Clock @@ -114,6 +117,8 @@ class NotifierTestCase(tests.unittest.HomeserverTestCase): ) token = StreamToken.START.copy_and_advance(StreamKeyType.RECEIPT, receipt_token) + counts_before = self._get_timeout_counts() + # Function under test wait_d = defer.ensureDeferred(self.notifier.wait_for_stream_token(token)) # Advance time a little bit to make the @@ -134,3 +139,62 @@ class NotifierTestCase(tests.unittest.HomeserverTestCase): # Make sure we gave up waiting and not caught-up (False) wait_result = self.get_success(wait_d) self.assertEqual(wait_result, False) + + # Receipts was the only lagging stream, so it should be the only one counted. + self.assertEqual( + self._get_timeout_counts() - counts_before, + Counter({StreamKeyType.RECEIPT.value: 1}), + ) + + def test_wait_for_stream_token_timeout_counts_each_lagging_stream(self) -> None: + """ + Test that a timeout while lagging on more than one stream is counted against + each of them. + """ + + lagging_stream_keys = [StreamKeyType.RECEIPT, StreamKeyType.DEVICE_LIST] + + # Create a new token with the stream IDs artificially advanced far into + # the future. + token = StreamToken.START + token = token.copy_and_advance( + StreamKeyType.RECEIPT, MultiWriterStreamToken(stream=1000000000) + ) + token = token.copy_and_advance( + StreamKeyType.DEVICE_LIST, MultiWriterStreamToken(stream=10000000000) + ) + + counts_before = self._get_timeout_counts() + + wait_d = defer.ensureDeferred(self.notifier.wait_for_stream_token(token)) + + # Advance time to make the `wait_for_stream_token(...)` sleep loop + # iterate enough times to hit the the timeout. + for _ in range(11): + self.reactor.advance(Duration(seconds=1).as_secs()) + + # Make sure we gave up waiting and not caught-up (False) + self.assertEqual(self.get_success(wait_d), False) + + self.assertEqual( + self._get_timeout_counts() - counts_before, + Counter({stream_key.value: 1 for stream_key in lagging_stream_keys}), + ) + + def _get_timeout_counts(self) -> "Counter[str]": + """The `wait_for_stream_token` timeout counts for this server, keyed by the + `stream_key` label. + + The counter is process-wide, and so shared between tests. Compare against a + count taken before the code under test ran. + """ + counts: Counter[str] = Counter() + for metric in wait_for_stream_token_timeout_counter.collect(): + for sample in metric.samples: + if ( + sample.name.endswith("_total") + and sample.labels[SERVER_NAME_LABEL] == self.hs.hostname + ): + counts[sample.labels["stream_key"]] += int(sample.value) + + return counts