diff --git a/synapse/replication/tcp/handler.py b/synapse/replication/tcp/handler.py
index 615f1828dd..9aba1cd451 100644
--- a/synapse/replication/tcp/handler.py
+++ b/synapse/replication/tcp/handler.py
@@ -537,7 +537,7 @@ class ReplicationCommandHandler:
# Ignore POSITION that are just our own echoes
return
- logger.info("Handling '%s %s'", cmd.NAME, cmd.to_line())
+ logger.debug("Handling '%s %s'", cmd.NAME, cmd.to_line())
self._add_command_to_stream_queue(conn, cmd)
@@ -567,6 +567,11 @@ class ReplicationCommandHandler:
# between then and now.
missing_updates = cmd.prev_token != current_token
while missing_updates:
+ # Note: There may very well not be any new updates, but we check to
+ # make sure. This can particularly happen for the event stream where
+ # event persisters continuously send `POSITION`. See `resource.py`
+ # for why this can happen.
+
logger.info(
"Fetching replication rows for '%s' between %i and %i",
stream_name,
@@ -590,7 +595,7 @@ class ReplicationCommandHandler:
[stream.parse_row(row) for row in rows],
)
- logger.info("Caught up with stream '%s' to %i", stream_name, cmd.new_token)
+ logger.info("Caught up with stream '%s' to %i", stream_name, cmd.new_token)
# We've now caught up to position sent to us, notify handler.
await self._replication_data_handler.on_position(
diff --git a/synapse/replication/tcp/resource.py b/synapse/replication/tcp/resource.py
index c6870df8f9..99f09669f0 100644
--- a/synapse/replication/tcp/resource.py
+++ b/synapse/replication/tcp/resource.py
@@ -204,6 +204,15 @@ class ReplicationStreamer:
# turns out that e.g. account data streams share
# their "current token" with each other, meaning
# that it is *not* safe to send a POSITION.
+
+ # Note: `last_token` may not *actually* be the
+ # last token we sent out in a RDATA or POSITION.
+ # This can happen if we sent out an RDATA for
+ # position X when our current token was say X+1.
+ # Other workers will see RDATA for X and then a
+ # POSITION with last token of X+1, which will
+ # cause them to check if there were any missing
+ # updates between X and X+1.
logger.info(
"Sending position: %s -> %s",
stream.NAME,
|
