diff options
| -rw-r--r-- | synapse/federation/replication.py | 27 | ||||
| -rw-r--r-- | synapse/handlers/federation.py | 55 | ||||
| -rw-r--r-- | synapse/util/distributor.py | 2 | ||||
| -rw-r--r-- | tests/federation/test_federation.py | 2 |
4 files changed, 55 insertions, 31 deletions
diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py index ec9b6e246b..8abf67b1b5 100644 --- a/synapse/federation/replication.py +++ b/synapse/federation/replication.py @@ -256,31 +256,35 @@ class ReplicationLayer(object): @defer.inlineCallbacks @log_function - def get_state_for_context(self, destination, context, event_id=None): + def get_state_for_context(self, destination, context, event_id): """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. + event_id (str): The id of the event we want the state at. Returns: Deferred: Results in a list of PDUs. """ - transaction_data = yield self.transport_layer.get_context_state( + result = yield self.transport_layer.get_context_state( destination, context, event_id=event_id, ) - transaction = Transaction(**transaction_data) pdus = [ + self.event_from_pdu_json(p, outlier=True) for p in result["pdus"] + ] + + auth_chain = [ self.event_from_pdu_json(p, outlier=True) - for p in transaction.pdus + for p in result.get("auth_chain", []) ] - defer.returnValue(pdus) + defer.returnValue((pdus, auth_chain)) @defer.inlineCallbacks @log_function @@ -383,10 +387,16 @@ class ReplicationLayer(object): context, event_id, ) + auth_chain = yield self.store.get_auth_chain( + [pdu.event_id for pdu in pdus] + ) else: raise NotImplementedError("Specify an event") - defer.returnValue((200, self._transaction_from_pdus(pdus).get_dict())) + defer.returnValue((200, { + "pdus": [pdu.get_pdu_json() for pdu in pdus], + "auth_chain": [pdu.get_pdu_json() for pdu in auth_chain], + })) @defer.inlineCallbacks @log_function @@ -573,6 +583,8 @@ class ReplicationLayer(object): state = None + auth_chain = [] + # We need to make sure we have all the auth events. # for e_id, _ in pdu.auth_events: # exists = yield self._get_persisted_pdu( @@ -645,7 +657,7 @@ class ReplicationLayer(object): "_handle_new_pdu getting state for %s", pdu.room_id ) - state = yield self.get_state_for_context( + state, auth_chain = yield self.get_state_for_context( origin, pdu.room_id, pdu.event_id, ) @@ -655,6 +667,7 @@ class ReplicationLayer(object): pdu, backfilled=backfilled, state=state, + auth_chain=auth_chain, ) else: ret = None diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index 2f6036145c..e23c5c2195 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -95,7 +95,8 @@ class FederationHandler(BaseHandler): @log_function @defer.inlineCallbacks - def on_receive_pdu(self, origin, pdu, backfilled, state=None): + def on_receive_pdu(self, origin, pdu, backfilled, state=None, + auth_chain=None): """ Called by the ReplicationLayer when we have a new pdu. We need to do auth checks and put it through the StateHandler. """ @@ -150,35 +151,35 @@ class FederationHandler(BaseHandler): if not is_in_room and not event.internal_metadata.outlier: logger.debug("Got event for room we're not in.") - replication_layer = self.replication_layer - auth_chain = yield replication_layer.get_event_auth( - origin, - context=event.room_id, - event_id=event.event_id, - ) + replication = self.replication_layer + + if not state: + state, auth_chain = yield replication.get_state_for_context( + origin, context=event.room_id, event_id=event.event_id, + ) + + if not auth_chain: + auth_chain = yield replication.get_event_auth( + origin, + context=event.room_id, + event_id=event.event_id, + ) for e in auth_chain: e.internal_metadata.outlier = True try: - yield self._handle_new_event(e, fetch_missing=False) + yield self._handle_new_event(e, fetch_auth_from=origin) except: logger.exception( "Failed to handle auth event %s", e.event_id, ) - if not state: - state = yield replication_layer.get_state_for_context( - origin, - context=event.room_id, - event_id=event.event_id, - ) - # FIXME: Get auth chain for these state events - current_state = state if state: for e in state: + logging.info("A :) %r", e) e.internal_metadata.outlier = True try: yield self._handle_new_event(e) @@ -392,7 +393,7 @@ class FederationHandler(BaseHandler): for e in auth_chain: e.internal_metadata.outlier = True try: - yield self._handle_new_event(e, fetch_missing=False) + yield self._handle_new_event(e) except: logger.exception( "Failed to handle auth event %s", @@ -404,8 +405,7 @@ class FederationHandler(BaseHandler): e.internal_metadata.outlier = True try: yield self._handle_new_event( - e, - fetch_missing=True + e, fetch_auth_from=target_host ) except: logger.exception( @@ -682,7 +682,7 @@ class FederationHandler(BaseHandler): @defer.inlineCallbacks def _handle_new_event(self, event, state=None, backfilled=False, - current_state=None, fetch_missing=True): + current_state=None, fetch_auth_from=None): logger.debug( "_handle_new_event: Before annotate: %s, sigs: %s", @@ -703,11 +703,20 @@ class FederationHandler(BaseHandler): known_ids = set( [s.event_id for s in context.auth_events.values()] ) + for e_id, _ in event.auth_events: if e_id not in known_ids: - e = yield self.store.get_event( - e_id, allow_none=True, - ) + e = yield self.store.get_event(e_id, allow_none=True) + + if not e and fetch_auth_from is not None: + # Grab the auth_chain over federation if we are missing + # auth events. + auth_chain = yield self.replication_layer.get_event_auth( + fetch_auth_from, event.event_id, event.room_id + ) + for auth_event in auth_chain: + yield self._handle_new_event(auth_event) + e = yield self.store.get_event(e_id, allow_none=True) if not e: # TODO: Do some conflict res to make sure that we're diff --git a/synapse/util/distributor.py b/synapse/util/distributor.py index 6e69296d65..6925ac96b6 100644 --- a/synapse/util/distributor.py +++ b/synapse/util/distributor.py @@ -120,5 +120,5 @@ class Signal(object): results = [] for deferred in deferreds: result = yield deferred - results.append(results) + results.append(result) defer.returnValue(results) diff --git a/tests/federation/test_federation.py b/tests/federation/test_federation.py index 79ac1ce10d..3e484cd303 100644 --- a/tests/federation/test_federation.py +++ b/tests/federation/test_federation.py @@ -52,6 +52,7 @@ class FederationTestCase(unittest.TestCase): "get_received_txn_response", "set_received_txn_response", "get_destination_retry_timings", + "get_auth_chain", ]) self.mock_persistence.get_received_txn_response.return_value = ( defer.succeed(None) @@ -59,6 +60,7 @@ class FederationTestCase(unittest.TestCase): self.mock_persistence.get_destination_retry_timings.return_value = ( defer.succeed(DestinationsTable.EntryType("", 0, 0)) ) + self.mock_persistence.get_auth_chain.return_value = [] self.mock_config = Mock() self.mock_config.signing_key = [MockKey()] self.clock = MockClock() |