Simplify token replication logic

1 files changed, 14 insertions, 23 deletions

diff --git a/synapse/replication/tcp/protocol.py b/synapse/replication/tcp/protocol.py
index dac4fbeef7..55630ba9a7 100644
--- a/synapse/replication/tcp/protocol.py
+++ b/synapse/replication/tcp/protocol.py
@@ -478,7 +478,7 @@ class ServerReplicationStreamProtocol(BaseReplicationStreamProtocol):
 
             # Now we can send any updates that came in while we were subscribing
             pending_rdata = self.pending_rdata.pop(stream_name, [])
-            batch_updates = []
+            updates = []
             for token, update in pending_rdata:
                 # If the token is null, it is part of a batch update. Batches
                 # are multiple updates that share a single token. To denote
@@ -489,34 +489,25 @@ class ServerReplicationStreamProtocol(BaseReplicationStreamProtocol):
                 # final token.
                 if token is None:
                     # Store this update as part of a batch
-                    batch_updates.append(update)
+                    updates.append(update)
                     continue
 
-                if len(batch_updates) > 0:
-                    # There is an ongoing batch and this is the end
-                    if current_token <= current_token:
-                        # This batch is older than current_token, dismiss it
-                        batch_updates = []
-                    else:
-                        # This is the end of the batch. Append final update of
-                        # this batch before sending
-                        batch_updates.append(update)
-
-                        # Send all updates that are part of this batch with the
-                        # found token
-                        for update in batch_updates:
-                            self.send_command(RdataCommand(stream_name, token, update))
-
-                        # Clear saved batch updates
-                        batch_updates = []
+                if token <= current_token:
+                    # This update or batch of updates is older than
+                    # current_token, dismiss it
+                    updates = []
                     continue
 
-                # This is an update that's not part of a batch.
-                #
-                # Only send updates newer than the current token
-                if token > current_token:
+                updates.append(update)
+
+                # Send all updates that are part of this batch with the
+                # found token
+                for update in updates:
                     self.send_command(RdataCommand(stream_name, token, update))
 
+                # Clear stored updates
+                updates = []
+
             # They're now fully subscribed
             self.replication_streams.add(stream_name)
         except Exception as e: