From 4d8735aa6f2e06764e4260405726e0e4d72f25f9 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 26 Oct 2023 15:32:47 +0100 Subject: [PATCH] Correctly handle the different cases --- synapse/replication/tcp/resource.py | 21 +++++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/synapse/replication/tcp/resource.py b/synapse/replication/tcp/resource.py index 558728a73021..6ab09c2aadb3 100644 --- a/synapse/replication/tcp/resource.py +++ b/synapse/replication/tcp/resource.py @@ -204,13 +204,22 @@ async def _run_notifier_loop(self) -> None: # The token has advanced but there is no data to # send, so we send a `POSITION` to inform other # workers of the updated position. - - # We skip this for the `caches` stream as a) it - # generates a lot of traffic as every worker would - # echo each write, and b) nothing cares if a given - # worker's caches stream position lags. + # + # There are two reasons for this: 1) this instance + # requested a stream ID but didn't use it, or 2) + # this instance advanced its own stream position due + # to receiving notifications about other instances + # advancing their stream position. + + # We skip sending `POSITION` for the `caches` stream + # for the second case as a) it generates a lot of + # traffic as every worker would echo each write, and + # b) nothing cares if a given worker's caches stream + # position lags. if stream.NAME == CachesStream.NAME: - continue + # If there haven't been any writes + if stream.minimal_local_current_token() <= last_token: + continue # Note: `last_token` may not *actually* be the # last token we sent out in a RDATA or POSITION.