From 12ce441e67a40bb73ac5aca0283c2fe4afac4021 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 30 Oct 2014 17:00:11 +0000 Subject: Convert event ids to be of the form :example.com --- synapse/types.py | 10 ++++++++++ 1 file changed, 10 insertions(+) (limited to 'synapse/types.py') diff --git a/synapse/types.py b/synapse/types.py index c51bc8e4f2..649ff2f7d7 100644 --- a/synapse/types.py +++ b/synapse/types.py @@ -78,6 +78,11 @@ class DomainSpecificString( """Create a structure on the local domain""" return cls(localpart=localpart, domain=hs.hostname, is_mine=True) + @classmethod + def create(cls, localpart, domain, hs): + is_mine = domain == hs.hostname + return cls(localpart=localpart, domain=domain, is_mine=is_mine) + class UserID(DomainSpecificString): """Structure representing a user ID.""" @@ -94,6 +99,11 @@ class RoomID(DomainSpecificString): SIGIL = "!" +class EventID(DomainSpecificString): + """Structure representing an event id. """ + SIGIL = "$" + + class StreamToken( namedtuple( "Token", -- cgit 1.5.1 From 440cbd5235e7e23dfe97d8e3d394cc0d35b35fd6 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 4 Nov 2014 14:17:55 +0000 Subject: Add support for sending failures --- synapse/federation/replication.py | 30 ++++++++++++++++++++++++++++-- synapse/federation/units.py | 1 + synapse/types.py | 34 ++++++++++++++++++++++++++++++++++ 3 files changed, 63 insertions(+), 2 deletions(-) (limited to 'synapse/types.py') diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py index 99dd390a64..680e7322a6 100644 --- a/synapse/federation/replication.py +++ b/synapse/federation/replication.py @@ -143,6 +143,11 @@ class ReplicationLayer(object): self._transaction_queue.enqueue_edu(edu) return defer.succeed(None) + @log_function + def send_failure(self, failure, destination): + self._transaction_queue.enqueue_failure(failure, destination) + return defer.succeed(None) + @log_function def make_query(self, destination, query_type, args, retry_on_dns_fail=True): @@ -558,6 +563,9 @@ class _TransactionQueue(object): # destination -> list of tuple(edu, deferred) self.pending_edus_by_dest = {} + # destination -> list of tuple(failure, deferred) + self.pending_failures_by_dest = {} + # HACK to get unique tx id self._next_txn_id = int(self._clock.time_msec()) @@ -610,6 +618,18 @@ class _TransactionQueue(object): return deferred + @defer.inlineCallbacks + def enqueue_failure(self, failure, destination): + deferred = defer.Deferred() + + self.pending_failures_by_dest.setdefault( + destination, [] + ).append( + (failure, deferred) + ) + + yield deferred + @defer.inlineCallbacks @log_function def _attempt_new_transaction(self, destination): @@ -619,8 +639,9 @@ class _TransactionQueue(object): # list of (pending_pdu, deferred, order) pending_pdus = self.pending_pdus_by_dest.pop(destination, []) pending_edus = self.pending_edus_by_dest.pop(destination, []) + pending_failures = self.pending_failures_by_dest(destination, []) - if not pending_pdus and not pending_edus: + if not pending_pdus and not pending_edus and not pending_failures: return logger.debug("TX [%s] Attempting new transaction", destination) @@ -630,7 +651,11 @@ class _TransactionQueue(object): 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] + failures = [x[0].get_dict() for x in pending_failures] + deferreds = [ + x[1] + for x in pending_pdus + pending_edus + pending_failures + ] try: self.pending_transactions[destination] = 1 @@ -644,6 +669,7 @@ class _TransactionQueue(object): destination=destination, pdus=pdus, edus=edus, + pdu_failures=failures, ) self._next_txn_id += 1 diff --git a/synapse/federation/units.py b/synapse/federation/units.py index 9b25556707..2070ffe1e2 100644 --- a/synapse/federation/units.py +++ b/synapse/federation/units.py @@ -157,6 +157,7 @@ class Transaction(JsonEncodedObject): "edus", "transaction_id", "destination", + "pdu_failures", ] internal_keys = [ diff --git a/synapse/types.py b/synapse/types.py index 649ff2f7d7..8fac20fd2e 100644 --- a/synapse/types.py +++ b/synapse/types.py @@ -128,3 +128,37 @@ class StreamToken( d = self._asdict() d[key] = new_value return StreamToken(**d) + + +class FederationError(RuntimeError): + """ This class is used to inform remote home servers about erroneous + PDUs they sent us. + + FATAL: The remote server could not interpret the source event. + (e.g., it was missing a required field) + ERROR: The remote server interpreted the event, but it failed some other + check (e.g. auth) + WARN: The remote server accepted the event, but believes some part of it + is wrong (e.g., it referred to an invalid event) + """ + + def __init__(self, level, code, reason, affected, source=None): + if level not in ["FATAL", "ERROR", "WARN"]: + raise ValueError("Level is not valid: %s" % (level,)) + self.level = level + self.code = code + self.reason = reason + self.affected = affected + self.source = source + + msg = "%s %s: %s" % (level, code, reason,) + super(FederationError, self).__init__(msg) + + def get_dict(self): + return { + "level": self.level, + "code": self.code, + "reason": self.reason, + "affected": self.affected, + "source": self.source if self.source else self.affected, + } -- cgit 1.5.1 From 8918422156a3555a61b7e5ba2ee0c04eb3e52cb5 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 4 Nov 2014 15:10:27 +0000 Subject: Move FederationError to synapse.api.errors --- synapse/api/errors.py | 34 ++++++++++++++++++++++++++++++++++ synapse/types.py | 34 ---------------------------------- 2 files changed, 34 insertions(+), 34 deletions(-) (limited to 'synapse/types.py') diff --git a/synapse/api/errors.py b/synapse/api/errors.py index 38ccb4f9d1..33d15072af 100644 --- a/synapse/api/errors.py +++ b/synapse/api/errors.py @@ -158,3 +158,37 @@ def cs_error(msg, code=Codes.UNKNOWN, **kwargs): for key, value in kwargs.iteritems(): err[key] = value return err + + +class FederationError(RuntimeError): + """ This class is used to inform remote home servers about erroneous + PDUs they sent us. + + FATAL: The remote server could not interpret the source event. + (e.g., it was missing a required field) + ERROR: The remote server interpreted the event, but it failed some other + check (e.g. auth) + WARN: The remote server accepted the event, but believes some part of it + is wrong (e.g., it referred to an invalid event) + """ + + def __init__(self, level, code, reason, affected, source=None): + if level not in ["FATAL", "ERROR", "WARN"]: + raise ValueError("Level is not valid: %s" % (level,)) + self.level = level + self.code = code + self.reason = reason + self.affected = affected + self.source = source + + msg = "%s %s: %s" % (level, code, reason,) + super(FederationError, self).__init__(msg) + + def get_dict(self): + return { + "level": self.level, + "code": self.code, + "reason": self.reason, + "affected": self.affected, + "source": self.source if self.source else self.affected, + } diff --git a/synapse/types.py b/synapse/types.py index 8fac20fd2e..649ff2f7d7 100644 --- a/synapse/types.py +++ b/synapse/types.py @@ -128,37 +128,3 @@ class StreamToken( d = self._asdict() d[key] = new_value return StreamToken(**d) - - -class FederationError(RuntimeError): - """ This class is used to inform remote home servers about erroneous - PDUs they sent us. - - FATAL: The remote server could not interpret the source event. - (e.g., it was missing a required field) - ERROR: The remote server interpreted the event, but it failed some other - check (e.g. auth) - WARN: The remote server accepted the event, but believes some part of it - is wrong (e.g., it referred to an invalid event) - """ - - def __init__(self, level, code, reason, affected, source=None): - if level not in ["FATAL", "ERROR", "WARN"]: - raise ValueError("Level is not valid: %s" % (level,)) - self.level = level - self.code = code - self.reason = reason - self.affected = affected - self.source = source - - msg = "%s %s: %s" % (level, code, reason,) - super(FederationError, self).__init__(msg) - - def get_dict(self): - return { - "level": self.level, - "code": self.code, - "reason": self.reason, - "affected": self.affected, - "source": self.source if self.source else self.affected, - } -- cgit 1.5.1