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) <noreply@anthropic.com>

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.

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
 

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 (

tests/test_notifier.py

@@ -11,10 +11,13 @@
 # <https://www.gnu.org/licenses/agpl-3.0.html>.
 
 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