From 24dd73028ad7fc7a5109a5d97eef5a79179225d4 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Mon, 5 Feb 2018 17:22:16 +0000
Subject: Add replication http endpoint for event sending

---
 synapse/replication/http/send_event.py | 108 +++++++++++++++++++++++++++++++++
 1 file changed, 108 insertions(+)
 create mode 100644 synapse/replication/http/send_event.py

(limited to 'synapse/replication/http/send_event.py')

diff --git a/synapse/replication/http/send_event.py b/synapse/replication/http/send_event.py
new file mode 100644
index 0000000000..ff9b9d2f10
--- /dev/null
+++ b/synapse/replication/http/send_event.py
@@ -0,0 +1,108 @@
+# -*- coding: utf-8 -*-
+# Copyright 2018 New Vector Ltd
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from twisted.internet import defer
+
+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.metrics import Measure
+from synapse.types import Requester
+
+import logging
+import re
+
+logger = logging.getLogger(__name__)
+
+
+def send_event_to_master(client, host, port, requester, event, context):
+    """Send event to be handled on the master
+
+    Args:
+        client (SimpleHttpClient)
+        host (str): host of master
+        port (int): port on master listening for HTTP replication
+        requester (Requester)
+        event (FrozenEvent)
+        context (EventContext)
+    """
+    uri = "http://%s:%s/_synapse/replication/send_event" % (host, port,)
+
+    payload = {
+        "event": event.get_pdu_json(),
+        "internal_metadata": event.internal_metadata.get_dict(),
+        "rejected_reason": event.rejected_reason,
+        "context": context.serialize(),
+        "requester": requester.serialize(),
+    }
+
+    return client.post_json_get_json(uri, payload)
+
+
+class ReplicationSendEventRestServlet(RestServlet):
+    """Handles events newly created on workers, including persisting and
+    notifying.
+
+    The API looks like:
+
+        POST /_synapse/replication/send_event
+
+        {
+            "event": { .. serialized event .. },
+            "internal_metadata": { .. serialized internal_metadata .. },
+            "rejected_reason": ..,   // The event.rejected_reason field
+            "context": { .. serialized event context .. },
+            "requester": { .. serialized requester .. },
+        }
+    """
+    PATTERNS = [re.compile("^/_synapse/replication/send_event$")]
+
+    def __init__(self, hs):
+        super(ReplicationSendEventRestServlet, self).__init__()
+
+        self.event_creation_handler = hs.get_event_creation_handler()
+        self.store = hs.get_datastore()
+        self.clock = hs.get_clock()
+
+    @defer.inlineCallbacks
+    def on_POST(self, request):
+        with Measure(self.clock, "repl_send_event_parse"):
+            content = parse_json_object_from_request(request)
+
+            event_dict = content["event"]
+            internal_metadata = content["internal_metadata"]
+            rejected_reason = content["rejected_reason"]
+            event = FrozenEvent(event_dict, internal_metadata, rejected_reason)
+
+            requester = Requester.deserialize(self.store, content["requester"])
+            context = EventContext.deserialize(self.store, content["context"])
+
+        if requester.user:
+            request.authenticated_entity = requester.user.to_string()
+
+        logger.info(
+            "Got event to send with ID: %s into room: %s",
+            event.event_id, event.room_id,
+        )
+
+        yield self.event_creation_handler.handle_new_client_event(
+            requester, event, context,
+        )
+
+        defer.returnValue((200, {}))
+
+
+def register_servlets(hs, http_server):
+    ReplicationSendEventRestServlet(hs).register(http_server)
-- 
cgit 1.5.1


From ef344b10e529d1c0b023f009a8aae806e8a981bc Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 15 Feb 2018 11:02:43 +0000
Subject: Don't log errors propogated from send_event

---
 synapse/replication/http/send_event.py | 11 ++++++++++-
 1 file changed, 10 insertions(+), 1 deletion(-)

(limited to 'synapse/replication/http/send_event.py')

diff --git a/synapse/replication/http/send_event.py b/synapse/replication/http/send_event.py
index ff9b9d2f10..7b21a2213c 100644
--- a/synapse/replication/http/send_event.py
+++ b/synapse/replication/http/send_event.py
@@ -15,6 +15,7 @@
 
 from twisted.internet import defer
 
+from synapse.api.errors import SynapseError, MatrixCodeMessageException
 from synapse.events import FrozenEvent
 from synapse.events.snapshot import EventContext
 from synapse.http.servlet import RestServlet, parse_json_object_from_request
@@ -27,6 +28,7 @@ import re
 logger = logging.getLogger(__name__)
 
 
+@defer.inlineCallbacks
 def send_event_to_master(client, host, port, requester, event, context):
     """Send event to be handled on the master
 
@@ -48,7 +50,14 @@ def send_event_to_master(client, host, port, requester, event, context):
         "requester": requester.serialize(),
     }
 
-    return client.post_json_get_json(uri, payload)
+    try:
+        result = yield client.post_json_get_json(uri, payload)
+    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
+        # importantly, not stack traces everywhere)
+        raise SynapseError(e.code, e.msg, e.errcode)
+    defer.returnValue(result)
 
 
 class ReplicationSendEventRestServlet(RestServlet):
-- 
cgit 1.5.1


From 106906a65e647d94a9d2faf1b3a626bc1f608a25 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 15 Feb 2018 13:53:18 +0000
Subject: Don't serialize current state over replication

---
 synapse/events/snapshot.py             | 41 +++++++++++++++++++++++++++++-----
 synapse/replication/http/send_event.py |  4 ++--
 synapse/storage/state.py               | 14 ++++++++++++
 3 files changed, 51 insertions(+), 8 deletions(-)

(limited to 'synapse/replication/http/send_event.py')

diff --git a/synapse/events/snapshot.py b/synapse/events/snapshot.py
index 7b80444f73..f9445bef13 100644
--- a/synapse/events/snapshot.py
+++ b/synapse/events/snapshot.py
@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from twisted.internet import defer
 
 from frozendict import frozendict
 
@@ -77,16 +78,30 @@ class EventContext(object):
 
         self.app_service = None
 
-    def serialize(self):
+    def serialize(self, event):
         """Converts self to a type that can be serialized as JSON, and then
         deserialized by `deserialize`
 
+        Args:
+            event (FrozenEvent): The event that this context relates to
+
         Returns:
             dict
         """
+
+        # We don't serialize the full state dicts, instead they get pulled out
+        # of the DB on the other side. However, the other side can't figure out
+        # the prev_state_ids, so if we're a state event we include the event
+        # id that we replaced in the state.
+        if event.is_state():
+            prev_state_id = self.prev_state_ids.get((event.type, event.state_key))
+        else:
+            prev_state_id = None
+
         return {
-            "current_state_ids": _encode_state_dict(self.current_state_ids),
-            "prev_state_ids": _encode_state_dict(self.prev_state_ids),
+            "prev_state_id": prev_state_id,
+            "event_type": event.type,
+            "event_state_key": event.state_key if event.is_state() else None,
             "state_group": self.state_group,
             "rejected": self.rejected,
             "push_actions": self.push_actions,
@@ -97,6 +112,7 @@ class EventContext(object):
         }
 
     @staticmethod
+    @defer.inlineCallbacks
     def deserialize(store, input):
         """Converts a dict that was produced by `serialize` back into a
         EventContext.
@@ -109,8 +125,6 @@ class EventContext(object):
             EventContext
         """
         context = EventContext()
-        context.current_state_ids = _decode_state_dict(input["current_state_ids"])
-        context.prev_state_ids = _decode_state_dict(input["prev_state_ids"])
         context.state_group = input["state_group"]
         context.rejected = input["rejected"]
         context.push_actions = input["push_actions"]
@@ -118,11 +132,26 @@ class EventContext(object):
         context.delta_ids = _decode_state_dict(input["delta_ids"])
         context.prev_state_events = input["prev_state_events"]
 
+        # We use the state_group and prev_state_id stuff to pull the
+        # current_state_ids out of the DB and construct prev_state_ids.
+        prev_state_id = input["prev_state_id"]
+        event_type = input["event_type"]
+        event_state_key = input["event_state_key"]
+
+        context.current_state_ids = yield store.get_state_ids_for_group(
+            context.state_group,
+        )
+        if prev_state_id and event_state_key:
+            context.prev_state_ids = dict(context.current_state_ids)
+            context.prev_state_ids[(event_type, event_state_key)] = prev_state_id
+        else:
+            context.prev_state_ids = context.current_state_ids
+
         app_service_id = input["app_service_id"]
         if app_service_id:
             context.app_service = store.get_app_service_by_id(app_service_id)
 
-        return context
+        defer.returnValue(context)
 
 
 def _encode_state_dict(state_dict):
diff --git a/synapse/replication/http/send_event.py b/synapse/replication/http/send_event.py
index 7b21a2213c..468f4b68f4 100644
--- a/synapse/replication/http/send_event.py
+++ b/synapse/replication/http/send_event.py
@@ -46,7 +46,7 @@ def send_event_to_master(client, host, port, requester, event, context):
         "event": event.get_pdu_json(),
         "internal_metadata": event.internal_metadata.get_dict(),
         "rejected_reason": event.rejected_reason,
-        "context": context.serialize(),
+        "context": context.serialize(event),
         "requester": requester.serialize(),
     }
 
@@ -96,7 +96,7 @@ class ReplicationSendEventRestServlet(RestServlet):
             event = FrozenEvent(event_dict, internal_metadata, rejected_reason)
 
             requester = Requester.deserialize(self.store, content["requester"])
-            context = EventContext.deserialize(self.store, content["context"])
+            context = yield EventContext.deserialize(self.store, content["context"])
 
         if requester.user:
             request.authenticated_entity = requester.user.to_string()
diff --git a/synapse/storage/state.py b/synapse/storage/state.py
index d0a840456a..2b325e1c1f 100644
--- a/synapse/storage/state.py
+++ b/synapse/storage/state.py
@@ -139,6 +139,20 @@ class StateGroupWorkerStore(SQLBaseStore):
 
         defer.returnValue(group_to_state)
 
+    @defer.inlineCallbacks
+    def get_state_ids_for_group(self, state_group):
+        """Get the state IDs for the given state group
+
+        Args:
+            state_group (int)
+
+        Returns:
+            Deferred[dict]: Resolves to a map of (type, state_key) -> event_id
+        """
+        group_to_state = yield self._get_state_for_groups((state_group,))
+
+        defer.returnValue(group_to_state[state_group])
+
     @defer.inlineCallbacks
     def get_state_groups(self, room_id, event_ids):
         """ Get the state groups for the given list of event_ids
-- 
cgit 1.5.1


From 28e973ac119e0b4ec5b9e45772a572a94d0e6643 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 15 Feb 2018 16:30:10 +0000
Subject: Calculate push actions on worker

---
 synapse/app/event_creator.py           |  8 ++++
 synapse/handlers/message.py            | 86 ++++++++++++++++++++++++----------
 synapse/replication/http/send_event.py |  2 +-
 3 files changed, 70 insertions(+), 26 deletions(-)

(limited to 'synapse/replication/http/send_event.py')

diff --git a/synapse/app/event_creator.py b/synapse/app/event_creator.py
index b2ce399258..fc0b9e8c04 100644
--- a/synapse/app/event_creator.py
+++ b/synapse/app/event_creator.py
@@ -27,10 +27,14 @@ from synapse.http.server import JsonResource
 from synapse.http.site import SynapseSite
 from synapse.metrics.resource import METRICS_PREFIX, MetricsResource
 from synapse.replication.slave.storage._base import BaseSlavedStore
+from synapse.replication.slave.storage.account_data import SlavedAccountDataStore
 from synapse.replication.slave.storage.appservice import SlavedApplicationServiceStore
 from synapse.replication.slave.storage.client_ips import SlavedClientIpStore
 from synapse.replication.slave.storage.devices import SlavedDeviceStore
 from synapse.replication.slave.storage.events import SlavedEventStore
+from synapse.replication.slave.storage.push_rule import SlavedPushRuleStore
+from synapse.replication.slave.storage.pushers import SlavedPusherStore
+from synapse.replication.slave.storage.receipts import SlavedReceiptsStore
 from synapse.replication.slave.storage.registration import SlavedRegistrationStore
 from synapse.replication.slave.storage.room import RoomStore
 from synapse.replication.tcp.client import ReplicationClientHandler
@@ -48,6 +52,10 @@ logger = logging.getLogger("synapse.app.event_creator")
 
 
 class EventCreatorSlavedStore(
+    SlavedAccountDataStore,
+    SlavedPusherStore,
+    SlavedReceiptsStore,
+    SlavedPushRuleStore,
     SlavedDeviceStore,
     SlavedClientIpStore,
     SlavedApplicationServiceStore,
diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index d99d8049b3..4c186965a7 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -553,21 +553,67 @@ class EventCreationHandler(object):
         event,
         context,
         ratelimit=True,
-        extra_users=[]
+        extra_users=[],
     ):
-        # We now need to go and hit out to wherever we need to hit out to.
-
-        # If we're a worker we need to hit out to the master.
-        if self.config.worker_app:
-            yield send_event_to_master(
-                self.http_client,
-                host=self.config.worker_replication_host,
-                port=self.config.worker_replication_http_port,
-                requester=requester,
-                event=event,
-                context=context,
+        """Processes a new event. This includes checking auth, persisting it,
+        notifying users, sending to remote servers, etc.
+
+        If called from a worker will hit out to the master process for final
+        processing.
+
+        Args:
+            requester (Requester)
+            event (FrozenEvent)
+            context (EventContext)
+            ratelimit (bool)
+            extra_users (list(str)): Any extra users to notify about event
+        """
+
+        yield self.action_generator.handle_push_actions_for_event(
+            event, context
+        )
+
+        try:
+            # We now need to go and hit out to wherever we need to hit out to.
+
+            # If we're a worker we need to hit out to the master.
+            if self.config.worker_app:
+                yield send_event_to_master(
+                    self.http_client,
+                    host=self.config.worker_replication_host,
+                    port=self.config.worker_replication_http_port,
+                    requester=requester,
+                    event=event,
+                    context=context,
+                )
+                return
+
+            yield self.persist_and_notify_client_event(
+                requester,
+                event,
+                context,
+                ratelimit=ratelimit,
+                extra_users=extra_users,
             )
-            return
+        except:  # noqa: E722, as we reraise the exception this is fine.
+            # Ensure that we actually remove the entries in the push actions
+            # staging area, if we calculated them.
+            preserve_fn(self.store.remove_push_actions_from_staging)(event.event_id)
+            raise
+
+    @defer.inlineCallbacks
+    def persist_and_notify_client_event(
+        self,
+        requester,
+        event,
+        context,
+        ratelimit=True,
+        extra_users=[],
+    ):
+        """Called when we have fully built and authed the event. This should
+        only be run on master.
+        """
+        assert not self.config.worker_app
 
         if ratelimit:
             yield self.base_handler.ratelimit(requester)
@@ -679,20 +725,10 @@ class EventCreationHandler(object):
                 "Changing the room create event is forbidden",
             )
 
-        yield self.action_generator.handle_push_actions_for_event(
-            event, context
+        (event_stream_id, max_stream_id) = yield self.store.persist_event(
+            event, context=context
         )
 
-        try:
-            (event_stream_id, max_stream_id) = yield self.store.persist_event(
-                event, context=context
-            )
-        except:  # noqa: E722, as we reraise the exception this is fine.
-            # Ensure that we actually remove the entries in the push actions
-            # staging area
-            preserve_fn(self.store.remove_push_actions_from_staging)(event.event_id)
-            raise
-
         # this intentionally does not yield: we don't care about the result
         # and don't need to wait for it.
         preserve_fn(self.pusher_pool.on_new_notifications)(
diff --git a/synapse/replication/http/send_event.py b/synapse/replication/http/send_event.py
index 468f4b68f4..3a99a88bc5 100644
--- a/synapse/replication/http/send_event.py
+++ b/synapse/replication/http/send_event.py
@@ -106,7 +106,7 @@ 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,
         )
 
-- 
cgit 1.5.1


From 6b8604239f4c9463023e59664e7810ba58b8f428 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 1 Mar 2018 10:08:28 +0000
Subject: Correctly send ratelimit and extra_users params

---
 synapse/handlers/message.py            |  2 ++
 synapse/replication/http/send_event.py | 14 +++++++++++++-
 2 files changed, 15 insertions(+), 1 deletion(-)

(limited to 'synapse/replication/http/send_event.py')

diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index c4151d73eb..5f88f84d38 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -583,6 +583,8 @@ class EventCreationHandler(object):
                     requester=requester,
                     event=event,
                     context=context,
+                    ratelimit=ratelimit,
+                    extra_users=extra_users,
                 )
                 return
 
diff --git a/synapse/replication/http/send_event.py b/synapse/replication/http/send_event.py
index 3a99a88bc5..439bfbb4f6 100644
--- a/synapse/replication/http/send_event.py
+++ b/synapse/replication/http/send_event.py
@@ -29,7 +29,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,6 +40,8 @@ def send_event_to_master(client, host, port, requester, event, context):
         requester (Requester)
         event (FrozenEvent)
         context (EventContext)
+        ratelimit (bool)
+        extra_users (list(str)): Any extra users to notify about event
     """
     uri = "http://%s:%s/_synapse/replication/send_event" % (host, port,)
 
@@ -48,6 +51,8 @@ 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": extra_users,
     }
 
     try:
@@ -74,6 +79,8 @@ 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$")]
@@ -98,6 +105,9 @@ class ReplicationSendEventRestServlet(RestServlet):
             requester = Requester.deserialize(self.store, content["requester"])
             context = yield EventContext.deserialize(self.store, content["context"])
 
+            ratelimit = content["ratelimit"]
+            extra_users = content["extra_users"]
+
         if requester.user:
             request.authenticated_entity = requester.user.to_string()
 
@@ -108,6 +118,8 @@ class ReplicationSendEventRestServlet(RestServlet):
 
         yield self.event_creation_handler.persist_and_notify_client_event(
             requester, event, context,
+            ratelimit=ratelimit,
+            extra_users=extra_users,
         )
 
         defer.returnValue((200, {}))
-- 
cgit 1.5.1


From 8ded8ba2c755f254bd98e15db7bc865eed997f07 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 1 Mar 2018 11:20:34 +0000
Subject: Make repl send_event idempotent and retry on timeouts

If we treated timeouts as failures on the worker we would attempt to
clean up e.g. push actions while the master might still process the
event.
---
 synapse/replication/http/send_event.py | 44 +++++++++++++++++++++++++++++-----
 1 file changed, 38 insertions(+), 6 deletions(-)

(limited to 'synapse/replication/http/send_event.py')

diff --git a/synapse/replication/http/send_event.py b/synapse/replication/http/send_event.py
index 439bfbb4f6..73cd3d91d9 100644
--- a/synapse/replication/http/send_event.py
+++ b/synapse/replication/http/send_event.py
@@ -15,10 +15,15 @@
 
 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
 
@@ -43,7 +48,9 @@ def send_event_to_master(client, host, port, requester, event, context,
         ratelimit (bool)
         extra_users (list(str)): 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(),
@@ -56,7 +63,20 @@ def send_event_to_master(client, host, port, requester, event, context,
     }
 
     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
+
+            # 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
@@ -71,7 +91,7 @@ class ReplicationSendEventRestServlet(RestServlet):
 
     The API looks like:
 
-        POST /_synapse/replication/send_event
+        POST /_synapse/replication/send_event/:event_id
 
         {
             "event": { .. serialized event .. },
@@ -83,7 +103,7 @@ class ReplicationSendEventRestServlet(RestServlet):
             "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__()
@@ -92,8 +112,20 @@ 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,
+                preserve_fn(self._handle_request)(request)
+            )
+        return make_deferred_yieldable(result)
+
     @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)
 
-- 
cgit 1.5.1


From 89f90d808ad923374a204377b0bc8ed15a9385a9 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 1 Mar 2018 11:59:16 +0000
Subject: Add some logging

---
 synapse/replication/http/send_event.py | 2 ++
 1 file changed, 2 insertions(+)

(limited to 'synapse/replication/http/send_event.py')

diff --git a/synapse/replication/http/send_event.py b/synapse/replication/http/send_event.py
index 73cd3d91d9..f490622b63 100644
--- a/synapse/replication/http/send_event.py
+++ b/synapse/replication/http/send_event.py
@@ -72,6 +72,7 @@ def send_event_to_master(client, host, port, requester, event, context,
                 break
             except CodeMessageException as e:
                 if e.code != 504:
+                    logger.warn("send_event request timed out")
                     raise
 
             # If we timed out we probably don't need to worry about backing
@@ -118,6 +119,7 @@ class ReplicationSendEventRestServlet(RestServlet):
     def on_PUT(self, request, event_id):
         result = self.response_cache.get(event_id)
         if not result:
+            logger.warn("Returning cached response")
             result = self.response_cache.set(
                 event_id,
                 preserve_fn(self._handle_request)(request)
-- 
cgit 1.5.1


From 157298f9862bd06e9268aed9cdffbc1c912fd74d Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 1 Mar 2018 11:59:45 +0000
Subject: Don't do preserve_fn for every request

---
 synapse/replication/http/send_event.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

(limited to 'synapse/replication/http/send_event.py')

diff --git a/synapse/replication/http/send_event.py b/synapse/replication/http/send_event.py
index f490622b63..665a56d2e6 100644
--- a/synapse/replication/http/send_event.py
+++ b/synapse/replication/http/send_event.py
@@ -122,10 +122,11 @@ class ReplicationSendEventRestServlet(RestServlet):
             logger.warn("Returning cached response")
             result = self.response_cache.set(
                 event_id,
-                preserve_fn(self._handle_request)(request)
+                self._handle_request(request)
             )
         return make_deferred_yieldable(result)
 
+    @preserve_fn
     @defer.inlineCallbacks
     def _handle_request(self, request):
         with Measure(self.clock, "repl_send_event_parse"):
-- 
cgit 1.5.1


From 126b9bf96f2c989e7402b5e0177ee39a2a20940e Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 1 Mar 2018 12:05:33 +0000
Subject: Log in the correct places

---
 synapse/replication/http/send_event.py | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

(limited to 'synapse/replication/http/send_event.py')

diff --git a/synapse/replication/http/send_event.py b/synapse/replication/http/send_event.py
index 665a56d2e6..70f2fe456a 100644
--- a/synapse/replication/http/send_event.py
+++ b/synapse/replication/http/send_event.py
@@ -72,9 +72,10 @@ def send_event_to_master(client, host, port, requester, event, context,
                 break
             except CodeMessageException as e:
                 if e.code != 504:
-                    logger.warn("send_event request timed out")
                     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)
@@ -119,11 +120,12 @@ class ReplicationSendEventRestServlet(RestServlet):
     def on_PUT(self, request, event_id):
         result = self.response_cache.get(event_id)
         if not result:
-            logger.warn("Returning cached response")
             result = self.response_cache.set(
                 event_id,
                 self._handle_request(request)
             )
+        else:
+            logger.warn("Returning cached response")
         return make_deferred_yieldable(result)
 
     @preserve_fn
-- 
cgit 1.5.1


From d0fcc48f9dfc09531619faf23d407807eec46df9 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 1 Mar 2018 17:39:58 +0000
Subject: extra_users is actually a list of UserIDs

---
 synapse/handlers/message.py            | 2 +-
 synapse/replication/http/send_event.py | 8 ++++----
 2 files changed, 5 insertions(+), 5 deletions(-)

(limited to 'synapse/replication/http/send_event.py')

diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index 42aab91c50..4f97c8db79 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -667,7 +667,7 @@ class EventCreationHandler(object):
             event (FrozenEvent)
             context (EventContext)
             ratelimit (bool)
-            extra_users (list(str)): Any extra users to notify about event
+            extra_users (list(UserID)): Any extra users to notify about event
         """
 
         try:
diff --git a/synapse/replication/http/send_event.py b/synapse/replication/http/send_event.py
index 70f2fe456a..bbe2f967b7 100644
--- a/synapse/replication/http/send_event.py
+++ b/synapse/replication/http/send_event.py
@@ -25,7 +25,7 @@ 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
@@ -46,7 +46,7 @@ def send_event_to_master(client, host, port, requester, event, context,
         event (FrozenEvent)
         context (EventContext)
         ratelimit (bool)
-        extra_users (list(str)): Any extra users to notify about event
+        extra_users (list(UserID)): Any extra users to notify about event
     """
     uri = "http://%s:%s/_synapse/replication/send_event/%s" % (
         host, port, event.event_id,
@@ -59,7 +59,7 @@ def send_event_to_master(client, host, port, requester, event, context,
         "context": context.serialize(event),
         "requester": requester.serialize(),
         "ratelimit": ratelimit,
-        "extra_users": extra_users,
+        "extra_users": [u.to_string() for u in extra_users],
     }
 
     try:
@@ -143,7 +143,7 @@ class ReplicationSendEventRestServlet(RestServlet):
             context = yield EventContext.deserialize(self.store, content["context"])
 
             ratelimit = content["ratelimit"]
-            extra_users = content["extra_users"]
+            extra_users = [UserID.from_string(u) for u in content["extra_users"]]
 
         if requester.user:
             request.authenticated_entity = requester.user.to_string()
-- 
cgit 1.5.1