From 53bd383fe8d8befe39844a5b4ee8bae8d815bc53 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Wed, 12 Aug 2026 12:44:36 +0000 Subject: [PATCH] Report each stream's current position as a metric When a stream advances in the database but stops being replicated to a process, that process's view of the stream freezes. Requests that wait for it to catch up to a token issued by another worker then time out and return empty responses indefinitely (see #20080), and nothing exported said so. Report `get_current_token` from every ID generator, on every process. The value is comparable between processes, so a stream that has stopped reaching one of them shows up as divergence with no client traffic needed. It is also the position that `wait_for_stream_token` waits on, so its divergence is the failure itself rather than a proxy for it. Being a watermark over gapless runs of persisted IDs, it also catches a single writer of a sharded stream going quiet, which a maximum across writers would hide behind the writers still being replicated. Co-Authored-By: Claude Opus 5 (1M context) --- changelog.d/20097.misc | 1 + synapse/storage/util/id_generators.py | 32 +++++++++++++++ tests/storage/test_id_generators.py | 57 ++++++++++++++++++++++++++- 3 files changed, 89 insertions(+), 1 deletion(-) create mode 100644 changelog.d/20097.misc diff --git a/changelog.d/20097.misc b/changelog.d/20097.misc new file mode 100644 index 0000000000..419bb794cc --- /dev/null +++ b/changelog.d/20097.misc @@ -0,0 +1 @@ +Add a `synapse_storage_stream_current_position` metric, reporting each stream's current position as each process sees it. diff --git a/synapse/storage/util/id_generators.py b/synapse/storage/util/id_generators.py index c9c339b235..88f04702b4 100644 --- a/synapse/storage/util/id_generators.py +++ b/synapse/storage/util/id_generators.py @@ -36,9 +36,11 @@ from typing import ( ) import attr +from prometheus_client import Gauge from sortedcontainers import SortedList, SortedSet from synapse.logging import issue9533_logger +from synapse.metrics import SERVER_NAME_LABEL from synapse.metrics.background_process_metrics import run_as_background_process from synapse.storage.database import ( DatabasePool, @@ -55,6 +57,13 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) +stream_current_position_gauge = Gauge( + "synapse_storage_stream_current_position", + "The stream's current position as this process sees it, i.e. what " + "`get_current_token` returns. Streams that count downwards report negative IDs", + labelnames=["stream_name", "instance_name", SERVER_NAME_LABEL], +) + T = TypeVar("T") @@ -232,6 +241,15 @@ class MultiWriterIdGenerator(AbstractStreamIdGenerator): self._writers = writers self._return_factor = 1 if positive else -1 + # The `instance_name` label is what separates one process's series from + # another's. In production the scrape labels would do that too, but tests + # run several homeservers in one process. + self._current_position_gauge = stream_current_position_gauge.labels( + stream_name=stream_name, + instance_name=instance_name, + **{SERVER_NAME_LABEL: server_name}, + ) + # We lock as some functions may be called from DB threads. self._lock = threading.Lock() @@ -340,6 +358,18 @@ class MultiWriterIdGenerator(AbstractStreamIdGenerator): # position with the current minimum. self._current_positions[self._instance_name] = self._persisted_upto_position + with self._lock: + self._report_current_position() + + def _report_current_position(self) -> None: + """Report the current position as a metric. + + Must be called with `_lock` held. + """ + self._current_position_gauge.set( + self._return_factor * self._persisted_upto_position + ) + def _load_current_ids( self, db_conn: LoggingDatabaseConnection, @@ -828,6 +858,8 @@ class MultiWriterIdGenerator(AbstractStreamIdGenerator): # do. break + self._report_current_position() + # Hacky debug logging to attempt to trace https://github.com/element-hq/synapse/issues/19795. # If this is the to-device stream, and we are a writer for that stream, log some stats if ( diff --git a/tests/storage/test_id_generators.py b/tests/storage/test_id_generators.py index 9a338607ee..6c01490dba 100644 --- a/tests/storage/test_id_generators.py +++ b/tests/storage/test_id_generators.py @@ -21,6 +21,7 @@ from twisted.internet.testing import MemoryReactor +from synapse.metrics import SERVER_NAME_LABEL from synapse.server import HomeServer from synapse.storage.database import ( DatabasePool, @@ -28,7 +29,10 @@ from synapse.storage.database import ( LoggingTransaction, ) from synapse.storage.types import Cursor -from synapse.storage.util.id_generators import MultiWriterIdGenerator +from synapse.storage.util.id_generators import ( + MultiWriterIdGenerator, + stream_current_position_gauge, +) from synapse.storage.util.sequence import ( LocalSequenceGenerator, PostgresSequenceGenerator, @@ -105,6 +109,24 @@ class MultiWriterIdGeneratorBase(HomeserverTestCase): ) return self.instances[instance_name] + def _get_reported_position(self, instance_name: str) -> int: + """The position the metric reports for the test stream on the given + process. + + The gauge outlives each test, so it still holds the label sets of ID + generators from earlier tests. + """ + for metric in stream_current_position_gauge.collect(): + for sample in metric.samples: + if ( + sample.labels["stream_name"] == "test_stream" + and sample.labels["instance_name"] == instance_name + and sample.labels[SERVER_NAME_LABEL] == self.hs.hostname + ): + return int(sample.value) + + raise AssertionError(f"No position reported for {instance_name}") + def _replicate(self, instance_name: str) -> None: """Similate a replication event for the given instance.""" @@ -225,6 +247,24 @@ class MultiWriterIdGeneratorTestCase(MultiWriterIdGeneratorBase): self.assertEqual(id_gen.get_positions(), {"master": 8}) self.assertEqual(id_gen.get_current_token_for_writer("master"), 8) + def test_current_position_metric(self) -> None: + """The reported position follows `get_current_token`.""" + + self._insert_rows("master", 7) + + id_gen = self._create_id_generator() + + self.assertEqual(self._get_reported_position("master"), 7) + + async def _get_next_async() -> None: + async with id_gen.get_next(): + pass + + self.get_success(_get_next_async()) + + self.assertEqual(id_gen.get_current_token(), 8) + self.assertEqual(self._get_reported_position("master"), 8) + def test_out_of_order_finish(self) -> None: """Test that IDs persisted out of order are correctly handled""" @@ -417,6 +457,21 @@ class WorkerMultiWriterIdGeneratorTestCase(MultiWriterIdGeneratorBase): id_gen.advance("second", 15) self.assertEqual(id_gen.get_persisted_upto_position(), 11) + def test_current_position_metric_diverges_when_not_replicated(self) -> None: + """A process that stops being told about a stream reports a position that + falls behind a process that is still being told about it. + """ + self._insert_row_with_id("writer", 3) + + # Two processes' view of a stream that a third process writes. + told = self._create_id_generator("told", writers=["writer"]) + self._create_id_generator("not_told", writers=["writer"]) + + told.advance("writer", 5) + + self.assertEqual(self._get_reported_position("told"), 5) + self.assertEqual(self._get_reported_position("not_told"), 3) + def test_get_persisted_upto_position_get_next(self) -> None: """Test that `get_persisted_upto_position` correctly tracks updates to positions when `get_next` is called.