Merge remote-tracking branch 'matrix-org/develop' into travis/new-worker-docs

1 files changed, 58 insertions, 9 deletions

diff --git a/synapse/replication/http/send_event.py b/synapse/replication/http/send_event.py
index 468f4b68f4..bbe2f967b7 100644
--- a/synapse/replication/http/send_event.py
+++ b/synapse/replication/http/send_event.py
@@ -15,12 +15,17 @@
 
 from twisted.internet import defer
 
-from synapse.api.errors import SynapseError, MatrixCodeMessageException
+from synapse.api.errors import (
+    SynapseError, MatrixCodeMessageException, CodeMessageException,
+)
 from synapse.events import FrozenEvent
 from synapse.events.snapshot import EventContext
 from synapse.http.servlet import RestServlet, parse_json_object_from_request
+from synapse.util.async import sleep
+from synapse.util.caches.response_cache import ResponseCache
+from synapse.util.logcontext import make_deferred_yieldable, preserve_fn
 from synapse.util.metrics import Measure
-from synapse.types import Requester
+from synapse.types import Requester, UserID
 
 import logging
 import re
@@ -29,7 +34,8 @@ logger = logging.getLogger(__name__)
 
 
 @defer.inlineCallbacks
-def send_event_to_master(client, host, port, requester, event, context):
+def send_event_to_master(client, host, port, requester, event, context,
+                         ratelimit, extra_users):
     """Send event to be handled on the master
 
     Args:
@@ -39,8 +45,12 @@ def send_event_to_master(client, host, port, requester, event, context):
         requester (Requester)
         event (FrozenEvent)
         context (EventContext)
+        ratelimit (bool)
+        extra_users (list(UserID)): Any extra users to notify about event
     """
-    uri = "http://%s:%s/_synapse/replication/send_event" % (host, port,)
+    uri = "http://%s:%s/_synapse/replication/send_event/%s" % (
+        host, port, event.event_id,
+    )
 
     payload = {
         "event": event.get_pdu_json(),
@@ -48,10 +58,27 @@ def send_event_to_master(client, host, port, requester, event, context):
         "rejected_reason": event.rejected_reason,
         "context": context.serialize(event),
         "requester": requester.serialize(),
+        "ratelimit": ratelimit,
+        "extra_users": [u.to_string() for u in extra_users],
     }
 
     try:
-        result = yield client.post_json_get_json(uri, payload)
+        # We keep retrying the same request for timeouts. This is so that we
+        # have a good idea that the request has either succeeded or failed on
+        # the master, and so whether we should clean up or not.
+        while True:
+            try:
+                result = yield client.put_json(uri, payload)
+                break
+            except CodeMessageException as e:
+                if e.code != 504:
+                    raise
+
+            logger.warn("send_event request timed out")
+
+            # If we timed out we probably don't need to worry about backing
+            # off too much, but lets just wait a little anyway.
+            yield sleep(1)
     except MatrixCodeMessageException as e:
         # We convert to SynapseError as we know that it was a SynapseError
         # on the master process that we should send to the client. (And
@@ -66,7 +93,7 @@ class ReplicationSendEventRestServlet(RestServlet):
 
     The API looks like:
 
-        POST /_synapse/replication/send_event
+        POST /_synapse/replication/send_event/:event_id
 
         {
             "event": { .. serialized event .. },
@@ -74,9 +101,11 @@ class ReplicationSendEventRestServlet(RestServlet):
             "rejected_reason": ..,   // The event.rejected_reason field
             "context": { .. serialized event context .. },
             "requester": { .. serialized requester .. },
+            "ratelimit": true,
+            "extra_users": [],
         }
     """
-    PATTERNS = [re.compile("^/_synapse/replication/send_event$")]
+    PATTERNS = [re.compile("^/_synapse/replication/send_event/(?P<event_id>[^/]+)$")]
 
     def __init__(self, hs):
         super(ReplicationSendEventRestServlet, self).__init__()
@@ -85,8 +114,23 @@ class ReplicationSendEventRestServlet(RestServlet):
         self.store = hs.get_datastore()
         self.clock = hs.get_clock()
 
+        # The responses are tiny, so we may as well cache them for a while
+        self.response_cache = ResponseCache(hs, timeout_ms=30 * 60 * 1000)
+
+    def on_PUT(self, request, event_id):
+        result = self.response_cache.get(event_id)
+        if not result:
+            result = self.response_cache.set(
+                event_id,
+                self._handle_request(request)
+            )
+        else:
+            logger.warn("Returning cached response")
+        return make_deferred_yieldable(result)
+
+    @preserve_fn
     @defer.inlineCallbacks
-    def on_POST(self, request):
+    def _handle_request(self, request):
         with Measure(self.clock, "repl_send_event_parse"):
             content = parse_json_object_from_request(request)
 
@@ -98,6 +142,9 @@ class ReplicationSendEventRestServlet(RestServlet):
             requester = Requester.deserialize(self.store, content["requester"])
             context = yield EventContext.deserialize(self.store, content["context"])
 
+            ratelimit = content["ratelimit"]
+            extra_users = [UserID.from_string(u) for u in content["extra_users"]]
+
         if requester.user:
             request.authenticated_entity = requester.user.to_string()
 
@@ -106,8 +153,10 @@ class ReplicationSendEventRestServlet(RestServlet):
             event.event_id, event.room_id,
         )
 
-        yield self.event_creation_handler.handle_new_client_event(
+        yield self.event_creation_handler.persist_and_notify_client_event(
             requester, event, context,
+            ratelimit=ratelimit,
+            extra_users=extra_users,
         )
 
         defer.returnValue((200, {}))