Lagging workers can return incomplete data because they don't wait to catch up to the given token

[MadLittleMods]

Background

Spawning from #19644 which has roots in #19558 (comment),

When using workers and multiple writers for a stream, each worker has it's own view of the 'current' world. They notify each other of updates via replication and catch up over time.

One request can be routed to a worker and respond with a given next_batch token, then the next request can be routed to a different worker which may be lagging behind in the stream. When this happens, if the worker just naively serves the request and fetches data, it could be missing data in the gap between where it's lagging and the next_batch token. This results in flawed responses (not seamless).

The way to solve this is to have the worker wait until it catches up to the given next_batch token. This way the worker knows it at-least has all of the data that the previous worker did; and can fetch whatever new data it has is between the next_batch and wherever it's current position happens to be.

Do we wait anywhere?

We currently only wait for the /sync endpoints 😵:

synapse/synapse/handlers/sync.py

Lines 415 to 423 in 62f23fe

	if since_token is not None:
	# 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.
	start = self.clock.time_msec()
	if not await self.notifier.wait_for_stream_token(since_token):
	logger.warning(
	"Timed out waiting for worker to catch up. Returning empty response"
	)

synapse/synapse/handlers/sliding_sync/__init__.py

Lines 148 to 155 in 62f23fe

	# If we're working with a user-provided token, we need to make sure to wait for
	# this worker to catch up with the token so we don't skip past any incoming
	# events or future events if the user is nefariously, manually modifying the
	# token.
	if from_token is not None:
	# 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.

We need to wait everywhere

Given I had to create a whole new function wait_for_multi_writer_stream_token(...) in order for us to wait properly for some new endpoint, it's a clear sign that we're not properly waiting anywhere else. Because we're not waiting, all of other endpoints that fetch data with tokens are subject to flawed pagination and data fetching.

Also, as a best practice, we should probably be using the full StreamToken for every endpoint even though some endpoints may only fetch data in a single category. To explain the contrary, if for example, an endpoint was only dealing with receipts, we could maybe get away with using the single receipts token. But if the endpoint starts querying anything else like events, it will end up giving flawed data as the requests are passed from worker to worker. And this problem is hard to spot as the code evolves over time. Best to just use a token that snapshots the entire position of what that worker knows about (StreamToken).

Another complication on top of everything is backfilled events which use a negative stream_ordering. Since the negative side isn't tracked in the token, there will be consistency issues with workers across /messages requests.

Dev notes

Related docs:

• docs/development/synapse_architecture/streams.md

Brief discussion in the weekly Backend team meeting on 2026-04-07

Todo

• List out and audit all endpoints to see whether they wait (including admin endpoints)

Definition of done

• Ensure all endpoints wait until they've caught up to the given token

[reivilibre]

We have wait_for_stream_position, which is used in several places to wait for a stream to advance, but I think it is best-effort only and times out after 10 seconds or something, failing open.

[MadLittleMods]

See #19558 (comment) for why wait_for_stream_position isn't what we need.

I see it being usefully used in a few places though (specifically rooms sharded to specific workers) 👍

[MadLittleMods]

I've added the Z-Read-After-Write label as its probably useful to also find this issue when filtering but isn't an exact match. This is more of a Read-after-Read consistency issue across workers.