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