Lift pausing on ratelimited requests to http layer (#18595)

When a request gets ratelimited we (optionally) wait ~500ms before
returning to mitigate clients that like to tightloop on request
failures. However, this is currently implemented by pausing request
processing when we check for ratelimits, which might be deep within
request processing, and e.g. while locks are held. Instead, let's hoist
the pause to the very top of the HTTP handler.

Hopefully, this mitigates the issue where a user sending lots of events
to a single room can see their requests time out due to the combination
of the linearizer and the pausing of the request. Instead, they should
see the requests 429 after ~500ms.

The first commit is a refactor to pass the `Clock` to `AsyncResource`,
the second commit is the behavioural change.

Author: erikjohnston

changelog.d/18595.misc

@@ -0,0 +1 @@
+Better handling of ratelimited requests.

synapse/api/errors.py

@@ -527,14 +527,19 @@ def error_dict(self, config: Optional["HomeServerConfig"]) -> "JsonDict":
 
 
 class LimitExceededError(SynapseError):
-    """A client has sent too many requests and is being throttled."""
+    """A client has sent too many requests and is being throttled.
+
+    Args:
+        pause: Optional time in seconds to pause before responding to the client.
+    """
 
     def __init__(
         self,
         limiter_name: str,
         code: int = 429,
         retry_after_ms: Optional[int] = None,
         errcode: str = Codes.LIMIT_EXCEEDED,
+        pause: Optional[float] = None,
     ):
         # Use HTTP header Retry-After to enable library-assisted retry handling.
         headers = (
@@ -545,6 +550,7 @@ def __init__(
         super().__init__(code, "Too Many Requests", errcode, headers=headers)
         self.retry_after_ms = retry_after_ms
         self.limiter_name = limiter_name
+        self.pause = pause
 
     def error_dict(self, config: Optional["HomeServerConfig"]) -> "JsonDict":
         return cs_error(self.msg, self.errcode, retry_after_ms=self.retry_after_ms)

synapse/api/ratelimiting.py

@@ -338,12 +338,10 @@ async def ratelimit(
         )
 
         if not allowed:
-            if pause:
-                await self.clock.sleep(pause)
-
             raise LimitExceededError(
                 limiter_name=self._limiter_name,
                 retry_after_ms=int(1000 * (time_allowed - time_now_s)),
+                pause=pause,
             )
 
 

synapse/http/additional_resource.py

@@ -53,7 +53,7 @@ def __init__(
             hs: homeserver
             handler: function to be called to handle the request.
         """
-        super().__init__()
+        super().__init__(hs.get_clock())
         self._handler = handler
 
     async def _async_render(self, request: Request) -> Optional[Tuple[int, Any]]:

synapse/http/proxy.py

@@ -106,7 +106,7 @@ class ProxyResource(_AsyncResource):
     isLeaf = True
 
     def __init__(self, reactor: ISynapseReactor, hs: "HomeServer"):
-        super().__init__(True)
+        super().__init__(hs.get_clock(), True)
 
         self.reactor = reactor
         self.agent = hs.get_federation_http_client().agent

synapse/http/server.py

@@ -67,14 +67,15 @@
 from synapse.api.errors import (
     CodeMessageException,
     Codes,
+    LimitExceededError,
     RedirectException,
     SynapseError,
     UnrecognizedRequestError,
 )
 from synapse.config.homeserver import HomeServerConfig
 from synapse.logging.context import defer_to_thread, preserve_fn, run_in_background
 from synapse.logging.opentracing import active_span, start_active_span, trace_servlet
-from synapse.util import json_encoder
+from synapse.util import Clock, json_encoder
 from synapse.util.caches import intern_dict
 from synapse.util.cancellation import is_function_cancellable
 from synapse.util.iterutils import chunk_seq
@@ -308,9 +309,10 @@ class _AsyncResource(resource.Resource, metaclass=abc.ABCMeta):
             context from the request the servlet is handling.
     """
 
-    def __init__(self, extract_context: bool = False):
+    def __init__(self, clock: Clock, extract_context: bool = False):
         super().__init__()
 
+        self._clock = clock
         self._extract_context = extract_context
 
     def render(self, request: "SynapseRequest") -> int:
@@ -329,7 +331,12 @@ async def _async_render_wrapper(self, request: "SynapseRequest") -> None:
             request.request_metrics.name = self.__class__.__name__
 
             with trace_servlet(request, self._extract_context):
-                callback_return = await self._async_render(request)
+                try:
+                    callback_return = await self._async_render(request)
+                except LimitExceededError as e:
+                    if e.pause:
+                        self._clock.sleep(e.pause)
+                    raise
 
                 if callback_return is not None:
                     code, response = callback_return
@@ -393,8 +400,10 @@ class DirectServeJsonResource(_AsyncResource):
     formatting responses and errors as JSON.
     """
 
-    def __init__(self, canonical_json: bool = False, extract_context: bool = False):
-        super().__init__(extract_context)
+    def __init__(
+        self, clock: Clock, canonical_json: bool = False, extract_context: bool = False
+    ):
+        super().__init__(clock, extract_context)
         self.canonical_json = canonical_json
 
     def _send_response(
@@ -450,8 +459,8 @@ def __init__(
         canonical_json: bool = True,
         extract_context: bool = False,
     ):
-        super().__init__(canonical_json, extract_context)
         self.clock = hs.get_clock()
+        super().__init__(self.clock, canonical_json, extract_context)
         # Map of path regex -> method -> callback.
         self._routes: Dict[Pattern[str], Dict[bytes, _PathEntry]] = {}
         self.hs = hs

synapse/rest/consent/consent_resource.py

@@ -81,7 +81,7 @@ class ConsentResource(DirectServeHtmlResource):
     """
 
     def __init__(self, hs: "HomeServer"):
-        super().__init__()
+        super().__init__(hs.get_clock())
 
         self.hs = hs
         self.store = hs.get_datastores().main

synapse/rest/synapse/client/federation_whitelist.py

@@ -44,7 +44,7 @@ class FederationWhitelistResource(DirectServeJsonResource):
     PATH = "/_synapse/client/v1/config/federation_whitelist"
 
     def __init__(self, hs: "HomeServer"):
-        super().__init__()
+        super().__init__(hs.get_clock())
 
         self._federation_whitelist = hs.config.federation.federation_domain_whitelist
 

synapse/rest/synapse/client/jwks.py

@@ -33,7 +33,7 @@
 
 class JwksResource(DirectServeJsonResource):
     def __init__(self, hs: "HomeServer"):
-        super().__init__(extract_context=True)
+        super().__init__(hs.get_clock(), extract_context=True)
 
         # Parameters that are allowed to be exposed in the public key.
         # This is done manually, because authlib's private to public key conversion

synapse/rest/synapse/client/new_user_consent.py

@@ -48,7 +48,7 @@ class NewUserConsentResource(DirectServeHtmlResource):
     """
 
     def __init__(self, hs: "HomeServer"):
-        super().__init__()
+        super().__init__(hs.get_clock())
         self._sso_handler = hs.get_sso_handler()
         self._server_name = hs.hostname
         self._consent_version = hs.config.consent.user_consent_version