Ensure only main or federation_sender process can send federation traffic

3 files changed, 21 insertions, 5 deletions

diff --git a/synapse/notifier.py b/synapse/notifier.py

index d528d1c1e0..054ca59ad2 100644
--- a/synapse/notifier.py
+++ b/synapse/notifier.py
@@ -143,7 +143,12 @@ class Notifier(object):
 
         self.clock = hs.get_clock()
         self.appservice_handler = hs.get_application_service_handler()
-        self.federation_sender = hs.get_federation_sender()
+
+        if hs.should_send_federation():
+            self.federation_sender = hs.get_federation_sender()
+        else:
+            self.federation_sender = None
+
         self.state_handler = hs.get_state_handler()
 
         self.clock.looping_call(
@@ -220,7 +225,9 @@ class Notifier(object):
         """Notify any user streams that are interested in this room event"""
         # poke any interested application service.
         self.appservice_handler.notify_interested_services(room_stream_id)
-        self.federation_sender.notify_new_events(room_stream_id)
+
+        if self.federation_sender:
+            self.federation_sender.notify_new_events(room_stream_id)
 
         if event.type == EventTypes.Member and event.membership == Membership.JOIN:
             self._user_joined_room(event.state_key, event.room_id)
diff --git a/synapse/replication/resource.py b/synapse/replication/resource.py

index cb9697e378..d79b421cba 100644
--- a/synapse/replication/resource.py
+++ b/synapse/replication/resource.py
@@ -460,7 +460,7 @@ class ReplicationResource(Resource):
             )
             upto_token = _position_from_rows(to_device_rows, current_position)
             writer.write_header_and_rows("to_device", to_device_rows, (
-                "position", "entity",
+                "position", "user_id", "device_id", "message_json"
             ), position=upto_token)
 
     @defer.inlineCallbacks
diff --git a/synapse/server.py b/synapse/server.py

index 6c57ab3e18..ef75ab434c 100644
--- a/synapse/server.py
+++ b/synapse/server.py
@@ -274,14 +274,23 @@ class HomeServer(object):
         return TransportLayerClient(self)
 
     def build_federation_sender(self):
-        if self.config.send_federation:
+        if self.should_send_federation():
             return TransactionQueue(self)
-        else:
+        elif not self.config.worker_app:
             return FederationRemoteSendQueue(self)
+        else:
+            raise Exception("Workers cannot send federation traffic")
 
     def remove_pusher(self, app_id, push_key, user_id):
         return self.get_pusherpool().remove_pusher(app_id, push_key, user_id)
 
+    def should_send_federation(self):
+        "Should this server be sending federation traffic directly?"
+        return self.config.send_federation and (
+            not self.config.worker_app
+            or self.config.worker_app == "synapse.app.federation_sender"
+        )
+
 
 def _make_dependency_method(depname):
     def _get(hs):