Don't wait for streams when asking for stream updates

synapse/replication/http/_base.py

@@ -98,6 +98,9 @@ class ReplicationEndpoint(metaclass=abc.ABCMeta):
             a connection error is received.
         RETRY_ON_CONNECT_ERROR_ATTEMPTS (int): Number of attempts to retry when
             receiving connection errors, each will backoff exponentially longer.
+        WAIT_FOR_STREAMS (bool): Whether to wait for replication streams to
+            catch up before processing the request and/or response. Defaults to
+            True.
     """
 
     NAME: str = abc.abstractproperty()  # type: ignore
@@ -108,6 +111,8 @@ class ReplicationEndpoint(metaclass=abc.ABCMeta):
     RETRY_ON_CONNECT_ERROR = True
     RETRY_ON_CONNECT_ERROR_ATTEMPTS = 5  # =63s (2^6-1)
 
+    WAIT_FOR_STREAMS = True
+
     def __init__(self, hs: "HomeServer"):
         if self.CACHE:
             self.response_cache: ResponseCache[str] = ResponseCache(
@@ -231,7 +236,7 @@ async def send_request(*, instance_name: str = "master", **kwargs: Any) -> Any:
 
                 data = await cls._serialize_payload(**kwargs)
 
-                if cls.METHOD != "GET":
+                if cls.METHOD != "GET" and cls.WAIT_FOR_STREAMS:
                     # Include the current stream positions that we write to. We
                     # don't do this for GETs as they don't have a body, and we
                     # generally assume that a GET won't rely on data we have
@@ -436,9 +441,10 @@ async def _check_auth_and_handle(
         if _STREAM_POSITION_KEY in response:
             raise Exception("data to send contains %r key", _STREAM_POSITION_KEY)
 
-        response[_STREAM_POSITION_KEY] = {
-            stream.NAME: stream.current_token(self._instance_name)
-            for stream in self._streams
-        }
+        if self.WAIT_FOR_STREAMS:
+            response[_STREAM_POSITION_KEY] = {
+                stream.NAME: stream.current_token(self._instance_name)
+                for stream in self._streams
+            }
 
         return code, response

synapse/replication/http/streams.py

@@ -54,6 +54,10 @@ class ReplicationGetStreamUpdates(ReplicationEndpoint):
     PATH_ARGS = ("stream_name",)
     METHOD = "GET"
 
+    # We don't want to wait for replication streams to catch up, as this gets
+    # called in the process of catching replication streams up.
+    WAIT_FOR_STREAMS = False
+
     def __init__(self, hs: "HomeServer"):
         super().__init__(hs)
 

tests/replication/http/test__base.py

@@ -54,6 +54,7 @@ class UncancellableReplicationEndpoint(ReplicationEndpoint):
     NAME = "uncancellable_sleep"
     PATH_ARGS = ()
     CACHE = False
+    WAIT_FOR_STREAMS = False
 
     def __init__(self, hs: HomeServer):
         super().__init__(hs)