Fix wait_for_stream_position for multiple waiters.

changelog.d/8196.misc

@@ -0,0 +1 @@
+Fix `wait_for_stream_position` to allow multiple waiters on same stream ID.

synapse/python_dependencies.py

@@ -66,8 +66,8 @@
     "msgpack>=0.5.2",
     "phonenumbers>=8.2.0",
     "prometheus_client>=0.0.18,<0.9.0",
-    # we use attr.validators.deep_iterable, which arrived in 19.1.0
-    "attrs>=19.1.0",
+    # we use attrs `order` param, which arrived in 19.2.0
+    "attrs>=19.2.0",
     "netaddr>=0.7.18",
     "Jinja2>=2.9",
     "bleach>=1.4.3",

synapse/replication/tcp/client.py

@@ -18,6 +18,8 @@
 import logging
 from typing import TYPE_CHECKING, Dict, List, Tuple
 
+import attr
+
 from twisted.internet.defer import Deferred
 from twisted.internet.protocol import ReconnectingClientFactory
 
@@ -109,9 +111,7 @@ def __init__(self, hs: "HomeServer"):
 
         # Map from stream to list of deferreds waiting for the stream to
         # arrive at a particular position. The lists are sorted by stream position.
-        self._streams_to_waiters = (
-            {}
-        )  # type: Dict[str, List[Tuple[int, Deferred[None]]]]
+        self._streams_to_waiters = {}  # type: Dict[str, List[_WaitForPositionEntry]]
 
     async def on_rdata(
         self, stream_name: str, instance_name: str, token: int, rows: list
@@ -169,11 +169,11 @@ async def on_rdata(
         # `len(list)` works for both cases.
         index_of_first_deferred_not_called = len(waiting_list)
 
-        for idx, (position, deferred) in enumerate(waiting_list):
-            if position <= token:
+        for idx, entry in enumerate(waiting_list):
+            if entry.position <= token:
                 try:
                     with PreserveLoggingContext():
-                        deferred.callback(None)
+                        entry.deferred.callback(None)
                 except Exception:
                     # The deferred has been cancelled or timed out.
                     pass
@@ -221,7 +221,7 @@ async def wait_for_stream_position(
 
         # We insert into the list using heapq as it is more efficient than
         # pushing then resorting each time.
-        heapq.heappush(waiting_list, (position, deferred))
+        heapq.heappush(waiting_list, _WaitForPositionEntry(position, deferred))
 
         # We measure here to get in flight counts and average waiting time.
         with Measure(self._clock, "repl.wait_for_stream_position"):
@@ -230,3 +230,15 @@ async def wait_for_stream_position(
             logger.info(
                 "Finished waiting for repl stream %r to reach %s", stream_name, position
             )
+
+
+@attr.s(
+    frozen=True, slots=True, order=True,
+)
+class _WaitForPositionEntry:
+    """Entry for type `_streams_to_waiters` that is comparable. A tuple can't
+    be used as `Deferred` is not comparable.
+    """
+
+    position = attr.ib(type=int)
+    deferred = attr.ib(type=Deferred, order=False)