diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py
new file mode 100644
index 0000000000..0f5b974291
--- /dev/null
+++ b/synapse/federation/replication.py
@@ -0,0 +1,582 @@
+# -*- coding: utf-8 -*-
+# Copyright 2014 matrix.org
+#
+# 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.
+"""This layer is responsible for replicating with remote home servers using
+a given transport.
+"""
+
+from twisted.internet import defer
+
+from .units import Transaction, Pdu, Edu
+
+from .persistence import PduActions, TransactionActions
+
+from synapse.util.logutils import log_function
+
+import logging
+
+
+logger = logging.getLogger(__name__)
+
+
+class ReplicationLayer(object):
+ """This layer is responsible for replicating with remote home servers over
+ the given transport. I.e., does the sending and receiving of PDUs to
+ remote home servers.
+
+ The layer communicates with the rest of the server via a registered
+ ReplicationHandler.
+
+ In more detail, the layer:
+ * Receives incoming data and processes it into transactions and pdus.
+ * Fetches any PDUs it thinks it might have missed.
+ * Keeps the current state for contexts up to date by applying the
+ suitable conflict resolution.
+ * Sends outgoing pdus wrapped in transactions.
+ * Fills out the references to previous pdus/transactions appropriately
+ for outgoing data.
+ """
+
+ def __init__(self, hs, transport_layer):
+ self.server_name = hs.hostname
+
+ self.transport_layer = transport_layer
+ self.transport_layer.register_received_handler(self)
+ self.transport_layer.register_request_handler(self)
+
+ self.store = hs.get_datastore()
+ self.pdu_actions = PduActions(self.store)
+ self.transaction_actions = TransactionActions(self.store)
+
+ self._transaction_queue = _TransactionQueue(
+ hs, self.transaction_actions, transport_layer
+ )
+
+ self.handler = None
+ self.edu_handlers = {}
+
+ self._order = 0
+
+ self._clock = hs.get_clock()
+
+ def set_handler(self, handler):
+ """Sets the handler that the replication layer will use to communicate
+ receipt of new PDUs from other home servers. The required methods are
+ documented on :py:class:`.ReplicationHandler`.
+ """
+ self.handler = handler
+
+ def register_edu_handler(self, edu_type, handler):
+ if edu_type in self.edu_handlers:
+ raise KeyError("Already have an EDU handler for %s" % (edu_type))
+
+ self.edu_handlers[edu_type] = handler
+
+ @defer.inlineCallbacks
+ @log_function
+ def send_pdu(self, pdu):
+ """Informs the replication layer about a new PDU generated within the
+ home server that should be transmitted to others.
+
+ This will fill out various attributes on the PDU object, e.g. the
+ `prev_pdus` key.
+
+ *Note:* The home server should always call `send_pdu` even if it knows
+ that it does not need to be replicated to other home servers. This is
+ in case e.g. someone else joins via a remote home server and then
+ paginates.
+
+ TODO: Figure out when we should actually resolve the deferred.
+
+ Args:
+ pdu (Pdu): The new Pdu.
+
+ Returns:
+ Deferred: Completes when we have successfully processed the PDU
+ and replicated it to any interested remote home servers.
+ """
+ order = self._order
+ self._order += 1
+
+ logger.debug("[%s] Persisting PDU", pdu.pdu_id)
+
+ #yield self.pdu_actions.populate_previous_pdus(pdu)
+
+ # Save *before* trying to send
+ yield self.pdu_actions.persist_outgoing(pdu)
+
+ logger.debug("[%s] Persisted PDU", pdu.pdu_id)
+ logger.debug("[%s] transaction_layer.enqueue_pdu... ", pdu.pdu_id)
+
+ # TODO, add errback, etc.
+ self._transaction_queue.enqueue_pdu(pdu, order)
+
+ logger.debug("[%s] transaction_layer.enqueue_pdu... done", pdu.pdu_id)
+
+ @log_function
+ def send_edu(self, destination, edu_type, content):
+ edu = Edu(
+ origin=self.server_name,
+ destination=destination,
+ edu_type=edu_type,
+ content=content,
+ )
+
+ # TODO, add errback, etc.
+ self._transaction_queue.enqueue_edu(edu)
+
+ @defer.inlineCallbacks
+ @log_function
+ def paginate(self, dest, context, limit):
+ """Requests some more historic PDUs for the given context from the
+ given destination server.
+
+ Args:
+ dest (str): The remote home server to ask.
+ context (str): The context to paginate back on.
+ limit (int): The maximum number of PDUs to return.
+
+ Returns:
+ Deferred: Results in the received PDUs.
+ """
+ extremities = yield self.store.get_oldest_pdus_in_context(context)
+
+ logger.debug("paginate extrem=%s", extremities)
+
+ # If there are no extremeties then we've (probably) reached the start.
+ if not extremities:
+ return
+
+ transaction_data = yield self.transport_layer.paginate(
+ dest, context, extremities, limit)
+
+ logger.debug("paginate transaction_data=%s", repr(transaction_data))
+
+ transaction = Transaction(**transaction_data)
+
+ pdus = [Pdu(outlier=False, **p) for p in transaction.pdus]
+ for pdu in pdus:
+ yield self._handle_new_pdu(pdu)
+
+ defer.returnValue(pdus)
+
+ @defer.inlineCallbacks
+ @log_function
+ def get_pdu(self, destination, pdu_origin, pdu_id, outlier=False):
+ """Requests the PDU with given origin and ID from the remote home
+ server.
+
+ This will persist the PDU locally upon receipt.
+
+ Args:
+ destination (str): Which home server to query
+ pdu_origin (str): The home server that originally sent the pdu.
+ pdu_id (str)
+ outlier (bool): Indicates whether the PDU is an `outlier`, i.e. if
+ it's from an arbitary point in the context as opposed to part
+ of the current block of PDUs. Defaults to `False`
+
+ Returns:
+ Deferred: Results in the requested PDU.
+ """
+
+ transaction_data = yield self.transport_layer.get_pdu(
+ destination, pdu_origin, pdu_id)
+
+ transaction = Transaction(**transaction_data)
+
+ pdu_list = [Pdu(outlier=outlier, **p) for p in transaction.pdus]
+
+ pdu = None
+ if pdu_list:
+ pdu = pdu_list[0]
+ yield self._handle_new_pdu(pdu)
+
+ defer.returnValue(pdu)
+
+ @defer.inlineCallbacks
+ @log_function
+ def get_state_for_context(self, destination, context):
+ """Requests all of the `current` state PDUs for a given context from
+ a remote home server.
+
+ Args:
+ destination (str): The remote homeserver to query for the state.
+ context (str): The context we're interested in.
+
+ Returns:
+ Deferred: Results in a list of PDUs.
+ """
+
+ transaction_data = yield self.transport_layer.get_context_state(
+ destination, context)
+
+ transaction = Transaction(**transaction_data)
+
+ pdus = [Pdu(outlier=True, **p) for p in transaction.pdus]
+ for pdu in pdus:
+ yield self._handle_new_pdu(pdu)
+
+ defer.returnValue(pdus)
+
+ @defer.inlineCallbacks
+ @log_function
+ def on_context_pdus_request(self, context):
+ pdus = yield self.pdu_actions.get_all_pdus_from_context(
+ context
+ )
+ defer.returnValue((200, self._transaction_from_pdus(pdus).get_dict()))
+
+ @defer.inlineCallbacks
+ @log_function
+ def on_paginate_request(self, context, versions, limit):
+
+ pdus = yield self.pdu_actions.paginate(context, versions, limit)
+
+ defer.returnValue((200, self._transaction_from_pdus(pdus).get_dict()))
+
+ @defer.inlineCallbacks
+ @log_function
+ def on_incoming_transaction(self, transaction_data):
+ transaction = Transaction(**transaction_data)
+
+ logger.debug("[%s] Got transaction", transaction.transaction_id)
+
+ response = yield self.transaction_actions.have_responded(transaction)
+
+ if response:
+ logger.debug("[%s] We've already responed to this request",
+ transaction.transaction_id)
+ defer.returnValue(response)
+ return
+
+ logger.debug("[%s] Transacition is new", transaction.transaction_id)
+
+ pdu_list = [Pdu(**p) for p in transaction.pdus]
+
+ dl = []
+ for pdu in pdu_list:
+ dl.append(self._handle_new_pdu(pdu))
+
+ if hasattr(transaction, "edus"):
+ for edu in [Edu(**x) for x in transaction.edus]:
+ self.received_edu(edu.origin, edu.edu_type, edu.content)
+
+ results = yield defer.DeferredList(dl)
+
+ ret = []
+ for r in results:
+ if r[0]:
+ ret.append({})
+ else:
+ logger.exception(r[1])
+ ret.append({"error": str(r[1])})
+
+ logger.debug("Returning: %s", str(ret))
+
+ yield self.transaction_actions.set_response(
+ transaction,
+ 200, response
+ )
+ defer.returnValue((200, response))
+
+ def received_edu(self, origin, edu_type, content):
+ if edu_type in self.edu_handlers:
+ self.edu_handlers[edu_type](origin, content)
+ else:
+ logger.warn("Received EDU of type %s with no handler", edu_type)
+
+ @defer.inlineCallbacks
+ @log_function
+ def on_context_state_request(self, context):
+ results = yield self.store.get_current_state_for_context(
+ context
+ )
+
+ logger.debug("Context returning %d results", len(results))
+
+ pdus = [Pdu.from_pdu_tuple(p) for p in results]
+ defer.returnValue((200, self._transaction_from_pdus(pdus).get_dict()))
+
+ @defer.inlineCallbacks
+ @log_function
+ def on_pdu_request(self, pdu_origin, pdu_id):
+ pdu = yield self._get_persisted_pdu(pdu_id, pdu_origin)
+
+ if pdu:
+ defer.returnValue(
+ (200, self._transaction_from_pdus([pdu]).get_dict())
+ )
+ else:
+ defer.returnValue((404, ""))
+
+ @defer.inlineCallbacks
+ @log_function
+ def on_pull_request(self, origin, versions):
+ transaction_id = max([int(v) for v in versions])
+
+ response = yield self.pdu_actions.after_transaction(
+ transaction_id,
+ origin,
+ self.server_name
+ )
+
+ if not response:
+ response = []
+
+ defer.returnValue(
+ (200, self._transaction_from_pdus(response).get_dict())
+ )
+
+ @defer.inlineCallbacks
+ @log_function
+ def _get_persisted_pdu(self, pdu_id, pdu_origin):
+ """ Get a PDU from the database with given origin and id.
+
+ Returns:
+ Deferred: Results in a `Pdu`.
+ """
+ pdu_tuple = yield self.store.get_pdu(pdu_id, pdu_origin)
+
+ defer.returnValue(Pdu.from_pdu_tuple(pdu_tuple))
+
+ def _transaction_from_pdus(self, pdu_list):
+ """Returns a new Transaction containing the given PDUs suitable for
+ transmission.
+ """
+ return Transaction(
+ pdus=[p.get_dict() for p in pdu_list],
+ origin=self.server_name,
+ ts=int(self._clock.time_msec()),
+ destination=None,
+ )
+
+ @defer.inlineCallbacks
+ @log_function
+ def _handle_new_pdu(self, pdu):
+ # We reprocess pdus when we have seen them only as outliers
+ existing = yield self._get_persisted_pdu(pdu.pdu_id, pdu.origin)
+
+ if existing and (not existing.outlier or pdu.outlier):
+ logger.debug("Already seen pdu %s %s", pdu.pdu_id, pdu.origin)
+ defer.returnValue({})
+ return
+
+ # Get missing pdus if necessary.
+ is_new = yield self.pdu_actions.is_new(pdu)
+ if is_new and not pdu.outlier:
+ # We only paginate backwards to the min depth.
+ min_depth = yield self.store.get_min_depth_for_context(pdu.context)
+
+ if min_depth and pdu.depth > min_depth:
+ for pdu_id, origin in pdu.prev_pdus:
+ exists = yield self._get_persisted_pdu(pdu_id, origin)
+
+ if not exists:
+ logger.debug("Requesting pdu %s %s", pdu_id, origin)
+
+ try:
+ yield self.get_pdu(
+ pdu.origin,
+ pdu_id=pdu_id,
+ pdu_origin=origin
+ )
+ logger.debug("Processed pdu %s %s", pdu_id, origin)
+ except:
+ # TODO(erikj): Do some more intelligent retries.
+ logger.exception("Failed to get PDU")
+
+ # Persist the Pdu, but don't mark it as processed yet.
+ yield self.pdu_actions.persist_received(pdu)
+
+ ret = yield self.handler.on_receive_pdu(pdu)
+
+ yield self.pdu_actions.mark_as_processed(pdu)
+
+ defer.returnValue(ret)
+
+ def __str__(self):
+ return "<ReplicationLayer(%s)>" % self.server_name
+
+
+class ReplicationHandler(object):
+ """This defines the methods that the :py:class:`.ReplicationLayer` will
+ use to communicate with the rest of the home server.
+ """
+ def on_receive_pdu(self, pdu):
+ raise NotImplementedError("on_receive_pdu")
+
+
+class _TransactionQueue(object):
+ """This class makes sure we only have one transaction in flight at
+ a time for a given destination.
+
+ It batches pending PDUs into single transactions.
+ """
+
+ def __init__(self, hs, transaction_actions, transport_layer):
+
+ self.server_name = hs.hostname
+ self.transaction_actions = transaction_actions
+ self.transport_layer = transport_layer
+
+ self._clock = hs.get_clock()
+
+ # Is a mapping from destinations -> deferreds. Used to keep track
+ # of which destinations have transactions in flight and when they are
+ # done
+ self.pending_transactions = {}
+
+ # Is a mapping from destination -> list of
+ # tuple(pending pdus, deferred, order)
+ self.pending_pdus_by_dest = {}
+ # destination -> list of tuple(edu, deferred)
+ self.pending_edus_by_dest = {}
+
+ # HACK to get unique tx id
+ self._next_txn_id = int(self._clock.time_msec())
+
+ @defer.inlineCallbacks
+ @log_function
+ def enqueue_pdu(self, pdu, order):
+ # We loop through all destinations to see whether we already have
+ # a transaction in progress. If we do, stick it in the pending_pdus
+ # table and we'll get back to it later.
+
+ destinations = [
+ d for d in pdu.destinations
+ if d != self.server_name
+ ]
+
+ logger.debug("Sending to: %s", str(destinations))
+
+ if not destinations:
+ return
+
+ deferreds = []
+
+ for destination in destinations:
+ deferred = defer.Deferred()
+ self.pending_pdus_by_dest.setdefault(destination, []).append(
+ (pdu, deferred, order)
+ )
+
+ self._attempt_new_transaction(destination)
+
+ deferreds.append(deferred)
+
+ yield defer.DeferredList(deferreds)
+
+ # NO inlineCallbacks
+ def enqueue_edu(self, edu):
+ destination = edu.destination
+
+ deferred = defer.Deferred()
+ self.pending_edus_by_dest.setdefault(destination, []).append(
+ (edu, deferred)
+ )
+
+ def eb(failure):
+ deferred.errback(failure)
+ self._attempt_new_transaction(destination).addErrback(eb)
+
+ return deferred
+
+ @defer.inlineCallbacks
+ @log_function
+ def _attempt_new_transaction(self, destination):
+ if destination in self.pending_transactions:
+ return
+
+ # list of (pending_pdu, deferred, order)
+ pending_pdus = self.pending_pdus_by_dest.pop(destination, [])
+ pending_edus = self.pending_edus_by_dest.pop(destination, [])
+
+ if not pending_pdus and not pending_edus:
+ return
+
+ logger.debug("TX [%s] Attempting new transaction", destination)
+
+ # Sort based on the order field
+ pending_pdus.sort(key=lambda t: t[2])
+
+ pdus = [x[0] for x in pending_pdus]
+ edus = [x[0] for x in pending_edus]
+ deferreds = [x[1] for x in pending_pdus + pending_edus]
+
+ try:
+ self.pending_transactions[destination] = 1
+
+ logger.debug("TX [%s] Persisting transaction...", destination)
+
+ transaction = Transaction.create_new(
+ ts=self._clock.time_msec(),
+ transaction_id=self._next_txn_id,
+ origin=self.server_name,
+ destination=destination,
+ pdus=pdus,
+ edus=edus,
+ )
+
+ self._next_txn_id += 1
+
+ yield self.transaction_actions.prepare_to_send(transaction)
+
+ logger.debug("TX [%s] Persisted transaction", destination)
+ logger.debug("TX [%s] Sending transaction...", destination)
+
+ # Actually send the transaction
+ code, response = yield self.transport_layer.send_transaction(
+ transaction
+ )
+
+ logger.debug("TX [%s] Sent transaction", destination)
+ logger.debug("TX [%s] Marking as delivered...", destination)
+
+ yield self.transaction_actions.delivered(
+ transaction, code, response
+ )
+
+ logger.debug("TX [%s] Marked as delivered", destination)
+ logger.debug("TX [%s] Yielding to callbacks...", destination)
+
+ for deferred in deferreds:
+ if code == 200:
+ deferred.callback(None)
+ else:
+ deferred.errback(RuntimeError("Got status %d" % code))
+
+ # Ensures we don't continue until all callbacks on that
+ # deferred have fired
+ yield deferred
+
+ logger.debug("TX [%s] Yielded to callbacks", destination)
+
+ except Exception as e:
+ logger.error("TX Problem in _attempt_transaction")
+
+ # We capture this here as there as nothing actually listens
+ # for this finishing functions deferred.
+ logger.exception(e)
+
+ for deferred in deferreds:
+ deferred.errback(e)
+ yield deferred
+
+ finally:
+ # We want to be *very* sure we delete this after we stop processing
+ self.pending_transactions.pop(destination, None)
+
+ # Check to see if there is anything else to send.
+ self._attempt_new_transaction(destination)
|
