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..7d28c2745c 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -553,24 +553,21 @@ 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,
- )
- return
+ """Processes a new event. This includes checking auth, persisting it,
+ notifying users, sending to remote servers, etc.
- if ratelimit:
- yield self.base_handler.ratelimit(requester)
+ 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
+ """
try:
yield self.auth.check_from_context(event, context)
@@ -586,6 +583,57 @@ class EventCreationHandler(object):
logger.exception("Failed to encode content: %r", event.content)
raise
+ yield self.action_generator.handle_push_actions_for_event(
+ event, context
+ )
+
+ try:
+ # 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,
+ ratelimit=ratelimit,
+ extra_users=extra_users,
+ )
+ return
+
+ yield self.persist_and_notify_client_event(
+ requester,
+ event,
+ context,
+ ratelimit=ratelimit,
+ extra_users=extra_users,
+ )
+ 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 the event, have already
+ calculated the push actions for the event, and checked auth.
+
+ This should only be run on master.
+ """
+ assert not self.config.worker_app
+
+ if ratelimit:
+ yield self.base_handler.ratelimit(requester)
+
yield self.base_handler.maybe_kick_guest_users(event, context)
if event.type == EventTypes.CanonicalAlias:
@@ -679,20 +727,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..70f2fe456a 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
@@ -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(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(),
@@ -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": 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 = 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, {}))
|
