From 8e358ef35a8e2640176003cdb2396a9b4d71fbce Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 28 Oct 2014 10:34:05 +0000 Subject: Add timer to LoggingTransaction --- synapse/storage/_base.py | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) (limited to 'synapse/storage/_base.py') diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 65a86e9056..e839704a8c 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -23,6 +23,7 @@ from synapse.util.logutils import log_function import collections import copy import json +import time logger = logging.getLogger(__name__) @@ -61,9 +62,15 @@ class LoggingTransaction(object): # TODO(paul): Here would be an excellent place to put some timing # measurements, and log (warning?) slow queries. - return object.__getattribute__(self, "txn").execute( - sql, *args, **kwargs - ) + + start = time.clock() * 1000 + try: + return object.__getattribute__(self, "txn").execute( + sql, *args, **kwargs + ) + finally: + end = time.clock() * 1000 + sql_logger.debug("[SQL time] %f", end - start) class SQLBaseStore(object): -- cgit 1.5.1 From 967ce43b59e90f36c21f3a493bc81ab4e57dae69 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 28 Oct 2014 10:53:11 +0000 Subject: Clean up LoggingTransaction --- synapse/storage/_base.py | 14 ++++---------- 1 file changed, 4 insertions(+), 10 deletions(-) (limited to 'synapse/storage/_base.py') diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index e839704a8c..d3e8741889 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -39,14 +39,11 @@ class LoggingTransaction(object): def __init__(self, txn): object.__setattr__(self, "txn", txn) - def __getattribute__(self, name): - if name == "execute": - return object.__getattribute__(self, "execute") - - return getattr(object.__getattribute__(self, "txn"), name) + def __getattr__(self, name): + return getattr(self.txn, name) def __setattr__(self, name, value): - setattr(object.__getattribute__(self, "txn"), name, value) + setattr(self.txn, name, value) def execute(self, sql, *args, **kwargs): # TODO(paul): Maybe use 'info' and 'debug' for values? @@ -60,12 +57,9 @@ class LoggingTransaction(object): # Don't let logging failures stop SQL from working pass - # TODO(paul): Here would be an excellent place to put some timing - # measurements, and log (warning?) slow queries. - start = time.clock() * 1000 try: - return object.__getattribute__(self, "txn").execute( + return self.txn.execute( sql, *args, **kwargs ) finally: -- cgit 1.5.1 From da1dda3e1d9d3272527d35c23162c4baf7339d74 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 28 Oct 2014 11:18:04 +0000 Subject: Add transaction level logging and timing information. Add a _simple_delete method --- synapse/storage/__init__.py | 3 +- synapse/storage/_base.py | 74 ++++++++++++++++++++++++++++++++--------- synapse/storage/directory.py | 1 + synapse/storage/pdu.py | 13 +++++++- synapse/storage/registration.py | 7 ++-- synapse/storage/room.py | 2 ++ synapse/storage/state.py | 1 + synapse/storage/stream.py | 5 ++- synapse/storage/transactions.py | 6 ++++ 9 files changed, 91 insertions(+), 21 deletions(-) (limited to 'synapse/storage/_base.py') diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index 15a72d0cd7..a50e19349a 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -109,6 +109,7 @@ class DataStore(RoomMemberStore, RoomStore, try: yield self.runInteraction( + "persist_event", self._persist_pdu_event_txn, pdu=pdu, event=event, @@ -394,7 +395,7 @@ class DataStore(RoomMemberStore, RoomStore, prev_state_pdu=prev_state_pdu, ) - return self.runInteraction(_snapshot) + return self.runInteraction("snapshot_room", _snapshot) class Snapshot(object): diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index d3e8741889..1192216971 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -29,15 +29,17 @@ import time logger = logging.getLogger(__name__) sql_logger = logging.getLogger("synapse.storage.SQL") +transaction_logger = logging.getLogger("synapse.storage.txn") class LoggingTransaction(object): """An object that almost-transparently proxies for the 'txn' object passed to the constructor. Adds logging to the .execute() method.""" - __slots__ = ["txn"] + __slots__ = ["txn", "name"] - def __init__(self, txn): + def __init__(self, txn, name): object.__setattr__(self, "txn", txn) + object.__setattr__(self, "name", name) def __getattr__(self, name): return getattr(self.txn, name) @@ -47,12 +49,15 @@ class LoggingTransaction(object): def execute(self, sql, *args, **kwargs): # TODO(paul): Maybe use 'info' and 'debug' for values? - sql_logger.debug("[SQL] %s", sql) + sql_logger.debug("[SQL] {%s} %s", self.name, sql) try: if args and args[0]: values = args[0] - sql_logger.debug("[SQL values] " + - ", ".join(("<%s>",) * len(values)), *values) + sql_logger.debug( + "[SQL values] {%s} " + ", ".join(("<%s>",) * len(values)), + self.name, + *values + ) except: # Don't let logging failures stop SQL from working pass @@ -64,10 +69,11 @@ class LoggingTransaction(object): ) finally: end = time.clock() * 1000 - sql_logger.debug("[SQL time] %f", end - start) + sql_logger.debug("[SQL time] {%s} %f", self.name, end - start) class SQLBaseStore(object): + _TXN_ID = 0 def __init__(self, hs): self.hs = hs @@ -75,10 +81,24 @@ class SQLBaseStore(object): self.event_factory = hs.get_event_factory() self._clock = hs.get_clock() - def runInteraction(self, func, *args, **kwargs): + def runInteraction(self, desc, func, *args, **kwargs): """Wraps the .runInteraction() method on the underlying db_pool.""" def inner_func(txn, *args, **kwargs): - return func(LoggingTransaction(txn), *args, **kwargs) + start = time.clock() * 1000 + txn_id = str(SQLBaseStore._TXN_ID) + SQLBaseStore._TXN_ID += 1 + + name = "%s-%s" % (desc, txn_id, ) + + transaction_logger.debug("[TXN START] {%s}", name) + try: + return func(LoggingTransaction(txn, name), *args, **kwargs) + finally: + end = time.clock() * 1000 + transaction_logger.debug( + "[TXN END] {%s} %f", + name, end - start + ) return self._db_pool.runInteraction(inner_func, *args, **kwargs) @@ -114,7 +134,7 @@ class SQLBaseStore(object): else: return cursor.fetchall() - return self.runInteraction(interaction) + return self.runInteraction("_execute", interaction) def _execute_and_decode(self, query, *args): return self._execute(self.cursor_to_dict, query, *args) @@ -131,6 +151,7 @@ class SQLBaseStore(object): or_replace : bool; if True performs an INSERT OR REPLACE """ return self.runInteraction( + "_simple_insert", self._simple_insert_txn, table, values, or_replace=or_replace, or_ignore=or_ignore, ) @@ -168,6 +189,7 @@ class SQLBaseStore(object): statement returns no rows """ return self._simple_selectupdate_one( + "_simple_select_one", table, keyvalues, retcols=retcols, allow_none=allow_none ) @@ -217,7 +239,7 @@ class SQLBaseStore(object): txn.execute(sql, keyvalues.values()) return txn.fetchall() - res = yield self.runInteraction(func) + res = yield self.runInteraction("_simple_select_onecol", func) defer.returnValue([r[0] for r in res]) @@ -240,7 +262,7 @@ class SQLBaseStore(object): txn.execute(sql, keyvalues.values()) return self.cursor_to_dict(txn) - return self.runInteraction(func) + return self.runInteraction("_simple_select_list", func) def _simple_update_one(self, table, keyvalues, updatevalues, retcols=None): @@ -308,7 +330,7 @@ class SQLBaseStore(object): raise StoreError(500, "More than one row matched") return ret - return self.runInteraction(func) + return self.runInteraction("_simple_selectupdate_one", func) def _simple_delete_one(self, table, keyvalues): """Executes a DELETE query on the named table, expecting to delete a @@ -320,7 +342,7 @@ class SQLBaseStore(object): """ sql = "DELETE FROM %s WHERE %s" % ( table, - " AND ".join("%s = ?" % (k) for k in keyvalues) + " AND ".join("%s = ?" % (k, ) for k in keyvalues) ) def func(txn): @@ -329,7 +351,25 @@ class SQLBaseStore(object): raise StoreError(404, "No row found") if txn.rowcount > 1: raise StoreError(500, "more than one row matched") - return self.runInteraction(func) + return self.runInteraction("_simple_delete_one", func) + + def _simple_delete(self, table, keyvalues): + """Executes a DELETE query on the named table. + + Args: + table : string giving the table name + keyvalues : dict of column names and values to select the row with + """ + + return self.runInteraction("_simple_delete", self._simple_delete_txn) + + def _simple_delete_txn(self, txn, table, keyvalues): + sql = "DELETE FROM %s WHERE %s" % ( + table, + " AND ".join("%s = ?" % (k, ) for k in keyvalues) + ) + + return txn.execute(sql, keyvalues.values()) def _simple_max_id(self, table): """Executes a SELECT query on the named table, expecting to return the @@ -347,7 +387,7 @@ class SQLBaseStore(object): return 0 return max_id - return self.runInteraction(func) + return self.runInteraction("_simple_max_id", func) def _parse_event_from_row(self, row_dict): d = copy.deepcopy({k: v for k, v in row_dict.items()}) @@ -371,7 +411,9 @@ class SQLBaseStore(object): ) def _parse_events(self, rows): - return self.runInteraction(self._parse_events_txn, rows) + return self.runInteraction( + "_parse_events", self._parse_events_txn, rows + ) def _parse_events_txn(self, txn, rows): events = [self._parse_event_from_row(r) for r in rows] diff --git a/synapse/storage/directory.py b/synapse/storage/directory.py index 52373a28a6..d6a7113b9c 100644 --- a/synapse/storage/directory.py +++ b/synapse/storage/directory.py @@ -95,6 +95,7 @@ class DirectoryStore(SQLBaseStore): def delete_room_alias(self, room_alias): return self.runInteraction( + "delete_room_alias", self._delete_room_alias_txn, room_alias, ) diff --git a/synapse/storage/pdu.py b/synapse/storage/pdu.py index 9bdc831fd8..4a4341907b 100644 --- a/synapse/storage/pdu.py +++ b/synapse/storage/pdu.py @@ -47,7 +47,7 @@ class PduStore(SQLBaseStore): """ return self.runInteraction( - self._get_pdu_tuple, pdu_id, origin + "get_pdu", self._get_pdu_tuple, pdu_id, origin ) def _get_pdu_tuple(self, txn, pdu_id, origin): @@ -108,6 +108,7 @@ class PduStore(SQLBaseStore): """ return self.runInteraction( + "get_current_state_for_context", self._get_current_state_for_context, context ) @@ -156,6 +157,7 @@ class PduStore(SQLBaseStore): """ return self.runInteraction( + "mark_pdu_as_processed", self._mark_as_processed, pdu_id, pdu_origin ) @@ -165,6 +167,7 @@ class PduStore(SQLBaseStore): def get_all_pdus_from_context(self, context): """Get a list of all PDUs for a given context.""" return self.runInteraction( + "get_all_pdus_from_context", self._get_all_pdus_from_context, context, ) @@ -192,6 +195,7 @@ class PduStore(SQLBaseStore): list: A list of PduTuples """ return self.runInteraction( + "get_backfill", self._get_backfill, context, pdu_list, limit ) @@ -253,6 +257,7 @@ class PduStore(SQLBaseStore): context (str) """ return self.runInteraction( + "get_min_depth_for_context", self._get_min_depth_for_context, context ) @@ -291,6 +296,7 @@ class PduStore(SQLBaseStore): def get_latest_pdus_in_context(self, context): return self.runInteraction( + "get_latest_pdus_in_context", self._get_latest_pdus_in_context, context ) @@ -370,6 +376,7 @@ class PduStore(SQLBaseStore): """ return self.runInteraction( + "is_pdu_new", self._is_pdu_new, pdu_id=pdu_id, origin=origin, @@ -523,6 +530,7 @@ class StatePduStore(SQLBaseStore): def get_unresolved_state_tree(self, new_state_pdu): return self.runInteraction( + "get_unresolved_state_tree", self._get_unresolved_state_tree, new_state_pdu ) @@ -562,6 +570,7 @@ class StatePduStore(SQLBaseStore): def update_current_state(self, pdu_id, origin, context, pdu_type, state_key): return self.runInteraction( + "update_current_state", self._update_current_state, pdu_id, origin, context, pdu_type, state_key ) @@ -601,6 +610,7 @@ class StatePduStore(SQLBaseStore): """ return self.runInteraction( + "get_current_state_pdu", self._get_current_state_pdu, context, pdu_type, state_key ) @@ -660,6 +670,7 @@ class StatePduStore(SQLBaseStore): bool: True if the new_pdu clobbered the current state, False if not """ return self.runInteraction( + "handle_new_state", self._handle_new_state, new_pdu ) diff --git a/synapse/storage/registration.py b/synapse/storage/registration.py index 719806f82b..a2ca6f9a69 100644 --- a/synapse/storage/registration.py +++ b/synapse/storage/registration.py @@ -62,8 +62,10 @@ class RegistrationStore(SQLBaseStore): Raises: StoreError if the user_id could not be registered. """ - yield self.runInteraction(self._register, user_id, token, - password_hash) + yield self.runInteraction( + "register", + self._register, user_id, token, password_hash + ) def _register(self, txn, user_id, token, password_hash): now = int(self.clock.time()) @@ -100,6 +102,7 @@ class RegistrationStore(SQLBaseStore): StoreError if no user was found. """ return self.runInteraction( + "get_user_by_token", self._query_for_auth, token ) diff --git a/synapse/storage/room.py b/synapse/storage/room.py index 8cd46334cf..7e48ce9cc3 100644 --- a/synapse/storage/room.py +++ b/synapse/storage/room.py @@ -150,6 +150,7 @@ class RoomStore(SQLBaseStore): def get_power_level(self, room_id, user_id): return self.runInteraction( + "get_power_level", self._get_power_level, room_id, user_id, ) @@ -183,6 +184,7 @@ class RoomStore(SQLBaseStore): def get_ops_levels(self, room_id): return self.runInteraction( + "get_ops_levels", self._get_ops_levels, room_id, ) diff --git a/synapse/storage/state.py b/synapse/storage/state.py index 0aa979c9f0..e08acd6404 100644 --- a/synapse/storage/state.py +++ b/synapse/storage/state.py @@ -59,6 +59,7 @@ class StateStore(SQLBaseStore): def store_state_groups(self, event): return self.runInteraction( + "store_state_groups", self._store_state_groups_txn, event ) diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py index d61f909939..8f7f61d29d 100644 --- a/synapse/storage/stream.py +++ b/synapse/storage/stream.py @@ -309,7 +309,10 @@ class StreamStore(SQLBaseStore): defer.returnValue(ret) def get_room_events_max_id(self): - return self.runInteraction(self._get_room_events_max_id_txn) + return self.runInteraction( + "get_room_events_max_id", + self._get_room_events_max_id_txn + ) def _get_room_events_max_id_txn(self, txn): txn.execute( diff --git a/synapse/storage/transactions.py b/synapse/storage/transactions.py index 2ba8e30efe..908014d38b 100644 --- a/synapse/storage/transactions.py +++ b/synapse/storage/transactions.py @@ -42,6 +42,7 @@ class TransactionStore(SQLBaseStore): """ return self.runInteraction( + "get_received_txn_response", self._get_received_txn_response, transaction_id, origin ) @@ -73,6 +74,7 @@ class TransactionStore(SQLBaseStore): """ return self.runInteraction( + "set_received_txn_response", self._set_received_txn_response, transaction_id, origin, code, response_dict ) @@ -106,6 +108,7 @@ class TransactionStore(SQLBaseStore): """ return self.runInteraction( + "prep_send_transaction", self._prep_send_transaction, transaction_id, destination, origin_server_ts, pdu_list ) @@ -161,6 +164,7 @@ class TransactionStore(SQLBaseStore): response_json (str) """ return self.runInteraction( + "delivered_txn", self._delivered_txn, transaction_id, destination, code, response_dict ) @@ -186,6 +190,7 @@ class TransactionStore(SQLBaseStore): list: A list of `ReceivedTransactionsTable.EntryType` """ return self.runInteraction( + "get_transactions_after", self._get_transactions_after, transaction_id, destination ) @@ -216,6 +221,7 @@ class TransactionStore(SQLBaseStore): list: A list of PduTuple """ return self.runInteraction( + "get_pdus_after_transaction", self._get_pdus_after_transaction, transaction_id, destination ) -- cgit 1.5.1 From 2d1dfb3b34583a4de7e1e53f685c2564a7fc731f Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 28 Oct 2014 16:42:35 +0000 Subject: Begin implementing all the PDU storage stuff in Events land --- synapse/api/events/__init__.py | 4 +- synapse/federation/pdu_codec.py | 11 ++- synapse/storage/__init__.py | 72 ++++++++++---- synapse/storage/_base.py | 53 +++++++---- synapse/storage/event_federation.py | 143 ++++++++++++++++++++++++++++ synapse/storage/schema/event_edges.sql | 51 ++++++++++ synapse/storage/schema/event_signatures.sql | 65 +++++++++++++ synapse/storage/schema/im.sql | 1 + synapse/storage/signatures.py | 127 ++++++++++++++++++++++++ 9 files changed, 485 insertions(+), 42 deletions(-) create mode 100644 synapse/storage/event_federation.py create mode 100644 synapse/storage/schema/event_edges.sql create mode 100644 synapse/storage/schema/event_signatures.sql (limited to 'synapse/storage/_base.py') diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py index a5a55742e0..b855811b98 100644 --- a/synapse/api/events/__init__.py +++ b/synapse/api/events/__init__.py @@ -71,7 +71,9 @@ class SynapseEvent(JsonEncodedObject): "outlier", "power_level", "redacted", - "prev_pdus", + "prev_events", + "hashes", + "signatures", ] required_keys = [ diff --git a/synapse/federation/pdu_codec.py b/synapse/federation/pdu_codec.py index 991aae2a56..2cd591410b 100644 --- a/synapse/federation/pdu_codec.py +++ b/synapse/federation/pdu_codec.py @@ -47,7 +47,10 @@ class PduCodec(object): kwargs["event_id"] = encode_event_id(pdu.pdu_id, pdu.origin) kwargs["room_id"] = pdu.context kwargs["etype"] = pdu.pdu_type - kwargs["prev_pdus"] = pdu.prev_pdus + kwargs["prev_events"] = [ + encode_event_id(i, o) + for i, o in pdu.prev_pdus + ] if hasattr(pdu, "prev_state_id") and hasattr(pdu, "prev_state_origin"): kwargs["prev_state"] = encode_event_id( @@ -78,8 +81,8 @@ class PduCodec(object): d["context"] = event.room_id d["pdu_type"] = event.type - if hasattr(event, "prev_pdus"): - d["prev_pdus"] = event.prev_pdus + if hasattr(event, "prev_events"): + d["prev_pdus"] = [decode_event_id(e) for e in event.prev_events] if hasattr(event, "prev_state"): d["prev_state_id"], d["prev_state_origin"] = ( @@ -92,7 +95,7 @@ class PduCodec(object): kwargs = copy.deepcopy(event.unrecognized_keys) kwargs.update({ k: v for k, v in d.items() - if k not in ["event_id", "room_id", "type"] + if k not in ["event_id", "room_id", "type", "prev_events"] }) if "origin_server_ts" not in kwargs: diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index a50e19349a..678de0cf50 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -40,6 +40,7 @@ from .stream import StreamStore from .pdu import StatePduStore, PduStore, PdusTable from .transactions import TransactionStore from .keys import KeyStore +from .event_federation import EventFederationStore from .state import StateStore from .signatures import SignatureStore @@ -69,6 +70,7 @@ SCHEMAS = [ "redactions", "state", "signatures", + "event_edges", ] @@ -83,10 +85,12 @@ class _RollbackButIsFineException(Exception): """ pass + class DataStore(RoomMemberStore, RoomStore, RegistrationStore, StreamStore, ProfileStore, FeedbackStore, PresenceStore, PduStore, StatePduStore, TransactionStore, - DirectoryStore, KeyStore, StateStore, SignatureStore): + DirectoryStore, KeyStore, StateStore, SignatureStore, + EventFederationStore, ): def __init__(self, hs): super(DataStore, self).__init__(hs) @@ -230,6 +234,10 @@ class DataStore(RoomMemberStore, RoomStore, elif event.type == RoomRedactionEvent.TYPE: self._store_redaction(txn, event) + outlier = False + if hasattr(event, "outlier"): + outlier = event.outlier + vals = { "topological_ordering": event.depth, "event_id": event.event_id, @@ -237,20 +245,20 @@ class DataStore(RoomMemberStore, RoomStore, "room_id": event.room_id, "content": json.dumps(event.content), "processed": True, + "outlier": outlier, + "depth": event.depth, } if stream_ordering is not None: vals["stream_ordering"] = stream_ordering - if hasattr(event, "outlier"): - vals["outlier"] = event.outlier - else: - vals["outlier"] = False - unrec = { k: v for k, v in event.get_full_dict().items() - if k not in vals.keys() and k not in ["redacted", "redacted_because"] + if k not in vals.keys() and k not in [ + "redacted", "redacted_because", "signatures", "hashes", + "prev_events", + ] } vals["unrecognized_keys"] = json.dumps(unrec) @@ -264,6 +272,14 @@ class DataStore(RoomMemberStore, RoomStore, ) raise _RollbackButIsFineException("_persist_event") + self._handle_prev_events( + txn, + outlier=outlier, + event_id=event.event_id, + prev_events=event.prev_events, + room_id=event.room_id, + ) + self._store_state_groups_txn(txn, event) is_state = hasattr(event, "state_key") and event.state_key is not None @@ -291,6 +307,28 @@ class DataStore(RoomMemberStore, RoomStore, } ) + signatures = event.signatures.get(event.origin, {}) + + for key_id, signature_base64 in signatures.items(): + signature_bytes = decode_base64(signature_base64) + self._store_event_origin_signature_txn( + txn, event.event_id, key_id, signature_bytes, + ) + + for prev_event_id, prev_hashes in event.prev_events: + for alg, hash_base64 in prev_hashes.items(): + hash_bytes = decode_base64(hash_base64) + self._store_prev_event_hash_txn( + txn, event.event_id, prev_event_id, alg, hash_bytes + ) + + (ref_alg, ref_hash_bytes) = compute_pdu_event_reference_hash(pdu) + self._store_pdu_reference_hash_txn( + txn, pdu.pdu_id, pdu.origin, ref_alg, ref_hash_bytes + ) + + self._update_min_depth_for_room_txn(txn, event.room_id, event.depth) + def _store_redaction(self, txn, event): txn.execute( "INSERT OR IGNORE INTO redactions " @@ -373,7 +411,7 @@ class DataStore(RoomMemberStore, RoomStore, """ def _snapshot(txn): membership_state = self._get_room_member(txn, user_id, room_id) - prev_pdus = self._get_latest_pdus_in_context( + prev_events = self._get_latest_events_in_room( txn, room_id ) @@ -388,7 +426,7 @@ class DataStore(RoomMemberStore, RoomStore, store=self, room_id=room_id, user_id=user_id, - prev_pdus=prev_pdus, + prev_events=prev_events, membership_state=membership_state, state_type=state_type, state_key=state_key, @@ -404,7 +442,7 @@ class Snapshot(object): store (DataStore): The datastore. room_id (RoomId): The room of the snapshot. user_id (UserId): The user this snapshot is for. - prev_pdus (list): The list of PDU ids this snapshot is after. + prev_events (list): The list of event ids this snapshot is after. membership_state (RoomMemberEvent): The current state of the user in the room. state_type (str, optional): State type captured by the snapshot @@ -413,29 +451,29 @@ class Snapshot(object): the previous value of the state type and key in the room. """ - def __init__(self, store, room_id, user_id, prev_pdus, + def __init__(self, store, room_id, user_id, prev_events, membership_state, state_type=None, state_key=None, prev_state_pdu=None): self.store = store self.room_id = room_id self.user_id = user_id - self.prev_pdus = prev_pdus + self.prev_events = prev_events self.membership_state = membership_state self.state_type = state_type self.state_key = state_key self.prev_state_pdu = prev_state_pdu def fill_out_prev_events(self, event): - if hasattr(event, "prev_pdus"): + if hasattr(event, "prev_events"): return - event.prev_pdus = [ + event.prev_events = [ (p_id, origin, hashes) - for p_id, origin, hashes, _ in self.prev_pdus + for p_id, origin, hashes, _ in self.prev_events ] - if self.prev_pdus: - event.depth = max([int(v) for _, _, _, v in self.prev_pdus]) + 1 + if self.prev_events: + event.depth = max([int(v) for _, _, _, v in self.prev_events]) + 1 else: event.depth = 0 diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 1192216971..30732caa83 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -193,7 +193,6 @@ class SQLBaseStore(object): table, keyvalues, retcols=retcols, allow_none=allow_none ) - @defer.inlineCallbacks def _simple_select_one_onecol(self, table, keyvalues, retcol, allow_none=False): """Executes a SELECT query on the named table, which is expected to @@ -204,19 +203,41 @@ class SQLBaseStore(object): keyvalues : dict of column names and values to select the row with retcol : string giving the name of the column to return """ - ret = yield self._simple_select_one( + return self.runInteraction( + "_simple_select_one_onecol_txn", + self._simple_select_one_onecol_txn, + table, keyvalues, retcol, allow_none=allow_none, + ) + + def _simple_select_one_onecol_txn(self, txn, table, keyvalues, retcol, + allow_none=False): + ret = self._simple_select_onecol_txn( + txn, table=table, keyvalues=keyvalues, - retcols=[retcol], - allow_none=allow_none + retcols=retcol, ) if ret: - defer.returnValue(ret[retcol]) + return ret[retcol] else: - defer.returnValue(None) + if allow_none: + return None + else: + raise StoreError(404, "No row found") + + def _simple_select_onecol_txn(self, txn, table, keyvalues, retcol): + sql = "SELECT %(retcol)s FROM %(table)s WHERE %(where)s" % { + "retcol": retcol, + "table": table, + "where": " AND ".join("%s = ?" % k for k in keyvalues.keys()), + } + + txn.execute(sql, keyvalues.values()) + + return [r[0] for r in txn.fetchall()] + - @defer.inlineCallbacks def _simple_select_onecol(self, table, keyvalues, retcol): """Executes a SELECT query on the named table, which returns a list comprising of the values of the named column from the selected rows. @@ -229,19 +250,11 @@ class SQLBaseStore(object): Returns: Deferred: Results in a list """ - sql = "SELECT %(retcol)s FROM %(table)s WHERE %(where)s" % { - "retcol": retcol, - "table": table, - "where": " AND ".join("%s = ?" % k for k in keyvalues.keys()), - } - - def func(txn): - txn.execute(sql, keyvalues.values()) - return txn.fetchall() - - res = yield self.runInteraction("_simple_select_onecol", func) - - defer.returnValue([r[0] for r in res]) + return self.runInteraction( + "_simple_select_onecol", + self._simple_select_onecol_txn, + table, keyvalues, retcol + ) def _simple_select_list(self, table, keyvalues, retcols): """Executes a SELECT query on the named table, which may return zero or diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py new file mode 100644 index 0000000000..27ad9aea4d --- /dev/null +++ b/synapse/storage/event_federation.py @@ -0,0 +1,143 @@ +# -*- coding: utf-8 -*- +# Copyright 2014 OpenMarket Ltd +# +# 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. + +from ._base import SQLBaseStore +from twisted.internet import defer + +import logging + + +logger = logging.getLogger(__name__) + + +class EventFederationStore(SQLBaseStore): + + def _get_latest_events_in_room(self, txn, room_id): + self._simple_select_onecol_txn( + txn, + table="event_forward_extremities", + keyvalues={ + "room_id": room_id, + }, + retcol="event_id", + ) + + results = [] + for pdu_id, origin, depth in txn.fetchall(): + hashes = self._get_pdu_reference_hashes_txn(txn, pdu_id, origin) + sha256_bytes = hashes["sha256"] + prev_hashes = {"sha256": encode_base64(sha256_bytes)} + results.append((pdu_id, origin, prev_hashes, depth)) + + def _get_min_depth_interaction(self, txn, room_id): + min_depth = self._simple_select_one_onecol_txn( + txn, + table="room_depth", + keyvalues={"room_id": room_id,}, + retcol="min_depth", + allow_none=True, + ) + + return int(min_depth) if min_depth is not None else None + + def _update_min_depth_for_room_txn(self, txn, room_id, depth): + min_depth = self._get_min_depth_interaction(txn, room_id) + + do_insert = depth < min_depth if min_depth else True + + if do_insert: + self._simple_insert_txn( + txn, + table="room_depth", + values={ + "room_id": room_id, + "min_depth": depth, + }, + or_replace=True, + ) + + def _handle_prev_events(self, txn, outlier, event_id, prev_events, + room_id): + for e_id in prev_events: + # TODO (erikj): This could be done as a bulk insert + self._simple_insert_txn( + txn, + table="event_edges", + values={ + "event_id": event_id, + "prev_event": e_id, + "room_id": room_id, + } + ) + + # Update the extremities table if this is not an outlier. + if not outlier: + for e_id in prev_events: + # TODO (erikj): This could be done as a bulk insert + self._simple_delete_txn( + txn, + table="event_forward_extremities", + keyvalues={ + "event_id": e_id, + "room_id": room_id, + } + ) + + + + # We only insert as a forward extremity the new pdu if there are no + # other pdus that reference it as a prev pdu + query = ( + "INSERT INTO %(table)s (event_id, room_id) " + "SELECT ?, ? WHERE NOT EXISTS (" + "SELECT 1 FROM %(event_edges)s WHERE " + "prev_event_id = ? " + ")" + ) % { + "table": "event_forward_extremities", + "event_edges": "event_edges", + } + + logger.debug("query: %s", query) + + txn.execute(query, (event_id, room_id, event_id)) + + # Insert all the prev_pdus as a backwards thing, they'll get + # deleted in a second if they're incorrect anyway. + for e_id in prev_events: + # TODO (erikj): This could be done as a bulk insert + self._simple_insert_txn( + txn, + table="event_backward_extremities", + values={ + "event_id": e_id, + "room_id": room_id, + } + ) + + # Also delete from the backwards extremities table all ones that + # reference pdus that we have already seen + query = ( + "DELETE FROM %(event_back)s as b WHERE EXISTS (" + "SELECT 1 FROM %(events)s AS events " + "WHERE " + "b.event_id = events.event_id " + "AND not events.outlier " + ")" + ) % { + "event_back": "event_backward_extremities", + "events": "events", + } + txn.execute(query) \ No newline at end of file diff --git a/synapse/storage/schema/event_edges.sql b/synapse/storage/schema/event_edges.sql new file mode 100644 index 0000000000..6a28314ece --- /dev/null +++ b/synapse/storage/schema/event_edges.sql @@ -0,0 +1,51 @@ + +CREATE TABLE IF NOT EXISTS event_forward_extremities( + event_id TEXT, + room_id TEXT, + CONSTRAINT uniqueness UNIQUE (event_id, room_id) ON CONFLICT REPLACE +); + +CREATE INDEX IF NOT EXISTS ev_extrem_room ON event_forward_extremities(room_id); +CREATE INDEX IF NOT EXISTS ev_extrem_id ON event_forward_extremities(event_id); +-- + +CREATE TABLE IF NOT EXISTS event_backward_extremities( + event_id TEXT, + room_id TEXT, + CONSTRAINT uniqueness UNIQUE (event_id, room_id) ON CONFLICT REPLACE +); + +CREATE INDEX IF NOT EXISTS ev_b_extrem_room ON event_backward_extremities(room_id); +CREATE INDEX IF NOT EXISTS ev_b_extrem_id ON event_backward_extremities(event_id); +-- + +CREATE TABLE IF NOT EXISTS event_edges( + event_id TEXT, + prev_event_id TEXT, + room_id TEXT, + CONSTRAINT uniqueness UNIQUE (event_id, prev_event_id, room_id) +); + +CREATE INDEX IF NOT EXISTS ev_edges_id ON event_edges(event_id); +CREATE INDEX IF NOT EXISTS ev_edges_prev_id ON event_edges(prev_event_id); +-- + + +CREATE TABLE IF NOT EXISTS room_depth( + room_id TEXT, + min_depth INTEGER, + CONSTRAINT uniqueness UNIQUE (room_id) +); + +CREATE INDEX IF NOT EXISTS room_depth_room ON room_depth(room_id); +-- + +create TABLE IF NOT EXISTS event_destinations( + event_id TEXT, + destination TEXT, + delivered_ts INTEGER DEFAULT 0, -- or 0 if not delivered + CONSTRAINT uniqueness UNIQUE (event_id, destination) ON CONFLICT REPLACE +); + +CREATE INDEX IF NOT EXISTS event_destinations_id ON event_destinations(event_id); +-- \ No newline at end of file diff --git a/synapse/storage/schema/event_signatures.sql b/synapse/storage/schema/event_signatures.sql new file mode 100644 index 0000000000..5491c7ecec --- /dev/null +++ b/synapse/storage/schema/event_signatures.sql @@ -0,0 +1,65 @@ +/* Copyright 2014 OpenMarket Ltd + * + * 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. + */ + +CREATE TABLE IF NOT EXISTS event_content_hashes ( + event_id TEXT, + algorithm TEXT, + hash BLOB, + CONSTRAINT uniqueness UNIQUE (event_id, algorithm) +); + +CREATE INDEX IF NOT EXISTS event_content_hashes_id ON event_content_hashes( + event_id +); + + +CREATE TABLE IF NOT EXISTS event_reference_hashes ( + event_id TEXT, + algorithm TEXT, + hash BLOB, + CONSTRAINT uniqueness UNIQUE (event_id, algorithm) +); + +CREATE INDEX IF NOT EXISTS event_reference_hashes_id ON event_reference_hashes ( + event_id +); + + +CREATE TABLE IF NOT EXISTS event_origin_signatures ( + event_id TEXT, + origin TEXT, + key_id TEXT, + signature BLOB, + CONSTRAINT uniqueness UNIQUE (event_id, key_id) +); + +CREATE INDEX IF NOT EXISTS event_origin_signatures_id ON event_origin_signatures ( + event_id +); + + +CREATE TABLE IF NOT EXISTS event_edge_hashes( + event_id TEXT, + prev_event_id TEXT, + algorithm TEXT, + hash BLOB, + CONSTRAINT uniqueness UNIQUE ( + event_id, prev_event_id, algorithm + ) +); + +CREATE INDEX IF NOT EXISTS event_edge_hashes_id ON event_edge_hashes( + event_id +); diff --git a/synapse/storage/schema/im.sql b/synapse/storage/schema/im.sql index 3aa83f5c8c..8d6f655993 100644 --- a/synapse/storage/schema/im.sql +++ b/synapse/storage/schema/im.sql @@ -23,6 +23,7 @@ CREATE TABLE IF NOT EXISTS events( unrecognized_keys TEXT, processed BOOL NOT NULL, outlier BOOL NOT NULL, + depth INTEGER DEFAULT 0 NOT NULL, CONSTRAINT ev_uniq UNIQUE (event_id) ); diff --git a/synapse/storage/signatures.py b/synapse/storage/signatures.py index 82be946d3f..b8f8fd44cb 100644 --- a/synapse/storage/signatures.py +++ b/synapse/storage/signatures.py @@ -153,3 +153,130 @@ class SignatureStore(SQLBaseStore): "algorithm": algorithm, "hash": buffer(hash_bytes), }) + + ## Events ## + + def _get_event_content_hashes_txn(self, txn, event_id): + """Get all the hashes for a given Event. + Args: + txn (cursor): + event_id (str): Id for the Event. + Returns: + A dict of algorithm -> hash. + """ + query = ( + "SELECT algorithm, hash" + " FROM event_content_hashes" + " WHERE event_id = ?" + ) + txn.execute(query, (event_id, )) + return dict(txn.fetchall()) + + def _store_event_content_hash_txn(self, txn, event_id, algorithm, + hash_bytes): + """Store a hash for a Event + Args: + txn (cursor): + event_id (str): Id for the Event. + algorithm (str): Hashing algorithm. + hash_bytes (bytes): Hash function output bytes. + """ + self._simple_insert_txn(txn, "event_content_hashes", { + "event_id": event_id, + "algorithm": algorithm, + "hash": buffer(hash_bytes), + }) + + def _get_event_reference_hashes_txn(self, txn, event_id): + """Get all the hashes for a given PDU. + Args: + txn (cursor): + event_id (str): Id for the Event. + Returns: + A dict of algorithm -> hash. + """ + query = ( + "SELECT algorithm, hash" + " FROM event_reference_hashes" + " WHERE event_id = ?" + ) + txn.execute(query, (event_id, )) + return dict(txn.fetchall()) + + def _store_event_reference_hash_txn(self, txn, event_id, algorithm, + hash_bytes): + """Store a hash for a PDU + Args: + txn (cursor): + event_id (str): Id for the Event. + algorithm (str): Hashing algorithm. + hash_bytes (bytes): Hash function output bytes. + """ + self._simple_insert_txn(txn, "event_reference_hashes", { + "event_id": event_id, + "algorithm": algorithm, + "hash": buffer(hash_bytes), + }) + + + def _get_event_origin_signatures_txn(self, txn, event_id): + """Get all the signatures for a given PDU. + Args: + txn (cursor): + event_id (str): Id for the Event. + Returns: + A dict of key_id -> signature_bytes. + """ + query = ( + "SELECT key_id, signature" + " FROM event_origin_signatures" + " WHERE event_id = ? " + ) + txn.execute(query, (event_id, )) + return dict(txn.fetchall()) + + def _store_event_origin_signature_txn(self, txn, event_id, origin, key_id, + signature_bytes): + """Store a signature from the origin server for a PDU. + Args: + txn (cursor): + event_id (str): Id for the Event. + origin (str): origin of the Event. + key_id (str): Id for the signing key. + signature (bytes): The signature. + """ + self._simple_insert_txn(txn, "event_origin_signatures", { + "event_id": event_id, + "origin": origin, + "key_id": key_id, + "signature": buffer(signature_bytes), + }) + + def _get_prev_event_hashes_txn(self, txn, event_id): + """Get all the hashes for previous PDUs of a PDU + Args: + txn (cursor): + event_id (str): Id for the Event. + Returns: + dict of (pdu_id, origin) -> dict of algorithm -> hash_bytes. + """ + query = ( + "SELECT prev_event_id, algorithm, hash" + " FROM event_edge_hashes" + " WHERE event_id = ?" + ) + txn.execute(query, (event_id, )) + results = {} + for prev_event_id, algorithm, hash_bytes in txn.fetchall(): + hashes = results.setdefault(prev_event_id, {}) + hashes[algorithm] = hash_bytes + return results + + def _store_prev_event_hash_txn(self, txn, event_id, prev_event_id, + algorithm, hash_bytes): + self._simple_insert_txn(txn, "event_edge_hashes", { + "event_id": event_id, + "prev_event_id": prev_event_id, + "algorithm": algorithm, + "hash": buffer(hash_bytes), + }) \ No newline at end of file -- cgit 1.5.1 From e7858b6d7ef37849a3d2d5004743cdd21ec330a8 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Wed, 29 Oct 2014 16:59:24 +0000 Subject: Start filling out and using new events tables --- synapse/federation/pdu_codec.py | 12 +++-- synapse/handlers/_base.py | 4 ++ synapse/handlers/federation.py | 90 +++++++++++++++++++--------------- synapse/state.py | 11 +++-- synapse/storage/__init__.py | 45 ++++++++++------- synapse/storage/_base.py | 33 ++++++++++--- synapse/storage/event_federation.py | 49 ++++++++++++------ synapse/storage/schema/event_edges.sql | 8 ++- 8 files changed, 159 insertions(+), 93 deletions(-) (limited to 'synapse/storage/_base.py') diff --git a/synapse/federation/pdu_codec.py b/synapse/federation/pdu_codec.py index 2cd591410b..dccbccb85b 100644 --- a/synapse/federation/pdu_codec.py +++ b/synapse/federation/pdu_codec.py @@ -48,8 +48,8 @@ class PduCodec(object): kwargs["room_id"] = pdu.context kwargs["etype"] = pdu.pdu_type kwargs["prev_events"] = [ - encode_event_id(i, o) - for i, o in pdu.prev_pdus + (encode_event_id(i, o), s) + for i, o, s in pdu.prev_pdus ] if hasattr(pdu, "prev_state_id") and hasattr(pdu, "prev_state_origin"): @@ -82,7 +82,13 @@ class PduCodec(object): d["pdu_type"] = event.type if hasattr(event, "prev_events"): - d["prev_pdus"] = [decode_event_id(e) for e in event.prev_events] + def f(e, s): + i, o = decode_event_id(e, self.server_name) + return i, o, s + d["prev_pdus"] = [ + f(e, s) + for e, s in event.prev_events + ] if hasattr(event, "prev_state"): d["prev_state_id"], d["prev_state_origin"] = ( diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py index cd6c35f194..787a01efc5 100644 --- a/synapse/handlers/_base.py +++ b/synapse/handlers/_base.py @@ -16,6 +16,8 @@ from twisted.internet import defer from synapse.api.errors import LimitExceededError +from synapse.util.async import run_on_reactor + class BaseHandler(object): def __init__(self, hs): @@ -45,6 +47,8 @@ class BaseHandler(object): @defer.inlineCallbacks def _on_new_room_event(self, event, snapshot, extra_destinations=[], extra_users=[], suppress_auth=False): + yield run_on_reactor() + snapshot.fill_out_prev_events(event) yield self.state_handler.annotate_state_groups(event) diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index b575986fc3..5f86ed03fa 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -22,6 +22,7 @@ from synapse.api.constants import Membership from synapse.util.logutils import log_function from synapse.federation.pdu_codec import PduCodec, encode_event_id from synapse.api.errors import SynapseError +from synapse.util.async import run_on_reactor from twisted.internet import defer, reactor @@ -81,6 +82,8 @@ class FederationHandler(BaseHandler): processing. """ + yield run_on_reactor() + pdu = self.pdu_codec.pdu_from_event(event) if not hasattr(pdu, "destinations") or not pdu.destinations: @@ -102,6 +105,8 @@ class FederationHandler(BaseHandler): self.room_queues[event.room_id].append(pdu) return + logger.debug("Processing event: %s", event.event_id) + if state: state = [self.pdu_codec.event_from_pdu(p) for p in state] @@ -216,58 +221,65 @@ class FederationHandler(BaseHandler): assert(event.state_key == joinee) assert(event.room_id == room_id) - self.room_queues[room_id] = [] - - event.event_id = self.event_factory.create_event_id() - event.content = content + event.outlier = False - state = yield self.replication_layer.send_join( - target_host, - self.pdu_codec.pdu_from_event(event) - ) + self.room_queues[room_id] = [] - state = [self.pdu_codec.event_from_pdu(p) for p in state] + try: + event.event_id = self.event_factory.create_event_id() + event.content = content - logger.debug("do_invite_join state: %s", state) + state = yield self.replication_layer.send_join( + target_host, + self.pdu_codec.pdu_from_event(event) + ) - is_new_state = yield self.state_handler.annotate_state_groups( - event, - state=state - ) + state = [self.pdu_codec.event_from_pdu(p) for p in state] - try: - yield self.store.store_room( - room_id=room_id, - room_creator_user_id="", - is_public=False - ) - except: - # FIXME - pass + logger.debug("do_invite_join state: %s", state) - for e in state: - # FIXME: Auth these. is_new_state = yield self.state_handler.annotate_state_groups( - e, + event, + state=state ) + logger.debug("do_invite_join event: %s", event) + + try: + yield self.store.store_room( + room_id=room_id, + room_creator_user_id="", + is_public=False + ) + except: + # FIXME + pass + + for e in state: + # FIXME: Auth these. + e.outlier = True + + yield self.state_handler.annotate_state_groups( + e, + ) + + yield self.store.persist_event( + e, + backfilled=False, + is_new_state=False + ) + yield self.store.persist_event( - e, + event, backfilled=False, - is_new_state=False + is_new_state=is_new_state ) + finally: + room_queue = self.room_queues[room_id] + del self.room_queues[room_id] - yield self.store.persist_event( - event, - backfilled=False, - is_new_state=is_new_state - ) - - room_queue = self.room_queues[room_id] - del self.room_queues[room_id] - - for p in room_queue: - yield self.on_receive_pdu(p, backfilled=False) + for p in room_queue: + yield self.on_receive_pdu(p, backfilled=False) defer.returnValue(True) diff --git a/synapse/state.py b/synapse/state.py index cc6a7db96b..993c4f18d3 100644 --- a/synapse/state.py +++ b/synapse/state.py @@ -143,7 +143,9 @@ class StateHandler(object): defer.returnValue(False) return - new_state = yield self.resolve_state_groups(event.prev_events) + new_state = yield self.resolve_state_groups( + [e for e, _ in event.prev_events] + ) event.old_state_events = copy.deepcopy(new_state) @@ -157,12 +159,11 @@ class StateHandler(object): @defer.inlineCallbacks def get_current_state(self, room_id, event_type=None, state_key=""): - # FIXME: HACK! - pdus = yield self.store.get_latest_pdus_in_context(room_id) + events = yield self.store.get_latest_events_in_room(room_id) event_ids = [ - encode_event_id(pdu_id, origin) - for pdu_id, origin, _ in pdus + e_id + for e_id, _ in events ] res = yield self.resolve_state_groups(event_ids) diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index f89e518690..d75c366834 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -71,6 +71,7 @@ SCHEMAS = [ "state", "signatures", "event_edges", + "event_signatures", ] @@ -134,7 +135,8 @@ class DataStore(RoomMemberStore, RoomStore, "type", "room_id", "content", - "unrecognized_keys" + "unrecognized_keys", + "depth", ], allow_none=allow_none, ) @@ -263,7 +265,12 @@ class DataStore(RoomMemberStore, RoomStore, vals["unrecognized_keys"] = json.dumps(unrec) try: - self._simple_insert_txn(txn, "events", vals) + self._simple_insert_txn( + txn, + "events", + vals, + or_replace=(not outlier), + ) except: logger.warn( "Failed to persist, probably duplicate: %s", @@ -307,13 +314,14 @@ class DataStore(RoomMemberStore, RoomStore, } ) - signatures = event.signatures.get(event.origin, {}) + if hasattr(event, "signatures"): + signatures = event.signatures.get(event.origin, {}) - for key_id, signature_base64 in signatures.items(): - signature_bytes = decode_base64(signature_base64) - self._store_event_origin_signature_txn( - txn, event.event_id, key_id, signature_bytes, - ) + for key_id, signature_base64 in signatures.items(): + signature_bytes = decode_base64(signature_base64) + self._store_event_origin_signature_txn( + txn, event.event_id, event.origin, key_id, signature_bytes, + ) for prev_event_id, prev_hashes in event.prev_events: for alg, hash_base64 in prev_hashes.items(): @@ -323,10 +331,10 @@ class DataStore(RoomMemberStore, RoomStore, ) # TODO - (ref_alg, ref_hash_bytes) = compute_pdu_event_reference_hash(pdu) - self._store_event_reference_hash_txn( - txn, event.event_id, ref_alg, ref_hash_bytes - ) + # (ref_alg, ref_hash_bytes) = compute_pdu_event_reference_hash(pdu) + # self._store_event_reference_hash_txn( + # txn, event.event_id, ref_alg, ref_hash_bytes + # ) self._update_min_depth_for_room_txn(txn, event.room_id, event.depth) @@ -412,9 +420,7 @@ class DataStore(RoomMemberStore, RoomStore, """ def _snapshot(txn): membership_state = self._get_room_member(txn, user_id, room_id) - prev_events = self._get_latest_events_in_room( - txn, room_id - ) + prev_events = self._get_latest_events_in_room(txn, room_id) if state_type is not None and state_key is not None: prev_state_pdu = self._get_current_state_pdu( @@ -469,12 +475,12 @@ class Snapshot(object): return event.prev_events = [ - (p_id, origin, hashes) - for p_id, origin, hashes, _ in self.prev_events + (event_id, hashes) + for event_id, hashes, _ in self.prev_events ] if self.prev_events: - event.depth = max([int(v) for _, _, _, v in self.prev_events]) + 1 + event.depth = max([int(v) for _, _, v in self.prev_events]) + 1 else: event.depth = 0 @@ -533,9 +539,10 @@ def prepare_database(db_conn): db_conn.commit() else: - sql_script = "BEGIN TRANSACTION;" + sql_script = "BEGIN TRANSACTION;\n" for sql_loc in SCHEMAS: sql_script += read_schema(sql_loc) + sql_script += "\n" sql_script += "COMMIT TRANSACTION;" c.executescript(sql_script) db_conn.commit() diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 30732caa83..464b12f032 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -19,10 +19,12 @@ from twisted.internet import defer from synapse.api.errors import StoreError from synapse.api.events.utils import prune_event from synapse.util.logutils import log_function +from syutil.base64util import encode_base64 import collections import copy import json +import sys import time @@ -67,6 +69,9 @@ class LoggingTransaction(object): return self.txn.execute( sql, *args, **kwargs ) + except: + logger.exception("[SQL FAIL] {%s}", self.name) + raise finally: end = time.clock() * 1000 sql_logger.debug("[SQL time] {%s} %f", self.name, end - start) @@ -85,14 +90,20 @@ class SQLBaseStore(object): """Wraps the .runInteraction() method on the underlying db_pool.""" def inner_func(txn, *args, **kwargs): start = time.clock() * 1000 - txn_id = str(SQLBaseStore._TXN_ID) - SQLBaseStore._TXN_ID += 1 + txn_id = SQLBaseStore._TXN_ID + + # We don't really need these to be unique, so lets stop it from + # growing really large. + self._TXN_ID = (self._TXN_ID + 1) % (sys.maxint - 1) - name = "%s-%s" % (desc, txn_id, ) + name = "%s-%x" % (desc, txn_id, ) transaction_logger.debug("[TXN START] {%s}", name) try: return func(LoggingTransaction(txn, name), *args, **kwargs) + except: + logger.exception("[TXN FAIL] {%s}", name) + raise finally: end = time.clock() * 1000 transaction_logger.debug( @@ -189,7 +200,6 @@ class SQLBaseStore(object): statement returns no rows """ return self._simple_selectupdate_one( - "_simple_select_one", table, keyvalues, retcols=retcols, allow_none=allow_none ) @@ -215,11 +225,11 @@ class SQLBaseStore(object): txn, table=table, keyvalues=keyvalues, - retcols=retcol, + retcol=retcol, ) if ret: - return ret[retcol] + return ret[0] else: if allow_none: return None @@ -434,6 +444,17 @@ class SQLBaseStore(object): sql = "SELECT * FROM events WHERE event_id = ?" for ev in events: + signatures = self._get_event_origin_signatures_txn( + txn, ev.event_id, + ) + + ev.signatures = { + k: encode_base64(v) for k, v in signatures.items() + } + + prev_events = self._get_latest_events_in_room(txn, ev.room_id) + ev.prev_events = [(e_id, s,) for e_id, s, _ in prev_events] + if hasattr(ev, "prev_state"): # Load previous state_content. # TODO: Should we be pulling this out above? diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py index 7688fc550f..5f94c31818 100644 --- a/synapse/storage/event_federation.py +++ b/synapse/storage/event_federation.py @@ -24,6 +24,13 @@ logger = logging.getLogger(__name__) class EventFederationStore(SQLBaseStore): + def get_latest_events_in_room(self, room_id): + return self.runInteraction( + "get_latest_events_in_room", + self._get_latest_events_in_room, + room_id, + ) + def _get_latest_events_in_room(self, txn, room_id): self._simple_select_onecol_txn( txn, @@ -34,12 +41,25 @@ class EventFederationStore(SQLBaseStore): retcol="event_id", ) + sql = ( + "SELECT e.event_id, e.depth FROM events as e " + "INNER JOIN event_forward_extremities as f " + "ON e.event_id = f.event_id " + "WHERE f.room_id = ?" + ) + + txn.execute(sql, (room_id, )) + results = [] - for pdu_id, origin, depth in txn.fetchall(): - hashes = self._get_prev_event_hashes_txn(txn, pdu_id, origin) - sha256_bytes = hashes["sha256"] - prev_hashes = {"sha256": encode_base64(sha256_bytes)} - results.append((pdu_id, origin, prev_hashes, depth)) + for event_id, depth in txn.fetchall(): + hashes = self._get_prev_event_hashes_txn(txn, event_id) + prev_hashes = { + k: encode_base64(v) for k, v in hashes.items() + if k == "sha256" + } + results.append((event_id, prev_hashes, depth)) + + return results def _get_min_depth_interaction(self, txn, room_id): min_depth = self._simple_select_one_onecol_txn( @@ -70,21 +90,21 @@ class EventFederationStore(SQLBaseStore): def _handle_prev_events(self, txn, outlier, event_id, prev_events, room_id): - for e_id in prev_events: + for e_id, _ in prev_events: # TODO (erikj): This could be done as a bulk insert self._simple_insert_txn( txn, table="event_edges", values={ "event_id": event_id, - "prev_event": e_id, + "prev_event_id": e_id, "room_id": room_id, } ) # Update the extremities table if this is not an outlier. if not outlier: - for e_id in prev_events: + for e_id, _ in prev_events: # TODO (erikj): This could be done as a bulk insert self._simple_delete_txn( txn, @@ -116,7 +136,7 @@ class EventFederationStore(SQLBaseStore): # Insert all the prev_pdus as a backwards thing, they'll get # deleted in a second if they're incorrect anyway. - for e_id in prev_events: + for e_id, _ in prev_events: # TODO (erikj): This could be done as a bulk insert self._simple_insert_txn( txn, @@ -130,14 +150,11 @@ class EventFederationStore(SQLBaseStore): # Also delete from the backwards extremities table all ones that # reference pdus that we have already seen query = ( - "DELETE FROM %(event_back)s as b WHERE EXISTS (" - "SELECT 1 FROM %(events)s AS events " + "DELETE FROM event_backward_extremities WHERE EXISTS (" + "SELECT 1 FROM events " "WHERE " - "b.event_id = events.event_id " + "event_backward_extremities.event_id = events.event_id " "AND not events.outlier " ")" - ) % { - "event_back": "event_backward_extremities", - "events": "events", - } + ) txn.execute(query) \ No newline at end of file diff --git a/synapse/storage/schema/event_edges.sql b/synapse/storage/schema/event_edges.sql index 6a28314ece..e5f768c705 100644 --- a/synapse/storage/schema/event_edges.sql +++ b/synapse/storage/schema/event_edges.sql @@ -7,7 +7,7 @@ CREATE TABLE IF NOT EXISTS event_forward_extremities( CREATE INDEX IF NOT EXISTS ev_extrem_room ON event_forward_extremities(room_id); CREATE INDEX IF NOT EXISTS ev_extrem_id ON event_forward_extremities(event_id); --- + CREATE TABLE IF NOT EXISTS event_backward_extremities( event_id TEXT, @@ -17,7 +17,7 @@ CREATE TABLE IF NOT EXISTS event_backward_extremities( CREATE INDEX IF NOT EXISTS ev_b_extrem_room ON event_backward_extremities(room_id); CREATE INDEX IF NOT EXISTS ev_b_extrem_id ON event_backward_extremities(event_id); --- + CREATE TABLE IF NOT EXISTS event_edges( event_id TEXT, @@ -28,7 +28,6 @@ CREATE TABLE IF NOT EXISTS event_edges( CREATE INDEX IF NOT EXISTS ev_edges_id ON event_edges(event_id); CREATE INDEX IF NOT EXISTS ev_edges_prev_id ON event_edges(prev_event_id); --- CREATE TABLE IF NOT EXISTS room_depth( @@ -38,7 +37,7 @@ CREATE TABLE IF NOT EXISTS room_depth( ); CREATE INDEX IF NOT EXISTS room_depth_room ON room_depth(room_id); --- + create TABLE IF NOT EXISTS event_destinations( event_id TEXT, @@ -48,4 +47,3 @@ create TABLE IF NOT EXISTS event_destinations( ); CREATE INDEX IF NOT EXISTS event_destinations_id ON event_destinations(event_id); --- \ No newline at end of file -- cgit 1.5.1 From b29517bd013b82302b1a73072da8bfc39564dc1a Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Thu, 30 Oct 2014 01:21:33 +0000 Subject: Add a request-id to each log line --- setup.py | 2 +- synapse/app/homeserver.py | 12 +++++- synapse/config/logger.py | 23 ++++++++---- synapse/crypto/keyclient.py | 10 +++-- synapse/http/client.py | 26 +++++++------ synapse/http/server.py | 13 ++++++- synapse/storage/_base.py | 16 ++++++-- synapse/util/async.py | 5 ++- synapse/util/logcontext.py | 85 ++++++++++++++++++++++++++++++++++++++++++ synapse/util/logutils.py | 1 + tests/util/test_log_context.py | 43 +++++++++++++++++++++ 11 files changed, 205 insertions(+), 31 deletions(-) create mode 100644 synapse/util/logcontext.py create mode 100644 tests/util/test_log_context.py (limited to 'synapse/storage/_base.py') diff --git a/setup.py b/setup.py index 660efd5b89..74eee31a78 100755 --- a/setup.py +++ b/setup.py @@ -54,6 +54,6 @@ setup( long_description=read("README.rst"), entry_points=""" [console_scripts] - synapse-homeserver=synapse.app.homeserver:run + synapse-homeserver=synapse.app.homeserver:main """ ) diff --git a/synapse/app/homeserver.py b/synapse/app/homeserver.py index 6394bc27d1..4e74f4d14c 100755 --- a/synapse/app/homeserver.py +++ b/synapse/app/homeserver.py @@ -33,6 +33,7 @@ from synapse.api.urls import ( ) from synapse.config.homeserver import HomeServerConfig from synapse.crypto import context_factory +from synapse.util.logcontext import LoggingContext from daemonize import Daemonize import twisted.manhole.telnet @@ -240,7 +241,7 @@ def setup(): daemon = Daemonize( app="synapse-homeserver", pid=config.pid_file, - action=reactor.run, + action=run, auto_close_fds=False, verbose=True, logger=logger, @@ -250,6 +251,13 @@ def setup(): else: reactor.run() +def run(): + with LoggingContext("run") as context: + reactor.run() + +def main(): + with LoggingContext("main") as context: + setup() if __name__ == '__main__': - setup() + main() diff --git a/synapse/config/logger.py b/synapse/config/logger.py index 56cd095433..2a59bf9d15 100644 --- a/synapse/config/logger.py +++ b/synapse/config/logger.py @@ -14,7 +14,7 @@ # limitations under the License. from ._base import Config - +from synapse.util.logcontext import LoggingContextFilter from twisted.python.log import PythonLoggingObserver import logging import logging.config @@ -45,7 +45,8 @@ class LoggingConfig(Config): def setup_logging(self): log_format = ( - '%(asctime)s - %(name)s - %(lineno)d - %(levelname)s - %(message)s' + "%(asctime)s - %(name)s - %(lineno)d - %(levelname)s - %(request)s" + " - %(message)s" ) if self.log_config is None: @@ -54,12 +55,20 @@ class LoggingConfig(Config): level = logging.DEBUG # FIXME: we need a logging.WARN for a -q quiet option + logger = logging.getLogger('') + logger.setLevel(level) + formatter = logging.Formatter(log_format) + if self.log_file: + handler = logging.FileHandler(self.log_file) + else: + handler = logging.StreamHandler() + print handler + handler.setFormatter(formatter) + + handler.addFilter(LoggingContextFilter(request="")) - logging.basicConfig( - level=level, - filename=self.log_file, - format=log_format - ) + logger.addHandler(handler) + logger.info("Test") else: logging.config.fileConfig(self.log_config) diff --git a/synapse/crypto/keyclient.py b/synapse/crypto/keyclient.py index 7cfec5148e..33fa9ca837 100644 --- a/synapse/crypto/keyclient.py +++ b/synapse/crypto/keyclient.py @@ -18,6 +18,7 @@ from twisted.web.http import HTTPClient from twisted.internet.protocol import Factory from twisted.internet import defer, reactor from synapse.http.endpoint import matrix_endpoint +from synapse.util.logcontext import PreserveLoggingContext import json import logging @@ -36,10 +37,11 @@ def fetch_server_key(server_name, ssl_context_factory): for i in range(5): try: - protocol = yield endpoint.connect(factory) - server_response, server_certificate = yield protocol.remote_key - defer.returnValue((server_response, server_certificate)) - return + with PreserveLoggingContext(): + protocol = yield endpoint.connect(factory) + server_response, server_certificate = yield protocol.remote_key + defer.returnValue((server_response, server_certificate)) + return except Exception as e: logger.exception(e) raise IOError("Cannot get key for %s" % server_name) diff --git a/synapse/http/client.py b/synapse/http/client.py index 46c90dbb76..8bda42364b 100644 --- a/synapse/http/client.py +++ b/synapse/http/client.py @@ -16,11 +16,14 @@ from twisted.internet import defer, reactor from twisted.internet.error import DNSLookupError -from twisted.web.client import _AgentBase, _URI, readBody, FileBodyProducer, PartialDownloadError +from twisted.web.client import ( + _AgentBase, _URI, readBody, FileBodyProducer, PartialDownloadError +) from twisted.web.http_headers import Headers from synapse.http.endpoint import matrix_endpoint from synapse.util.async import sleep +from synapse.util.logcontext import PreserveLoggingContext from syutil.jsonutil import encode_canonical_json @@ -106,16 +109,17 @@ class BaseHttpClient(object): producer = body_callback(method, url_bytes, headers_dict) try: - response = yield self.agent.request( - destination, - endpoint, - method, - path_bytes, - param_bytes, - query_bytes, - Headers(headers_dict), - producer - ) + with PreserveLoggingContext(): + response = yield self.agent.request( + destination, + endpoint, + method, + path_bytes, + param_bytes, + query_bytes, + Headers(headers_dict), + producer + ) logger.debug("Got response to %s", method) break diff --git a/synapse/http/server.py b/synapse/http/server.py index 8d419c02dd..ed1f1170cb 100644 --- a/synapse/http/server.py +++ b/synapse/http/server.py @@ -20,6 +20,7 @@ from syutil.jsonutil import ( from synapse.api.errors import ( cs_exception, SynapseError, CodeMessageException ) +from synapse.util.logcontext import LoggingContext from twisted.internet import defer, reactor from twisted.web import server, resource @@ -88,9 +89,19 @@ class JsonResource(HttpServer, resource.Resource): def render(self, request): """ This get's called by twisted every time someone sends us a request. """ - self._async_render(request) + self._async_render_with_logging_context(request) return server.NOT_DONE_YET + _request_id = 0 + + @defer.inlineCallbacks + def _async_render_with_logging_context(self, request): + request_id = "%s-%s" % (request.method, JsonResource._request_id) + JsonResource._request_id += 1 + with LoggingContext(request_id) as request_context: + request_context.request = request_id + yield self._async_render(request) + @defer.inlineCallbacks def _async_render(self, request): """ This get's called by twisted every time someone sends us a request. diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 65a86e9056..2faa63904e 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -19,6 +19,7 @@ from twisted.internet import defer from synapse.api.errors import StoreError from synapse.api.events.utils import prune_event from synapse.util.logutils import log_function +from synapse.util.logcontext import PreserveLoggingContext, LoggingContext import collections import copy @@ -74,12 +75,19 @@ class SQLBaseStore(object): self.event_factory = hs.get_event_factory() self._clock = hs.get_clock() + @defer.inlineCallbacks def runInteraction(self, func, *args, **kwargs): """Wraps the .runInteraction() method on the underlying db_pool.""" + current_context = LoggingContext.current_context() def inner_func(txn, *args, **kwargs): - return func(LoggingTransaction(txn), *args, **kwargs) - - return self._db_pool.runInteraction(inner_func, *args, **kwargs) + with LoggingContext("runInteraction") as context: + current_context.copy_to(context) + return func(LoggingTransaction(txn), *args, **kwargs) + with PreserveLoggingContext(): + result = yield self._db_pool.runInteraction( + inner_func, *args, **kwargs + ) + defer.returnValue(result) def cursor_to_dict(self, cursor): """Converts a SQL cursor into an list of dicts. @@ -146,7 +154,7 @@ class SQLBaseStore(object): ) logger.debug( - "[SQL] %s Args=%s Func=%s", + "[SQL] %s Args=%s", sql, values.values(), ) diff --git a/synapse/util/async.py b/synapse/util/async.py index 647ea6142c..3d3fbe182c 100644 --- a/synapse/util/async.py +++ b/synapse/util/async.py @@ -16,8 +16,11 @@ from twisted.internet import defer, reactor +from .logcontext import PreserveLoggingContext +@defer.inlineCallbacks def sleep(seconds): d = defer.Deferred() reactor.callLater(seconds, d.callback, seconds) - return d + with PreserveLoggingContext(): + yield d diff --git a/synapse/util/logcontext.py b/synapse/util/logcontext.py new file mode 100644 index 0000000000..46a2855a15 --- /dev/null +++ b/synapse/util/logcontext.py @@ -0,0 +1,85 @@ +from functools import wraps + +import threading +import logging + +class LoggingContext(object): + __slots__ = ["parent_context", "name", "__dict__"] + + thread_local = threading.local() + + class Sentinel(object): + __slots__ = [] + def copy_to(self, record): + pass + + sentinel = Sentinel() + + def __init__(self, name=None): + self.parent_context = None + self.name = name + + def __str__(self): + return "%s@%x" % (self.name, id(self)) + + @classmethod + def current_context(cls): + return getattr(cls.thread_local, "current_context", cls.sentinel) + + def __enter__(self): + if self.parent_context is not None: + raise Exception("Attempt to enter logging context multiple times") + self.parent_context = self.current_context() + self.thread_local.current_context = self + return self + + def __exit__(self, type, value, traceback): + if self.thread_local.current_context is not self: + logging.error( + "Current logging context %s is not the expected context %s", + self.thread_local.current_context, + self + ) + self.thread_local.current_context = self.parent_context + self.parent_context = None + + def __getattr__(self, name): + return getattr(self.parent_context, name) + + def copy_to(self, record): + if self.parent_context is not None: + self.parent_context.copy_to(record) + for key, value in self.__dict__.items(): + setattr(record, key, value) + + @classmethod + def wrap_callback(cls, callback): + context = cls.current_context() + @wraps(callback) + def wrapped(*args, **kargs): + cls.thread_local.current_context = context + return callback(*args, **kargs) + return wrapped + + +class LoggingContextFilter(logging.Filter): + def __init__(self, **defaults): + self.defaults = defaults + + def filter(self, record): + context = LoggingContext.current_context() + for key, value in self.defaults.items(): + setattr(record, key, value) + context.copy_to(record) + return True + + +class PreserveLoggingContext(object): + __slots__ = ["current_context"] + def __enter__(self): + self.current_context = LoggingContext.current_context() + + def __exit__(self, type, value, traceback): + LoggingContext.thread_local.current_context = self.current_context + + diff --git a/synapse/util/logutils.py b/synapse/util/logutils.py index fadf0bd510..903a6cf1b3 100644 --- a/synapse/util/logutils.py +++ b/synapse/util/logutils.py @@ -75,6 +75,7 @@ def trace_function(f): linenum = f.func_code.co_firstlineno pathname = f.func_code.co_filename + @wraps(f) def wrapped(*args, **kwargs): name = f.__module__ logger = logging.getLogger(name) diff --git a/tests/util/test_log_context.py b/tests/util/test_log_context.py new file mode 100644 index 0000000000..efa0f28bad --- /dev/null +++ b/tests/util/test_log_context.py @@ -0,0 +1,43 @@ +from twisted.internet import defer +from twisted.internet import reactor +from .. import unittest + +from synapse.util.async import sleep +from synapse.util.logcontext import LoggingContext + +class LoggingContextTestCase(unittest.TestCase): + + def _check_test_key(self, value): + self.assertEquals( + LoggingContext.current_context().test_key, value + ) + + def test_with_context(self): + with LoggingContext() as context_one: + context_one.test_key = "test" + self._check_test_key("test") + + def test_chaining(self): + with LoggingContext() as context_one: + context_one.test_key = "one" + with LoggingContext() as context_two: + self._check_test_key("one") + context_two.test_key = "two" + self._check_test_key("two") + self._check_test_key("one") + + @defer.inlineCallbacks + def test_sleep(self): + @defer.inlineCallbacks + def competing_callback(): + with LoggingContext() as competing_context: + competing_context.test_key = "competing" + yield sleep(0) + self._check_test_key("competing") + + reactor.callLater(0, competing_callback) + + with LoggingContext() as context_one: + context_one.test_key = "one" + yield sleep(0) + self._check_test_key("one") -- cgit 1.5.1 From aa76bf39aba2eadf506f57952a1dffce629f2637 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 4 Nov 2014 14:14:02 +0000 Subject: Remove unused imports --- synapse/federation/persistence.py | 2 -- synapse/federation/units.py | 3 --- synapse/handlers/federation.py | 6 ++---- synapse/state.py | 2 -- synapse/storage/_base.py | 2 -- 5 files changed, 2 insertions(+), 13 deletions(-) (limited to 'synapse/storage/_base.py') diff --git a/synapse/federation/persistence.py b/synapse/federation/persistence.py index b04fbb4177..73dc844d59 100644 --- a/synapse/federation/persistence.py +++ b/synapse/federation/persistence.py @@ -21,8 +21,6 @@ These actions are mostly only used by the :py:mod:`.replication` module. from twisted.internet import defer -from .units import Pdu - from synapse.util.logutils import log_function import json diff --git a/synapse/federation/units.py b/synapse/federation/units.py index c2d8dca8f3..9b25556707 100644 --- a/synapse/federation/units.py +++ b/synapse/federation/units.py @@ -18,11 +18,8 @@ server protocol. """ from synapse.util.jsonobject import JsonEncodedObject -from syutil.base64util import encode_base64 import logging -import json -import copy logger = logging.getLogger(__name__) diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index bdd28f04bb..49bfff88a4 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -17,15 +17,13 @@ from ._base import BaseHandler -from synapse.api.events.room import InviteJoinEvent, RoomMemberEvent +from synapse.api.events.room import RoomMemberEvent from synapse.api.constants import Membership from synapse.util.logutils import log_function from synapse.federation.pdu_codec import PduCodec -from synapse.api.errors import SynapseError from synapse.util.async import run_on_reactor -from synapse.types import EventID -from twisted.internet import defer, reactor +from twisted.internet import defer import logging diff --git a/synapse/state.py b/synapse/state.py index f4efc287c9..9771883bc3 100644 --- a/synapse/state.py +++ b/synapse/state.py @@ -19,8 +19,6 @@ from twisted.internet import defer from synapse.util.logutils import log_function from synapse.util.async import run_on_reactor -from synapse.types import EventID - from collections import namedtuple import copy diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 464b12f032..5e00c23fd1 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -14,8 +14,6 @@ # limitations under the License. import logging -from twisted.internet import defer - from synapse.api.errors import StoreError from synapse.api.events.utils import prune_event from synapse.util.logutils import log_function -- cgit 1.5.1 From cc44ecc62f69436a9217745292af6c55b5f8fe81 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Wed, 5 Nov 2014 13:23:35 +0000 Subject: Get correct prev_events --- synapse/storage/_base.py | 11 +++++------ synapse/storage/event_federation.py | 30 +++++++++++++++++++++--------- 2 files changed, 26 insertions(+), 15 deletions(-) (limited to 'synapse/storage/_base.py') diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 5e00c23fd1..7d445b4633 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -439,7 +439,7 @@ class SQLBaseStore(object): def _parse_events_txn(self, txn, rows): events = [self._parse_event_from_row(r) for r in rows] - sql = "SELECT * FROM events WHERE event_id = ?" + select_event_sql = "SELECT * FROM events WHERE event_id = ?" for ev in events: signatures = self._get_event_origin_signatures_txn( @@ -450,13 +450,12 @@ class SQLBaseStore(object): k: encode_base64(v) for k, v in signatures.items() } - prev_events = self._get_latest_events_in_room(txn, ev.room_id) - ev.prev_events = [(e_id, s,) for e_id, s, _ in prev_events] + ev.prev_events = self._get_prev_events(txn, ev.event_id) if hasattr(ev, "prev_state"): # Load previous state_content. # TODO: Should we be pulling this out above? - cursor = txn.execute(sql, (ev.prev_state,)) + cursor = txn.execute(select_event_sql, (ev.prev_state,)) prevs = self.cursor_to_dict(cursor) if prevs: prev = self._parse_event_from_row(prevs[0]) @@ -468,8 +467,8 @@ class SQLBaseStore(object): if ev.redacted: # Get the redaction event. - sql = "SELECT * FROM events WHERE event_id = ?" - txn.execute(sql, (ev.redacted,)) + select_event_sql = "SELECT * FROM events WHERE event_id = ?" + txn.execute(select_event_sql, (ev.redacted,)) del_evs = self._parse_events_txn( txn, self.cursor_to_dict(txn) diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py index dcc116bad2..f427aba879 100644 --- a/synapse/storage/event_federation.py +++ b/synapse/storage/event_federation.py @@ -49,15 +49,6 @@ class EventFederationStore(SQLBaseStore): ) def _get_latest_events_in_room(self, txn, room_id): - self._simple_select_onecol_txn( - txn, - table="event_forward_extremities", - keyvalues={ - "room_id": room_id, - }, - retcol="event_id", - ) - sql = ( "SELECT e.event_id, e.depth FROM events as e " "INNER JOIN event_forward_extremities as f " @@ -78,6 +69,27 @@ class EventFederationStore(SQLBaseStore): return results + def _get_prev_events(self, txn, event_id): + prev_ids = self._simple_select_onecol_txn( + txn, + table="event_edges", + keyvalues={ + "event_id": event_id, + }, + retcol="prev_event_id", + ) + + results = [] + for prev_event_id in prev_ids: + hashes = self._get_event_reference_hashes_txn(txn, prev_event_id) + prev_hashes = { + k: encode_base64(v) for k, v in hashes.items() + if k == "sha256" + } + results.append((event_id, prev_hashes)) + + return results + def get_min_depth(self, room_id): return self.runInteraction( "get_min_depth", -- cgit 1.5.1 From 4317c8e5835f0c15bf882f737d3e3c2a5b85f73f Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 6 Nov 2014 15:10:55 +0000 Subject: Implement new replace_state and changed prev_state `prev_state` is now a list of previous state ids, similiar to prev_events. `replace_state` now points to what we think was replaced. --- synapse/api/events/__init__.py | 1 + synapse/handlers/directory.py | 5 +- synapse/handlers/federation.py | 4 +- synapse/handlers/message.py | 11 ++-- synapse/handlers/profile.py | 6 +-- synapse/handlers/room.py | 16 ++---- synapse/rest/room.py | 2 +- synapse/state.py | 39 ++------------ synapse/storage/__init__.py | 92 +++++++++++++++++++++++++--------- synapse/storage/_base.py | 66 +++++++++++++++++------- synapse/storage/event_federation.py | 64 ++++++++++++++++++++--- synapse/storage/schema/event_edges.sql | 40 ++++++++++----- synapse/util/jsonobject.py | 2 +- 13 files changed, 220 insertions(+), 128 deletions(-) (limited to 'synapse/storage/_base.py') diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py index 168b812311..fc3f350570 100644 --- a/synapse/api/events/__init__.py +++ b/synapse/api/events/__init__.py @@ -60,6 +60,7 @@ class SynapseEvent(JsonEncodedObject): "age_ts", "prev_content", "prev_state", + "replaces_state", "redacted_because", "origin_server_ts", ] diff --git a/synapse/handlers/directory.py b/synapse/handlers/directory.py index 6e897e915d..164363cdc5 100644 --- a/synapse/handlers/directory.py +++ b/synapse/handlers/directory.py @@ -147,10 +147,7 @@ class DirectoryHandler(BaseHandler): content={"aliases": aliases}, ) - snapshot = yield self.store.snapshot_room( - room_id=room_id, - user_id=user_id, - ) + snapshot = yield self.store.snapshot_room(event) yield self._on_new_room_event( event, snapshot, extra_users=[user_id], suppress_auth=True diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index 1464a60937..513ec9a5e3 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -313,9 +313,7 @@ class FederationHandler(BaseHandler): state_key=user_id, ) - snapshot = yield self.store.snapshot_room( - event.room_id, event.user_id, - ) + snapshot = yield self.store.snapshot_room(event) snapshot.fill_out_prev_events(event) yield self.state_handler.annotate_state_groups(event) diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py index c6f6ab14d1..8394013df3 100644 --- a/synapse/handlers/message.py +++ b/synapse/handlers/message.py @@ -81,7 +81,7 @@ class MessageHandler(BaseHandler): user = self.hs.parse_userid(event.user_id) assert user.is_mine, "User must be our own: %s" % (user,) - snapshot = yield self.store.snapshot_room(event.room_id, event.user_id) + snapshot = yield self.store.snapshot_room(event) yield self._on_new_room_event( event, snapshot, suppress_auth=suppress_auth @@ -141,12 +141,7 @@ class MessageHandler(BaseHandler): SynapseError if something went wrong. """ - snapshot = yield self.store.snapshot_room( - event.room_id, - event.user_id, - state_type=event.type, - state_key=event.state_key, - ) + snapshot = yield self.store.snapshot_room(event) yield self._on_new_room_event(event, snapshot) @@ -214,7 +209,7 @@ class MessageHandler(BaseHandler): @defer.inlineCallbacks def send_feedback(self, event): - snapshot = yield self.store.snapshot_room(event.room_id, event.user_id) + snapshot = yield self.store.snapshot_room(event) # store message in db yield self._on_new_room_event(event, snapshot) diff --git a/synapse/handlers/profile.py b/synapse/handlers/profile.py index 4cd0a06093..e47814483a 100644 --- a/synapse/handlers/profile.py +++ b/synapse/handlers/profile.py @@ -17,7 +17,6 @@ from twisted.internet import defer from synapse.api.errors import SynapseError, AuthError, CodeMessageException from synapse.api.constants import Membership -from synapse.api.events.room import RoomMemberEvent from ._base import BaseHandler @@ -196,10 +195,7 @@ class ProfileHandler(BaseHandler): ) for j in joins: - snapshot = yield self.store.snapshot_room( - j.room_id, j.state_key, RoomMemberEvent.TYPE, - j.state_key - ) + snapshot = yield self.store.snapshot_room(j) content = { "membership": j.content["membership"], diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py index f176ad39bf..55c893eb58 100644 --- a/synapse/handlers/room.py +++ b/synapse/handlers/room.py @@ -122,10 +122,7 @@ class RoomCreationHandler(BaseHandler): @defer.inlineCallbacks def handle_event(event): - snapshot = yield self.store.snapshot_room( - room_id=room_id, - user_id=user_id, - ) + snapshot = yield self.store.snapshot_room(event) logger.debug("Event: %s", event) @@ -364,10 +361,8 @@ class RoomMemberHandler(BaseHandler): """ target_user_id = event.state_key - snapshot = yield self.store.snapshot_room( - event.room_id, event.user_id, - RoomMemberEvent.TYPE, target_user_id - ) + snapshot = yield self.store.snapshot_room(event) + ## TODO(markjh): get prev state from snapshot. prev_state = yield self.store.get_room_member( target_user_id, event.room_id @@ -442,10 +437,7 @@ class RoomMemberHandler(BaseHandler): content=content, ) - snapshot = yield self.store.snapshot_room( - room_id, joinee.to_string(), RoomMemberEvent.TYPE, - joinee.to_string() - ) + snapshot = yield self.store.snapshot_room(new_event) yield self._do_join(new_event, snapshot, room_host=host, do_auth=True) diff --git a/synapse/rest/room.py b/synapse/rest/room.py index ec0ce78fda..997895dab0 100644 --- a/synapse/rest/room.py +++ b/synapse/rest/room.py @@ -138,7 +138,7 @@ class RoomStateEventRestServlet(RestServlet): raise SynapseError( 404, "Event not found.", errcode=Codes.NOT_FOUND ) - defer.returnValue((200, data[0].get_dict()["content"])) + defer.returnValue((200, data.get_dict()["content"])) @defer.inlineCallbacks def on_PUT(self, request, room_id, event_type, state_key): diff --git a/synapse/state.py b/synapse/state.py index 32744e047c..97a8160a33 100644 --- a/synapse/state.py +++ b/synapse/state.py @@ -45,40 +45,6 @@ class StateHandler(object): self.server_name = hs.hostname self.hs = hs - @defer.inlineCallbacks - @log_function - def handle_new_event(self, event, snapshot): - """ Given an event this works out if a) we have sufficient power level - to update the state and b) works out what the prev_state should be. - - Returns: - Deferred: Resolved with a boolean indicating if we successfully - updated the state. - - Raised: - AuthError - """ - # This needs to be done in a transaction. - - if not hasattr(event, "state_key"): - return - - # Now I need to fill out the prev state and work out if it has auth - # (w.r.t. to power levels) - - snapshot.fill_out_prev_events(event) - yield self.annotate_state_groups(event) - - if event.old_state_events: - current_state = event.old_state_events.get( - (event.type, event.state_key) - ) - - if current_state: - event.prev_state = current_state.event_id - - defer.returnValue(True) - @defer.inlineCallbacks @log_function def annotate_state_groups(self, event, old_state=None): @@ -111,7 +77,10 @@ class StateHandler(object): event.old_state_events = copy.deepcopy(new_state) if hasattr(event, "state_key"): - new_state[(event.type, event.state_key)] = event + key = (event.type, event.state_key) + if key in new_state: + event.replaces_state = new_state[key].event_id + new_state[key] = event event.state_group = None event.state_events = new_state diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index 6b8fed4502..2d62fc2ed0 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -242,8 +242,8 @@ class DataStore(RoomMemberStore, RoomStore, "state_key": event.state_key, } - if hasattr(event, "prev_state"): - vals["prev_state"] = event.prev_state + if hasattr(event, "replaces_state"): + vals["prev_state"] = event.replaces_state self._simple_insert_txn(txn, "state_events", vals) @@ -258,6 +258,40 @@ class DataStore(RoomMemberStore, RoomStore, } ) + for e_id, h in event.prev_state: + self._simple_insert_txn( + txn, + table="event_edges", + values={ + "event_id": event.event_id, + "prev_event_id": e_id, + "room_id": event.room_id, + "is_state": 1, + }, + or_ignore=True, + ) + + if not backfilled: + self._simple_insert_txn( + txn, + table="state_forward_extremities", + values={ + "event_id": event.event_id, + "room_id": event.room_id, + "type": event.type, + "state_key": event.state_key, + } + ) + + for prev_state_id, _ in event.prev_state: + self._simple_delete_txn( + txn, + table="state_forward_extremities", + keyvalues={ + "event_id": prev_state_id, + } + ) + for hash_alg, hash_base64 in event.hashes.items(): hash_bytes = decode_base64(hash_base64) self._store_event_content_hash_txn( @@ -357,7 +391,7 @@ class DataStore(RoomMemberStore, RoomStore, ], ) - def snapshot_room(self, room_id, user_id, state_type=None, state_key=None): + def snapshot_room(self, event): """Snapshot the room for an update by a user Args: room_id (synapse.types.RoomId): The room to snapshot. @@ -368,16 +402,29 @@ class DataStore(RoomMemberStore, RoomStore, synapse.storage.Snapshot: A snapshot of the state of the room. """ def _snapshot(txn): - membership_state = self._get_room_member(txn, user_id, room_id) - prev_events = self._get_latest_events_in_room(txn, room_id) + prev_events = self._get_latest_events_in_room( + txn, + event.room_id + ) + + prev_state = None + state_key = None + if hasattr(event, "state_key"): + state_key = event.state_key + prev_state = self._get_latest_state_in_room( + txn, + event.room_id, + type=event.type, + state_key=state_key, + ) return Snapshot( store=self, - room_id=room_id, - user_id=user_id, + room_id=event.room_id, + user_id=event.user_id, prev_events=prev_events, - membership_state=membership_state, - state_type=state_type, + prev_state=prev_state, + state_type=event.type, state_key=state_key, ) @@ -400,30 +447,29 @@ class Snapshot(object): """ def __init__(self, store, room_id, user_id, prev_events, - membership_state, state_type=None, state_key=None, - prev_state_pdu=None): + prev_state, state_type=None, state_key=None): self.store = store self.room_id = room_id self.user_id = user_id self.prev_events = prev_events - self.membership_state = membership_state + self.prev_state = prev_state self.state_type = state_type self.state_key = state_key - self.prev_state_pdu = prev_state_pdu def fill_out_prev_events(self, event): - if hasattr(event, "prev_events"): - return + if not hasattr(event, "prev_events"): + event.prev_events = [ + (event_id, hashes) + for event_id, hashes, _ in self.prev_events + ] - event.prev_events = [ - (event_id, hashes) - for event_id, hashes, _ in self.prev_events - ] + if self.prev_events: + event.depth = max([int(v) for _, _, v in self.prev_events]) + 1 + else: + event.depth = 0 - if self.prev_events: - event.depth = max([int(v) for _, _, v in self.prev_events]) + 1 - else: - event.depth = 0 + if not hasattr(event, "prev_state") and self.prev_state is not None: + event.prev_state = self.prev_state def schema_path(schema): diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 7d445b4633..7821fc4726 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -245,7 +245,6 @@ class SQLBaseStore(object): return [r[0] for r in txn.fetchall()] - def _simple_select_onecol(self, table, keyvalues, retcol): """Executes a SELECT query on the named table, which returns a list comprising of the values of the named column from the selected rows. @@ -273,17 +272,30 @@ class SQLBaseStore(object): keyvalues : dict of column names and values to select the rows with retcols : list of strings giving the names of the columns to return """ + return self.runInteraction( + "_simple_select_list", + self._simple_select_list_txn, + table, keyvalues, retcols + ) + + def _simple_select_list_txn(self, txn, table, keyvalues, retcols): + """Executes a SELECT query on the named table, which may return zero or + more rows, returning the result as a list of dicts. + + Args: + txn : Transaction object + table : string giving the table name + keyvalues : dict of column names and values to select the rows with + retcols : list of strings giving the names of the columns to return + """ sql = "SELECT %s FROM %s WHERE %s" % ( ", ".join(retcols), table, - " AND ".join("%s = ?" % (k) for k in keyvalues) + " AND ".join("%s = ?" % (k, ) for k in keyvalues) ) - def func(txn): - txn.execute(sql, keyvalues.values()) - return self.cursor_to_dict(txn) - - return self.runInteraction("_simple_select_list", func) + txn.execute(sql, keyvalues.values()) + return self.cursor_to_dict(txn) def _simple_update_one(self, table, keyvalues, updatevalues, retcols=None): @@ -417,6 +429,10 @@ class SQLBaseStore(object): d.pop("topological_ordering", None) d.pop("processed", None) d["origin_server_ts"] = d.pop("ts", 0) + replaces_state = d.pop("prev_state", None) + + if replaces_state: + d["replaces_state"] = replaces_state d.update(json.loads(row_dict["unrecognized_keys"])) d["content"] = json.loads(d["content"]) @@ -450,16 +466,32 @@ class SQLBaseStore(object): k: encode_base64(v) for k, v in signatures.items() } - ev.prev_events = self._get_prev_events(txn, ev.event_id) - - if hasattr(ev, "prev_state"): - # Load previous state_content. - # TODO: Should we be pulling this out above? - cursor = txn.execute(select_event_sql, (ev.prev_state,)) - prevs = self.cursor_to_dict(cursor) - if prevs: - prev = self._parse_event_from_row(prevs[0]) - ev.prev_content = prev.content + prevs = self._get_prev_events_and_state(txn, ev.event_id) + + ev.prev_events = [ + (e_id, h) + for e_id, h, is_state in prevs + if is_state == 0 + ] + + if hasattr(ev, "state_key"): + ev.prev_state = [ + (e_id, h) + for e_id, h, is_state in prevs + if is_state == 1 + ] + + if hasattr(ev, "replaces_state"): + # Load previous state_content. + # FIXME (erikj): Handle multiple prev_states. + cursor = txn.execute( + select_event_sql, + (ev.replaces_state,) + ) + prevs = self.cursor_to_dict(cursor) + if prevs: + prev = self._parse_event_from_row(prevs[0]) + ev.prev_content = prev.content if not hasattr(ev, "redacted"): logger.debug("Doesn't have redacted key: %s", ev) diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py index f427aba879..180a764134 100644 --- a/synapse/storage/event_federation.py +++ b/synapse/storage/event_federation.py @@ -69,19 +69,21 @@ class EventFederationStore(SQLBaseStore): return results - def _get_prev_events(self, txn, event_id): - prev_ids = self._simple_select_onecol_txn( + def _get_latest_state_in_room(self, txn, room_id, type, state_key): + event_ids = self._simple_select_onecol_txn( txn, - table="event_edges", + table="state_forward_extremities", keyvalues={ - "event_id": event_id, + "room_id": room_id, + "type": type, + "state_key": state_key, }, - retcol="prev_event_id", + retcol="event_id", ) results = [] - for prev_event_id in prev_ids: - hashes = self._get_event_reference_hashes_txn(txn, prev_event_id) + for event_id in event_ids: + hashes = self._get_event_reference_hashes_txn(txn, event_id) prev_hashes = { k: encode_base64(v) for k, v in hashes.items() if k == "sha256" @@ -90,6 +92,53 @@ class EventFederationStore(SQLBaseStore): return results + def _get_prev_events(self, txn, event_id): + results = self._get_prev_events_and_state( + txn, + event_id, + is_state=0, + ) + + return [(e_id, h, ) for e_id, h, _ in results] + + def _get_prev_state(self, txn, event_id): + results = self._get_prev_events_and_state( + txn, + event_id, + is_state=1, + ) + + return [(e_id, h, ) for e_id, h, _ in results] + + def _get_prev_events_and_state(self, txn, event_id, is_state=None): + keyvalues = { + "event_id": event_id, + } + + if is_state is not None: + keyvalues["is_state"] = is_state + + res = self._simple_select_list_txn( + txn, + table="event_edges", + keyvalues=keyvalues, + retcols=["prev_event_id", "is_state"], + ) + + results = [] + for d in res: + hashes = self._get_event_reference_hashes_txn( + txn, + d["prev_event_id"] + ) + prev_hashes = { + k: encode_base64(v) for k, v in hashes.items() + if k == "sha256" + } + results.append((d["prev_event_id"], prev_hashes, d["is_state"])) + + return results + def get_min_depth(self, room_id): return self.runInteraction( "get_min_depth", @@ -135,6 +184,7 @@ class EventFederationStore(SQLBaseStore): "event_id": event_id, "prev_event_id": e_id, "room_id": room_id, + "is_state": 0, }, or_ignore=True, ) diff --git a/synapse/storage/schema/event_edges.sql b/synapse/storage/schema/event_edges.sql index e5f768c705..51695826a8 100644 --- a/synapse/storage/schema/event_edges.sql +++ b/synapse/storage/schema/event_edges.sql @@ -1,7 +1,7 @@ CREATE TABLE IF NOT EXISTS event_forward_extremities( - event_id TEXT, - room_id TEXT, + event_id TEXT NOT NULL, + room_id TEXT NOT NULL, CONSTRAINT uniqueness UNIQUE (event_id, room_id) ON CONFLICT REPLACE ); @@ -10,8 +10,8 @@ CREATE INDEX IF NOT EXISTS ev_extrem_id ON event_forward_extremities(event_id); CREATE TABLE IF NOT EXISTS event_backward_extremities( - event_id TEXT, - room_id TEXT, + event_id TEXT NOT NULL, + room_id TEXT NOT NULL, CONSTRAINT uniqueness UNIQUE (event_id, room_id) ON CONFLICT REPLACE ); @@ -20,10 +20,11 @@ CREATE INDEX IF NOT EXISTS ev_b_extrem_id ON event_backward_extremities(event_id CREATE TABLE IF NOT EXISTS event_edges( - event_id TEXT, - prev_event_id TEXT, - room_id TEXT, - CONSTRAINT uniqueness UNIQUE (event_id, prev_event_id, room_id) + event_id TEXT NOT NULL, + prev_event_id TEXT NOT NULL, + room_id TEXT NOT NULL, + is_state INTEGER NOT NULL, + CONSTRAINT uniqueness UNIQUE (event_id, prev_event_id, room_id, is_state) ); CREATE INDEX IF NOT EXISTS ev_edges_id ON event_edges(event_id); @@ -31,8 +32,8 @@ CREATE INDEX IF NOT EXISTS ev_edges_prev_id ON event_edges(prev_event_id); CREATE TABLE IF NOT EXISTS room_depth( - room_id TEXT, - min_depth INTEGER, + room_id TEXT NOT NULL, + min_depth INTEGER NOT NULL, CONSTRAINT uniqueness UNIQUE (room_id) ); @@ -40,10 +41,25 @@ CREATE INDEX IF NOT EXISTS room_depth_room ON room_depth(room_id); create TABLE IF NOT EXISTS event_destinations( - event_id TEXT, - destination TEXT, + event_id TEXT NOT NULL, + destination TEXT NOT NULL, delivered_ts INTEGER DEFAULT 0, -- or 0 if not delivered CONSTRAINT uniqueness UNIQUE (event_id, destination) ON CONFLICT REPLACE ); CREATE INDEX IF NOT EXISTS event_destinations_id ON event_destinations(event_id); + + +CREATE TABLE IF NOT EXISTS state_forward_extremities( + event_id TEXT NOT NULL, + room_id TEXT NOT NULL, + type TEXT NOT NULL, + state_key TEXT NOT NULL, + CONSTRAINT uniqueness UNIQUE (event_id, room_id) ON CONFLICT REPLACE +); + +CREATE INDEX IF NOT EXISTS st_extrem_keys ON state_forward_extremities( + room_id, type, state_key +); +CREATE INDEX IF NOT EXISTS st_extrem_id ON state_forward_extremities(event_id); + diff --git a/synapse/util/jsonobject.py b/synapse/util/jsonobject.py index c91eb897a8..e79b68f661 100644 --- a/synapse/util/jsonobject.py +++ b/synapse/util/jsonobject.py @@ -80,7 +80,7 @@ class JsonEncodedObject(object): def get_full_dict(self): d = { - k: v for (k, v) in self.__dict__.items() + k: _encode(v) for (k, v) in self.__dict__.items() if k in self.valid_keys or k in self.internal_keys } d.update(self.unrecognized_keys) -- cgit 1.5.1 From bf6b72eb558cca94e209a541188079750bfefea0 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 6 Nov 2014 18:42:18 +0000 Subject: Start implementing auth chains --- synapse/api/auth.py | 3 +- synapse/api/events/__init__.py | 2 +- synapse/handlers/_base.py | 59 ++++++++++++++++++++++++++++++++-- synapse/storage/__init__.py | 12 ++++++- synapse/storage/_base.py | 2 ++ synapse/storage/event_federation.py | 21 ++++++++++++ synapse/storage/schema/event_edges.sql | 10 ++++++ synapse/storage/signatures.py | 12 +++++++ 8 files changed, 115 insertions(+), 6 deletions(-) (limited to 'synapse/storage/_base.py') diff --git a/synapse/api/auth.py b/synapse/api/auth.py index bb25c4ec55..e1302553d7 100644 --- a/synapse/api/auth.py +++ b/synapse/api/auth.py @@ -21,8 +21,7 @@ from synapse.api.constants import Membership, JoinRules from synapse.api.errors import AuthError, StoreError, Codes, SynapseError from synapse.api.events.room import ( RoomMemberEvent, RoomPowerLevelsEvent, RoomRedactionEvent, - RoomJoinRulesEvent, InviteJoinEvent, - RoomCreateEvent, + RoomJoinRulesEvent, RoomCreateEvent, ) from synapse.util.logutils import log_function diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py index 84d3a98365..513a48f568 100644 --- a/synapse/api/events/__init__.py +++ b/synapse/api/events/__init__.py @@ -61,7 +61,7 @@ class SynapseEvent(JsonEncodedObject): "replaces_state", "redacted_because", "origin_server_ts", - "auth_chains", + "auth_events", ] internal_keys = [ diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py index 509f7b550c..2613fa7fce 100644 --- a/synapse/handlers/_base.py +++ b/synapse/handlers/_base.py @@ -14,11 +14,15 @@ # limitations under the License. from twisted.internet import defer -from synapse.api.errors import LimitExceededError +from synapse.api.errors import LimitExceededError from synapse.util.async import run_on_reactor - from synapse.crypto.event_signing import add_hashes_and_signatures +from synapse.api.events.room import ( + RoomCreateEvent, RoomMemberEvent, RoomPowerLevelsEvent, RoomJoinRulesEvent, +) +from synapse.api.constants import Membership, JoinRules +from syutil.base64util import encode_base64 import logging @@ -55,6 +59,53 @@ class BaseHandler(object): retry_after_ms=int(1000*(time_allowed - time_now)), ) + @defer.inlineCallbacks + def _add_auth(self, event): + if event.type == RoomCreateEvent.TYPE: + event.auth_events = [] + return + + auth_events = [] + + key = (RoomPowerLevelsEvent.TYPE, "", ) + power_level_event = event.old_state_events.get(key) + + if power_level_event: + auth_events.append(power_level_event.event_id) + + key = (RoomJoinRulesEvent.TYPE, "", ) + join_rule_event = event.old_state_events.get(key) + + key = (RoomMemberEvent.TYPE, event.user_id, ) + member_event = event.old_state_events.get(key) + + if join_rule_event: + join_rule = join_rule_event.content.get("join_rule") + is_public = join_rule == JoinRules.PUBLIC if join_rule else False + + if event.type == RoomMemberEvent.TYPE: + if event.content["membership"] == Membership.JOIN: + if is_public: + auth_events.append(join_rule_event.event_id) + elif member_event: + auth_events.append(member_event.event_id) + + if member_event: + if member_event.content["membership"] == Membership.JOIN: + auth_events.append(member_event.event_id) + + hashes = yield self.store.get_event_reference_hashes( + auth_events + ) + hashes = [ + { + k: encode_base64(v) for k, v in h.items() + if k == "sha256" + } + for h in hashes + ] + event.auth_events = zip(auth_events, hashes) + @defer.inlineCallbacks def _on_new_room_event(self, event, snapshot, extra_destinations=[], extra_users=[], suppress_auth=False): @@ -64,6 +115,8 @@ class BaseHandler(object): yield self.state_handler.annotate_state_groups(event) + yield self._add_auth(event) + logger.debug("Signing event...") add_hashes_and_signatures( @@ -76,6 +129,8 @@ class BaseHandler(object): logger.debug("Authing...") self.auth.check(event, raises=True) logger.debug("Authed") + else: + logger.debug("Suppressed auth.") yield self.store.persist_event(event) diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index 2a1970914f..48ad4d864f 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -19,7 +19,6 @@ from synapse.api.events.room import ( RoomMemberEvent, RoomTopicEvent, FeedbackEvent, RoomNameEvent, RoomJoinRulesEvent, - RoomPowerLevelsEvent, RoomRedactionEvent, ) @@ -302,6 +301,17 @@ class DataStore(RoomMemberStore, RoomStore, txn, event.event_id, prev_event_id, alg, hash_bytes ) + for auth_id, _ in event.auth_events: + self._simple_insert_txn( + txn, + table="event_auth", + values={ + "event_id": event.event_id, + "room_id": event.room_id, + "auth_id": auth_id, + }, + ) + (ref_alg, ref_hash_bytes) = compute_event_reference_hash(event) self._store_event_reference_hash_txn( txn, event.event_id, ref_alg, ref_hash_bytes diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 7821fc4726..9aa404695d 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -474,6 +474,8 @@ class SQLBaseStore(object): if is_state == 0 ] + ev.auth_events = self._get_auth_events(txn, ev.event_id) + if hasattr(ev, "state_key"): ev.prev_state = [ (e_id, h) diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py index 180a764134..86c68ebf87 100644 --- a/synapse/storage/event_federation.py +++ b/synapse/storage/event_federation.py @@ -139,6 +139,27 @@ class EventFederationStore(SQLBaseStore): return results + def _get_auth_events(self, txn, event_id): + auth_ids = self._simple_select_onecol_txn( + txn, + table="event_auth", + keyvalues={ + "event_id": event_id, + }, + retcol="auth_id", + ) + + results = [] + for auth_id in auth_ids: + hashes = self._get_event_reference_hashes_txn(txn, auth_id) + prev_hashes = { + k: encode_base64(v) for k, v in hashes.items() + if k == "sha256" + } + results.append((auth_id, prev_hashes)) + + return results + def get_min_depth(self, room_id): return self.runInteraction( "get_min_depth", diff --git a/synapse/storage/schema/event_edges.sql b/synapse/storage/schema/event_edges.sql index 51695826a8..be1c72a775 100644 --- a/synapse/storage/schema/event_edges.sql +++ b/synapse/storage/schema/event_edges.sql @@ -63,3 +63,13 @@ CREATE INDEX IF NOT EXISTS st_extrem_keys ON state_forward_extremities( ); CREATE INDEX IF NOT EXISTS st_extrem_id ON state_forward_extremities(event_id); + +CREATE TABLE IF NOT EXISTS event_auth( + event_id TEXT NOT NULL, + auth_id TEXT NOT NULL, + room_id TEXT NOT NULL, + CONSTRAINT uniqueness UNIQUE (event_id, auth_id, room_id) +); + +CREATE INDEX IF NOT EXISTS evauth_edges_id ON event_auth(event_id); +CREATE INDEX IF NOT EXISTS evauth_edges_auth_id ON event_auth(auth_id); \ No newline at end of file diff --git a/synapse/storage/signatures.py b/synapse/storage/signatures.py index b4b3d5d7ea..84a49088a2 100644 --- a/synapse/storage/signatures.py +++ b/synapse/storage/signatures.py @@ -55,6 +55,18 @@ class SignatureStore(SQLBaseStore): or_ignore=True, ) + def get_event_reference_hashes(self, event_ids): + def f(txn): + return [ + self._get_event_reference_hashes_txn(txn, ev) + for ev in event_ids + ] + + return self.runInteraction( + "get_event_reference_hashes", + f + ) + def _get_event_reference_hashes_txn(self, txn, event_id): """Get all the hashes for a given PDU. Args: -- cgit 1.5.1 From 1c06806f90a6368cdc3b9fa3b9053021b7c40e94 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Mon, 10 Nov 2014 10:21:32 +0000 Subject: Finish redaction algorithm. --- synapse/api/events/__init__.py | 4 ++-- synapse/api/events/utils.py | 39 ++++++++++++++++++++++++++------------- synapse/crypto/event_signing.py | 7 ++----- synapse/federation/units.py | 6 ++---- synapse/storage/_base.py | 2 +- 5 files changed, 33 insertions(+), 25 deletions(-) (limited to 'synapse/storage/_base.py') diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py index 8d65c29ac1..f1e53f23ab 100644 --- a/synapse/api/events/__init__.py +++ b/synapse/api/events/__init__.py @@ -86,8 +86,8 @@ class SynapseEvent(JsonEncodedObject): def __init__(self, raises=True, **kwargs): super(SynapseEvent, self).__init__(**kwargs) - if "content" in kwargs: - self.check_json(self.content, raises=raises) + # if "content" in kwargs: + # self.check_json(self.content, raises=raises) def get_content_template(self): """ Retrieve the JSON template for this event as a dict. diff --git a/synapse/api/events/utils.py b/synapse/api/events/utils.py index 5fc79105b5..802648f8f7 100644 --- a/synapse/api/events/utils.py +++ b/synapse/api/events/utils.py @@ -18,24 +18,31 @@ from .room import ( RoomAliasesEvent, RoomCreateEvent, ) + def prune_event(event): - """ Prunes the given event of all keys we don't know about or think could - potentially be dodgy. + """ Returns a pruned version of the given event, which removes all keys we + don't know about or think could potentially be dodgy. This is used when we "redact" an event. We want to remove all fields that the user has specified, but we do want to keep necessary information like type, state_key etc. """ - return _prune_event_or_pdu(event.type, event) - -def prune_pdu(pdu): - """Removes keys that contain unrestricted and non-essential data from a PDU - """ - return _prune_event_or_pdu(pdu.type, pdu) + event_type = event.type -def _prune_event_or_pdu(event_type, event): - # Remove all extraneous fields. - event.unrecognized_keys = {} + allowed_keys = [ + "event_id", + "user_id", + "room_id", + "hashes", + "signatures", + "content", + "type", + "state_key", + "depth", + "prev_events", + "prev_state", + "auth_events", + ] new_content = {} @@ -65,6 +72,12 @@ def _prune_event_or_pdu(event_type, event): elif event_type == RoomAliasesEvent.TYPE: add_fields("aliases") - event.content = new_content + allowed_fields = { + k: v + for k, v in event.get_full_dict().items() + if k in allowed_keys + } + + allowed_fields["content"] = new_content - return event + return type(event)(**allowed_fields) diff --git a/synapse/crypto/event_signing.py b/synapse/crypto/event_signing.py index 7d800615fe..056e8f6ca4 100644 --- a/synapse/crypto/event_signing.py +++ b/synapse/crypto/event_signing.py @@ -55,9 +55,7 @@ def _compute_content_hash(event, hash_algorithm): def compute_event_reference_hash(event, hash_algorithm=hashlib.sha256): - # FIXME(erikj): GenericEvent! - tmp_event = GenericEvent(**event.get_full_dict()) - tmp_event = prune_event(tmp_event) + tmp_event = prune_event(event) event_json = tmp_event.get_dict() event_json.pop("signatures", None) event_json.pop("age_ts", None) @@ -68,8 +66,7 @@ def compute_event_reference_hash(event, hash_algorithm=hashlib.sha256): def compute_event_signature(event, signature_name, signing_key): - tmp_event = copy.deepcopy(event) - tmp_event = prune_event(tmp_event) + tmp_event = prune_event(event) redact_json = tmp_event.get_full_dict() redact_json.pop("signatures", None) redact_json.pop("age_ts", None) diff --git a/synapse/federation/units.py b/synapse/federation/units.py index 2070ffe1e2..d98014cac7 100644 --- a/synapse/federation/units.py +++ b/synapse/federation/units.py @@ -56,17 +56,15 @@ class Pdu(JsonEncodedObject): "origin_server_ts", "type", "destinations", - "transaction_id", "prev_events", "depth", "content", - "outlier", "hashes", + "user_id", + "auth_events", "signatures", # Below this are keys valid only for State Pdus. "state_key", "prev_state", - "required_power_level", - "user_id", ] internal_keys = [ diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 9aa404695d..3ab81a78d5 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -509,7 +509,7 @@ class SQLBaseStore(object): ) if del_evs: - prune_event(ev) + ev = prune_event(ev) ev.redacted_because = del_evs[0] return events -- cgit 1.5.1 From 6447db063a0d01135582bdfb3392b419f16a19e7 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Mon, 10 Nov 2014 11:59:51 +0000 Subject: Fix backfill to work. Add auth to backfill request --- synapse/api/auth.py | 6 ++++++ synapse/federation/replication.py | 36 ++++++++++++++++++++++++++++-------- synapse/federation/transport.py | 6 +++--- synapse/handlers/federation.py | 10 +++++----- synapse/storage/_base.py | 12 ++++++++++++ synapse/storage/event_federation.py | 4 ++-- 6 files changed, 56 insertions(+), 18 deletions(-) (limited to 'synapse/storage/_base.py') diff --git a/synapse/api/auth.py b/synapse/api/auth.py index 3e5d878eed..48f9d460a3 100644 --- a/synapse/api/auth.py +++ b/synapse/api/auth.py @@ -104,6 +104,12 @@ class Auth(object): pass defer.returnValue(None) + @defer.inlineCallbacks + def check_host_in_room(self, room_id, host): + joined_hosts = yield self.store.get_joined_hosts_for_room(room_id) + + defer.returnValue(host in joined_hosts) + def check_event_sender_in_room(self, event): key = (RoomMemberEvent.TYPE, event.user_id, ) member_event = event.state_events.get(key) diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py index 719bfcc42c..7837f1c252 100644 --- a/synapse/federation/replication.py +++ b/synapse/federation/replication.py @@ -205,7 +205,7 @@ class ReplicationLayer(object): pdus = [Pdu(outlier=False, **p) for p in transaction.pdus] for pdu in pdus: - yield self._handle_new_pdu(pdu, backfilled=True) + yield self._handle_new_pdu(dest, pdu, backfilled=True) defer.returnValue(pdus) @@ -274,9 +274,9 @@ class ReplicationLayer(object): @defer.inlineCallbacks @log_function - def on_backfill_request(self, context, versions, limit): + def on_backfill_request(self, origin, context, versions, limit): pdus = yield self.handler.on_backfill_request( - context, versions, limit + origin, context, versions, limit ) defer.returnValue((200, self._transaction_from_pdus(pdus).get_dict())) @@ -408,13 +408,22 @@ class ReplicationLayer(object): @defer.inlineCallbacks def on_make_join_request(self, context, user_id): pdu = yield self.handler.on_make_join_request(context, user_id) - defer.returnValue(pdu.get_dict()) + defer.returnValue({ + "event": pdu.get_dict(), + }) @defer.inlineCallbacks def on_invite_request(self, origin, content): pdu = Pdu(**content) ret_pdu = yield self.handler.on_invite_request(origin, pdu) - defer.returnValue((200, ret_pdu.get_dict())) + defer.returnValue( + ( + 200, + { + "event": ret_pdu.get_dict(), + } + ) + ) @defer.inlineCallbacks def on_send_join_request(self, origin, content): @@ -429,16 +438,25 @@ class ReplicationLayer(object): @defer.inlineCallbacks def on_event_auth(self, origin, context, event_id): auth_pdus = yield self.handler.on_event_auth(event_id) - defer.returnValue((200, [a.get_dict() for a in auth_pdus])) + defer.returnValue( + ( + 200, + { + "auth_chain": [a.get_dict() for a in auth_pdus], + } + ) + ) @defer.inlineCallbacks def make_join(self, destination, context, user_id): - pdu_dict = yield self.transport_layer.make_join( + ret = yield self.transport_layer.make_join( destination=destination, context=context, user_id=user_id, ) + pdu_dict = ret["event"] + logger.debug("Got response to make_join: %s", pdu_dict) defer.returnValue(Pdu(**pdu_dict)) @@ -467,13 +485,15 @@ class ReplicationLayer(object): @defer.inlineCallbacks def send_invite(self, destination, context, event_id, pdu): - code, pdu_dict = yield self.transport_layer.send_invite( + code, content = yield self.transport_layer.send_invite( destination=destination, context=context, event_id=event_id, content=pdu.get_dict(), ) + pdu_dict = content["event"] + logger.debug("Got response to send_invite: %s", pdu_dict) defer.returnValue(Pdu(**pdu_dict)) diff --git a/synapse/federation/transport.py b/synapse/federation/transport.py index babe8447eb..92a1f4ce17 100644 --- a/synapse/federation/transport.py +++ b/synapse/federation/transport.py @@ -413,7 +413,7 @@ class TransportLayer(object): self._with_authentication( lambda origin, content, query, context: self._on_backfill_request( - context, query["v"], query["limit"] + origin, context, query["v"], query["limit"] ) ) ) @@ -552,7 +552,7 @@ class TransportLayer(object): defer.returnValue(data) @log_function - def _on_backfill_request(self, context, v_list, limits): + def _on_backfill_request(self, origin, context, v_list, limits): if not limits: return defer.succeed( (400, {"error": "Did not include limit param"}) @@ -563,7 +563,7 @@ class TransportLayer(object): versions = v_list return self.request_handler.on_backfill_request( - context, versions, limit + origin, context, versions, limit ) @defer.inlineCallbacks diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index 9a59fe94d2..00d10609b8 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -193,10 +193,7 @@ class FederationHandler(BaseHandler): dest, room_id, limit, - extremities=[ - self.pdu_codec.decode_event_id(e) - for e in extremities - ] + extremities=extremities, ) events = [] @@ -473,7 +470,10 @@ class FederationHandler(BaseHandler): @defer.inlineCallbacks @log_function - def on_backfill_request(self, context, pdu_list, limit): + def on_backfill_request(self, origin, context, pdu_list, limit): + in_room = yield self.auth.check_host_in_room(context, origin) + if not in_room: + raise AuthError(403, "Host not in room.") events = yield self.store.get_backfill_events( context, diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 3ab81a78d5..a23f2b941b 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -447,6 +447,18 @@ class SQLBaseStore(object): **d ) + def _get_events_txn(self, txn, event_ids): + # FIXME (erikj): This should be batched? + + sql = "SELECT * FROM events WHERE event_id = ?" + + event_rows = [] + for e_id in event_ids: + c = txn.execute(sql, (e_id,)) + event_rows.extend(self.cursor_to_dict(c)) + + return self._parse_events_txn(txn, event_rows) + def _parse_events(self, rows): return self.runInteraction( "_parse_events", self._parse_events_txn, rows diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py index 06e32d592d..a707030145 100644 --- a/synapse/storage/event_federation.py +++ b/synapse/storage/event_federation.py @@ -371,10 +371,10 @@ class EventFederationStore(SQLBaseStore): "_backfill_interaction: got id=%s", *row ) - new_front.append(row) + new_front.append(row[0]) front = new_front event_results += new_front # We also want to update the `prev_pdus` attributes before returning. - return self._get_pdu_tuples(txn, event_results) + return self._get_events_txn(txn, event_results) -- cgit 1.5.1 From cdc1b5d629b49300de848647bd2a80a024fae8f7 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Mon, 10 Nov 2014 15:21:30 +0000 Subject: Fix regression where we did not return redacted events. --- synapse/storage/_base.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) (limited to 'synapse/storage/_base.py') diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index a23f2b941b..2df64bdfeb 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -469,7 +469,7 @@ class SQLBaseStore(object): select_event_sql = "SELECT * FROM events WHERE event_id = ?" - for ev in events: + for i, ev in enumerate(events): signatures = self._get_event_origin_signatures_txn( txn, ev.event_id, ) @@ -522,6 +522,7 @@ class SQLBaseStore(object): if del_evs: ev = prune_event(ev) + events[i] = ev ev.redacted_because = del_evs[0] return events -- cgit 1.5.1 From a8e565eca8cbfcedbdfd812c98a6545c2fc31afd Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Mon, 10 Nov 2014 18:24:43 +0000 Subject: Add an EventValidator. Fix bugs in auth ++ storage --- synapse/api/auth.py | 16 +++++++---- synapse/api/events/__init__.py | 61 ----------------------------------------- synapse/handlers/profile.py | 11 +++++--- synapse/rest/base.py | 2 ++ synapse/rest/room.py | 13 +++++++++ synapse/server.py | 5 ++++ synapse/storage/_base.py | 2 +- synapse/storage/registration.py | 6 +++- synapse/storage/room.py | 38 +++++++++++++------------ 9 files changed, 64 insertions(+), 90 deletions(-) (limited to 'synapse/storage/_base.py') diff --git a/synapse/api/auth.py b/synapse/api/auth.py index a5c6964707..6c2d3db26e 100644 --- a/synapse/api/auth.py +++ b/synapse/api/auth.py @@ -70,6 +70,7 @@ class Auth(object): logger.debug("Denying! %s", event) return allowed + self.check_event_sender_in_room(event) self._can_send_event(event) if event.type == RoomPowerLevelsEvent.TYPE: @@ -83,8 +84,10 @@ class Auth(object): else: raise AuthError(500, "Unknown event: %s" % event) except AuthError as e: - logger.info("Event auth check failed on event %s with msg: %s", - event, e.msg) + logger.info( + "Event auth check failed on event %s with msg: %s", + event, e.msg + ) logger.info("Denying! %s", event) if raises: raise e @@ -277,7 +280,7 @@ class Auth(object): default=[""] )[0] if user and access_token and ip_addr: - self.store.insert_client_ip( + yield self.store.insert_client_ip( user=user, access_token=access_token, device_id=user_info["device_id"], @@ -349,7 +352,8 @@ class Auth(object): if event.type == RoomMemberEvent.TYPE: e_type = event.content["membership"] if e_type in [Membership.JOIN, Membership.INVITE]: - auth_events.append(join_rule_event.event_id) + if join_rule_event: + auth_events.append(join_rule_event.event_id) if member_event and not is_public: auth_events.append(member_event.event_id) @@ -405,7 +409,9 @@ class Auth(object): if user_level < send_level: raise AuthError( - 403, "You don't have permission to post that to the room" + 403, + "You don't have permission to post that to the room. " + + "user_level (%d) < send_level (%d)" % (user_level, send_level) ) return True diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py index f1e53f23ab..1d8bed2906 100644 --- a/synapse/api/events/__init__.py +++ b/synapse/api/events/__init__.py @@ -13,7 +13,6 @@ # See the License for the specific language governing permissions and # limitations under the License. -from synapse.api.errors import SynapseError, Codes from synapse.util.jsonobject import JsonEncodedObject @@ -118,66 +117,6 @@ class SynapseEvent(JsonEncodedObject): """ raise NotImplementedError("get_content_template not implemented.") - def check_json(self, content, raises=True): - """Checks the given JSON content abides by the rules of the template. - - Args: - content : A JSON object to check. - raises: True to raise a SynapseError if the check fails. - Returns: - True if the content passes the template. Returns False if the check - fails and raises=False. - Raises: - SynapseError if the check fails and raises=True. - """ - # recursively call to inspect each layer - err_msg = self._check_json(content, self.get_content_template()) - if err_msg: - if raises: - raise SynapseError(400, err_msg, Codes.BAD_JSON) - else: - return False - else: - return True - - def _check_json(self, content, template): - """Check content and template matches. - - If the template is a dict, each key in the dict will be validated with - the content, else it will just compare the types of content and - template. This basic type check is required because this function will - be recursively called and could be called with just strs or ints. - - Args: - content: The content to validate. - template: The validation template. - Returns: - str: An error message if the validation fails, else None. - """ - if type(content) != type(template): - return "Mismatched types: %s" % template - - if type(template) == dict: - for key in template: - if key not in content: - return "Missing %s key" % key - - if type(content[key]) != type(template[key]): - return "Key %s is of the wrong type (got %s, want %s)" % ( - key, type(content[key]), type(template[key])) - - if type(content[key]) == dict: - # we must go deeper - msg = self._check_json(content[key], template[key]) - if msg: - return msg - elif type(content[key]) == list: - # make sure each item type in content matches the template - for entry in content[key]: - msg = self._check_json(entry, template[key][0]) - if msg: - return msg - class SynapseStateEvent(SynapseEvent): diff --git a/synapse/handlers/profile.py b/synapse/handlers/profile.py index e47814483a..834b37f5f3 100644 --- a/synapse/handlers/profile.py +++ b/synapse/handlers/profile.py @@ -152,10 +152,13 @@ class ProfileHandler(BaseHandler): if not user.is_mine: defer.returnValue(None) - (displayname, avatar_url) = yield defer.gatherResults([ - self.store.get_profile_displayname(user.localpart), - self.store.get_profile_avatar_url(user.localpart), - ]) + (displayname, avatar_url) = yield defer.gatherResults( + [ + self.store.get_profile_displayname(user.localpart), + self.store.get_profile_avatar_url(user.localpart), + ], + consumeErrors=True + ) state["displayname"] = displayname state["avatar_url"] = avatar_url diff --git a/synapse/rest/base.py b/synapse/rest/base.py index dc784c1527..79fc4dfb84 100644 --- a/synapse/rest/base.py +++ b/synapse/rest/base.py @@ -67,6 +67,8 @@ class RestServlet(object): self.auth = hs.get_auth() self.txns = HttpTransactionStore() + self.validator = hs.get_event_validator() + def register(self, http_server): """ Register this servlet with the given HTTP server. """ if hasattr(self, "PATTERN"): diff --git a/synapse/rest/room.py b/synapse/rest/room.py index 5c9c9d3af4..05da0be090 100644 --- a/synapse/rest/room.py +++ b/synapse/rest/room.py @@ -154,6 +154,9 @@ class RoomStateEventRestServlet(RestServlet): user_id=user.to_string(), state_key=urllib.unquote(state_key) ) + + self.validator.validate(event) + if event_type == RoomMemberEvent.TYPE: # membership events are special handler = self.handlers.room_member_handler @@ -188,6 +191,8 @@ class RoomSendEventRestServlet(RestServlet): content=content ) + self.validator.validate(event) + msg_handler = self.handlers.message_handler yield msg_handler.send_message(event) @@ -253,6 +258,9 @@ class JoinRoomAliasServlet(RestServlet): user_id=user.to_string(), state_key=user.to_string() ) + + self.validator.validate(event) + handler = self.handlers.room_member_handler yield handler.change_membership(event) defer.returnValue((200, {})) @@ -424,6 +432,9 @@ class RoomMembershipRestServlet(RestServlet): user_id=user.to_string(), state_key=state_key ) + + self.validator.validate(event) + handler = self.handlers.room_member_handler yield handler.change_membership(event) defer.returnValue((200, {})) @@ -461,6 +472,8 @@ class RoomRedactEventRestServlet(RestServlet): redacts=urllib.unquote(event_id), ) + self.validator.validate(event) + msg_handler = self.handlers.message_handler yield msg_handler.send_message(event) diff --git a/synapse/server.py b/synapse/server.py index d770b20b19..da0a44433a 100644 --- a/synapse/server.py +++ b/synapse/server.py @@ -22,6 +22,7 @@ from synapse.federation import initialize_http_replication from synapse.api.events import serialize_event from synapse.api.events.factory import EventFactory +from synapse.api.events.validator import EventValidator from synapse.notifier import Notifier from synapse.api.auth import Auth from synapse.handlers import Handlers @@ -80,6 +81,7 @@ class BaseHomeServer(object): 'event_sources', 'ratelimiter', 'keyring', + 'event_validator', ] def __init__(self, hostname, **kwargs): @@ -223,6 +225,9 @@ class HomeServer(BaseHomeServer): def build_keyring(self): return Keyring(self) + def build_event_validator(self): + return EventValidator(self) + def register_servlets(self): """ Register all servlets associated with this HomeServer. """ diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 2df64bdfeb..a1ee0318f6 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -212,7 +212,7 @@ class SQLBaseStore(object): retcol : string giving the name of the column to return """ return self.runInteraction( - "_simple_select_one_onecol_txn", + "_simple_select_one_onecol", self._simple_select_one_onecol_txn, table, keyvalues, retcol, allow_none=allow_none, ) diff --git a/synapse/storage/registration.py b/synapse/storage/registration.py index a2ca6f9a69..1f89d77344 100644 --- a/synapse/storage/registration.py +++ b/synapse/storage/registration.py @@ -107,13 +107,17 @@ class RegistrationStore(SQLBaseStore): token ) + @defer.inlineCallbacks def is_server_admin(self, user): - return self._simple_select_one_onecol( + res = yield self._simple_select_one_onecol( table="users", keyvalues={"name": user.to_string()}, retcol="admin", + allow_none=True, ) + defer.returnValue(res if res else False) + def _query_for_auth(self, txn, token): sql = ( "SELECT users.name, users.admin, access_tokens.device_id " diff --git a/synapse/storage/room.py b/synapse/storage/room.py index ca70506d28..cc0513b8d2 100644 --- a/synapse/storage/room.py +++ b/synapse/storage/room.py @@ -133,26 +133,28 @@ class RoomStore(SQLBaseStore): defer.returnValue(ret) def _store_room_topic_txn(self, txn, event): - self._simple_insert_txn( - txn, - "topics", - { - "event_id": event.event_id, - "room_id": event.room_id, - "topic": event.topic, - } - ) + if hasattr(event, "topic"): + self._simple_insert_txn( + txn, + "topics", + { + "event_id": event.event_id, + "room_id": event.room_id, + "topic": event.topic, + } + ) def _store_room_name_txn(self, txn, event): - self._simple_insert_txn( - txn, - "room_names", - { - "event_id": event.event_id, - "room_id": event.room_id, - "name": event.name, - } - ) + if hasattr(event, "name"): + self._simple_insert_txn( + txn, + "room_names", + { + "event_id": event.event_id, + "room_id": event.room_id, + "name": event.name, + } + ) class RoomsTable(Table): -- cgit 1.5.1 From f04b3d5042b85fa81efff9b561ca7af8d9709756 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Wed, 12 Nov 2014 17:02:18 +0000 Subject: Store all signatures on events rather than just dropping them --- synapse/storage/__init__.py | 15 ++++++++------- synapse/storage/_base.py | 7 +++++-- synapse/storage/schema/event_signatures.sql | 6 +++--- synapse/storage/signatures.py | 24 +++++++++++++++--------- 4 files changed, 31 insertions(+), 21 deletions(-) (limited to 'synapse/storage/_base.py') diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index 72290eb5a0..d8f351a675 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -279,13 +279,14 @@ class DataStore(RoomMemberStore, RoomStore, ) if hasattr(event, "signatures"): - signatures = event.signatures.get(event.origin, {}) - - for key_id, signature_base64 in signatures.items(): - signature_bytes = decode_base64(signature_base64) - self._store_event_origin_signature_txn( - txn, event.event_id, event.origin, key_id, signature_bytes, - ) + logger.debug("sigs: %s", event.signatures) + for name, sigs in event.signatures.items(): + for key_id, signature_base64 in sigs.items(): + signature_bytes = decode_base64(signature_base64) + self._store_event_signature_txn( + txn, event.event_id, name, key_id, + signature_bytes, + ) for prev_event_id, prev_hashes in event.prev_events: for alg, hash_base64 in prev_hashes.items(): diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index a1ee0318f6..670387b04a 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -470,12 +470,15 @@ class SQLBaseStore(object): select_event_sql = "SELECT * FROM events WHERE event_id = ?" for i, ev in enumerate(events): - signatures = self._get_event_origin_signatures_txn( + signatures = self._get_event_signatures_txn( txn, ev.event_id, ) ev.signatures = { - k: encode_base64(v) for k, v in signatures.items() + n: { + k: encode_base64(v) for k, v in s.items() + } + for n, s in signatures.items() } prevs = self._get_prev_events_and_state(txn, ev.event_id) diff --git a/synapse/storage/schema/event_signatures.sql b/synapse/storage/schema/event_signatures.sql index 5491c7ecec..4efa8a3e63 100644 --- a/synapse/storage/schema/event_signatures.sql +++ b/synapse/storage/schema/event_signatures.sql @@ -37,15 +37,15 @@ CREATE INDEX IF NOT EXISTS event_reference_hashes_id ON event_reference_hashes ( ); -CREATE TABLE IF NOT EXISTS event_origin_signatures ( +CREATE TABLE IF NOT EXISTS event_signatures ( event_id TEXT, - origin TEXT, + signature_name TEXT, key_id TEXT, signature BLOB, CONSTRAINT uniqueness UNIQUE (event_id, key_id) ); -CREATE INDEX IF NOT EXISTS event_origin_signatures_id ON event_origin_signatures ( +CREATE INDEX IF NOT EXISTS event_signatures_id ON event_signatures ( event_id ); diff --git a/synapse/storage/signatures.py b/synapse/storage/signatures.py index 84a49088a2..d90e08fff1 100644 --- a/synapse/storage/signatures.py +++ b/synapse/storage/signatures.py @@ -103,24 +103,30 @@ class SignatureStore(SQLBaseStore): or_ignore=True, ) - - def _get_event_origin_signatures_txn(self, txn, event_id): + def _get_event_signatures_txn(self, txn, event_id): """Get all the signatures for a given PDU. Args: txn (cursor): event_id (str): Id for the Event. Returns: - A dict of key_id -> signature_bytes. + A dict of sig name -> dict(key_id -> signature_bytes) """ query = ( - "SELECT key_id, signature" - " FROM event_origin_signatures" + "SELECT signature_name, key_id, signature" + " FROM event_signatures" " WHERE event_id = ? " ) txn.execute(query, (event_id, )) - return dict(txn.fetchall()) + rows = txn.fetchall() + + res = {} + + for name, key, sig in rows: + res.setdefault(name, {})[key] = sig + + return res - def _store_event_origin_signature_txn(self, txn, event_id, origin, key_id, + def _store_event_signature_txn(self, txn, event_id, signature_name, key_id, signature_bytes): """Store a signature from the origin server for a PDU. Args: @@ -132,10 +138,10 @@ class SignatureStore(SQLBaseStore): """ self._simple_insert_txn( txn, - "event_origin_signatures", + "event_signatures", { "event_id": event_id, - "origin": origin, + "signature_name": signature_name, "key_id": key_id, "signature": buffer(signature_bytes), }, -- cgit 1.5.1 From cb4b6c844a0c9e2d4a96165958ff5680ed82e160 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Fri, 14 Nov 2014 21:25:02 +0000 Subject: Merge PDUs and Events into one object --- synapse/api/events/__init__.py | 6 +++ synapse/api/events/utils.py | 2 + synapse/crypto/event_signing.py | 15 ++----- synapse/federation/pdu_codec.py | 54 ------------------------- synapse/federation/replication.py | 61 +++++++++++++++++++--------- synapse/federation/units.py | 79 +------------------------------------ synapse/handlers/federation.py | 65 +++++++++++------------------- synapse/storage/_base.py | 8 ++++ tests/federation/test_federation.py | 8 ++-- tests/handlers/test_federation.py | 5 ++- 10 files changed, 91 insertions(+), 212 deletions(-) delete mode 100644 synapse/federation/pdu_codec.py (limited to 'synapse/storage/_base.py') diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py index 1d8bed2906..63c0bd7ae7 100644 --- a/synapse/api/events/__init__.py +++ b/synapse/api/events/__init__.py @@ -117,6 +117,12 @@ class SynapseEvent(JsonEncodedObject): """ raise NotImplementedError("get_content_template not implemented.") + def get_pdu_json(self): + pdu_json = self.get_full_dict() + pdu_json.pop("destination", None) + pdu_json.pop("outlier", None) + return pdu_json + class SynapseStateEvent(SynapseEvent): diff --git a/synapse/api/events/utils.py b/synapse/api/events/utils.py index 802648f8f7..d6019d56eb 100644 --- a/synapse/api/events/utils.py +++ b/synapse/api/events/utils.py @@ -42,6 +42,8 @@ def prune_event(event): "prev_events", "prev_state", "auth_events", + "origin", + "origin_server_ts", ] new_content = {} diff --git a/synapse/crypto/event_signing.py b/synapse/crypto/event_signing.py index 79274fd552..4dff2c0ec2 100644 --- a/synapse/crypto/event_signing.py +++ b/synapse/crypto/event_signing.py @@ -16,7 +16,6 @@ from synapse.api.events.utils import prune_event -from synapse.federation.units import Pdu from syutil.jsonutil import encode_canonical_json from syutil.base64util import encode_base64, decode_base64 from syutil.crypto.jsonsign import sign_json @@ -53,8 +52,7 @@ def check_event_content_hash(event, hash_algorithm=hashlib.sha256): def _compute_content_hash(event, hash_algorithm): - event_json = event.get_full_dict() - # TODO: We need to sign the JSON that is going out via fedaration. + event_json = event.get_pdu_json() event_json.pop("age_ts", None) event_json.pop("unsigned", None) event_json.pop("signatures", None) @@ -67,7 +65,7 @@ def _compute_content_hash(event, hash_algorithm): def compute_event_reference_hash(event, hash_algorithm=hashlib.sha256): tmp_event = prune_event(event) - event_json = tmp_event.get_dict() + event_json = tmp_event.get_pdu_json() event_json.pop("signatures", None) event_json.pop("age_ts", None) event_json.pop("unsigned", None) @@ -78,14 +76,7 @@ def compute_event_reference_hash(event, hash_algorithm=hashlib.sha256): def compute_event_signature(event, signature_name, signing_key): tmp_event = prune_event(event) - tmp_event.origin = event.origin - tmp_event.origin_server_ts = event.origin_server_ts - d = tmp_event.get_full_dict() - kwargs = dict(event.unrecognized_keys) - kwargs.update({k: v for k, v in d.items()}) - tmp_pdu = Pdu(**kwargs) - redact_json = tmp_pdu.get_dict() - redact_json.pop("signatures", None) + redact_json = tmp_event.get_pdu_json() redact_json.pop("age_ts", None) redact_json.pop("unsigned", None) logger.debug("Signing event: %s", redact_json) diff --git a/synapse/federation/pdu_codec.py b/synapse/federation/pdu_codec.py deleted file mode 100644 index 52c84efb5b..0000000000 --- a/synapse/federation/pdu_codec.py +++ /dev/null @@ -1,54 +0,0 @@ -# -*- coding: utf-8 -*- -# Copyright 2014 OpenMarket Ltd -# -# 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. - -from .units import Pdu - -import copy - - -class PduCodec(object): - - def __init__(self, hs): - self.signing_key = hs.config.signing_key[0] - self.server_name = hs.hostname - self.event_factory = hs.get_event_factory() - self.clock = hs.get_clock() - self.hs = hs - - def event_from_pdu(self, pdu): - kwargs = {} - - kwargs["etype"] = pdu.type - - kwargs.update({ - k: v - for k, v in pdu.get_full_dict().items() - if k not in [ - "type", - ] - }) - - return self.event_factory.create_event(**kwargs) - - def pdu_from_event(self, event): - d = event.get_full_dict() - - kwargs = copy.deepcopy(event.unrecognized_keys) - kwargs.update({ - k: v for k, v in d.items() - }) - - pdu = Pdu(**kwargs) - return pdu diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py index a07e307849..8ee74de005 100644 --- a/synapse/federation/replication.py +++ b/synapse/federation/replication.py @@ -19,7 +19,7 @@ a given transport. from twisted.internet import defer -from .units import Transaction, Pdu, Edu +from .units import Transaction, Edu from .persistence import TransactionActions @@ -72,6 +72,8 @@ class ReplicationLayer(object): self._clock = hs.get_clock() + self.event_factory = hs.get_event_factory() + 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 @@ -203,7 +205,10 @@ class ReplicationLayer(object): transaction = Transaction(**transaction_data) - pdus = [Pdu(outlier=False, **p) for p in transaction.pdus] + pdus = [ + self.event_from_pdu_json(p, outlier=False) + for p in transaction.pdus + ] for pdu in pdus: yield self._handle_new_pdu(dest, pdu, backfilled=True) @@ -235,7 +240,10 @@ class ReplicationLayer(object): transaction = Transaction(**transaction_data) - pdu_list = [Pdu(outlier=outlier, **p) for p in transaction.pdus] + pdu_list = [ + self.event_from_pdu_json(p, outlier=outlier) + for p in transaction.pdus + ] pdu = None if pdu_list: @@ -265,8 +273,10 @@ class ReplicationLayer(object): ) transaction = Transaction(**transaction_data) - - pdus = [Pdu(outlier=True, **p) for p in transaction.pdus] + pdus = [ + self.event_from_pdu_json(p, outlier=True) + for p in transaction.pdus + ] defer.returnValue(pdus) @@ -293,7 +303,9 @@ class ReplicationLayer(object): p["age_ts"] = int(self._clock.time_msec()) - int(p["age"]) del p["age"] - pdu_list = [Pdu(**p) for p in transaction.pdus] + pdu_list = [ + self.event_from_pdu_json(p) for p in transaction.pdus + ] logger.debug("[%s] Got transaction", transaction.transaction_id) @@ -388,30 +400,30 @@ class ReplicationLayer(object): def on_make_join_request(self, context, user_id): pdu = yield self.handler.on_make_join_request(context, user_id) defer.returnValue({ - "event": pdu.get_dict(), + "event": pdu.get_pdu_json(), }) @defer.inlineCallbacks def on_invite_request(self, origin, content): - pdu = Pdu(**content) + pdu = self.event_from_pdu_json(content) ret_pdu = yield self.handler.on_invite_request(origin, pdu) defer.returnValue( ( 200, { - "event": ret_pdu.get_dict(), + "event": ret_pdu.get_pdu_json(), } ) ) @defer.inlineCallbacks def on_send_join_request(self, origin, content): - pdu = Pdu(**content) + pdu = self.event_from_pdu_json(content) res_pdus = yield self.handler.on_send_join_request(origin, pdu) defer.returnValue((200, { - "state": [p.get_dict() for p in res_pdus["state"]], - "auth_chain": [p.get_dict() for p in res_pdus["auth_chain"]], + "state": [p.get_pdu_json() for p in res_pdus["state"]], + "auth_chain": [p.get_pdu_json() for p in res_pdus["auth_chain"]], })) @defer.inlineCallbacks @@ -421,7 +433,7 @@ class ReplicationLayer(object): ( 200, { - "auth_chain": [a.get_dict() for a in auth_pdus], + "auth_chain": [a.get_pdu_json() for a in auth_pdus], } ) ) @@ -438,7 +450,7 @@ class ReplicationLayer(object): logger.debug("Got response to make_join: %s", pdu_dict) - defer.returnValue(Pdu(**pdu_dict)) + defer.returnValue(self.event_from_pdu_json(pdu_dict)) @defer.inlineCallbacks def send_join(self, destination, pdu): @@ -446,12 +458,15 @@ class ReplicationLayer(object): destination, pdu.room_id, pdu.event_id, - pdu.get_dict(), + pdu.get_pdu_json(), ) logger.debug("Got content: %s", content) - state = [Pdu(outlier=True, **p) for p in content.get("state", [])] + state = [ + self.event_from_pdu_json(p, outlier=True) + for p in content.get("state", []) + ] # FIXME: We probably want to do something with the auth_chain given # to us @@ -468,14 +483,14 @@ class ReplicationLayer(object): destination=destination, context=context, event_id=event_id, - content=pdu.get_dict(), + content=pdu.get_pdu_json(), ) pdu_dict = content["event"] logger.debug("Got response to send_invite: %s", pdu_dict) - defer.returnValue(Pdu(**pdu_dict)) + defer.returnValue(self.event_from_pdu_json(pdu_dict)) @log_function def _get_persisted_pdu(self, origin, event_id): @@ -490,7 +505,7 @@ class ReplicationLayer(object): """Returns a new Transaction containing the given PDUs suitable for transmission. """ - pdus = [p.get_dict() for p in pdu_list] + pdus = [p.get_pdu_json() for p in pdu_list] time_now = self._clock.time_msec() for p in pdus: if "age_ts" in p: @@ -563,6 +578,14 @@ class ReplicationLayer(object): def __str__(self): return "" % self.server_name + def event_from_pdu_json(self, pdu_json, outlier=False): + #TODO: Check we have all the PDU keys here + pdu_json.setdefault("hashes", {}) + pdu_json.setdefault("signatures", {}) + return self.event_factory.create_event( + pdu_json["type"], outlier=outlier, **pdu_json + ) + class _TransactionQueue(object): """This class makes sure we only have one transaction in flight at diff --git a/synapse/federation/units.py b/synapse/federation/units.py index 70412439cd..6e708edb8c 100644 --- a/synapse/federation/units.py +++ b/synapse/federation/units.py @@ -25,83 +25,6 @@ import logging logger = logging.getLogger(__name__) -class Pdu(JsonEncodedObject): - """ A Pdu represents a piece of data sent from a server and is associated - with a context. - - A Pdu can be classified as "state". For a given context, we can efficiently - retrieve all state pdu's that haven't been clobbered. Clobbering is done - via a unique constraint on the tuple (context, type, state_key). A pdu - is a state pdu if `is_state` is True. - - Example pdu:: - - { - "event_id": "$78c:example.com", - "origin_server_ts": 1404835423000, - "origin": "bar", - "prev_ids": [ - ["23b", "foo"], - ["56a", "bar"], - ], - "content": { ... }, - } - - """ - - valid_keys = [ - "event_id", - "room_id", - "origin", - "origin_server_ts", - "type", - "destinations", - "prev_events", - "depth", - "content", - "hashes", - "user_id", - "auth_events", - "signatures", # Below this are keys valid only for State Pdus. - "state_key", - "prev_state", - ] - - internal_keys = [ - "destinations", - "transaction_id", - "outlier", - ] - - required_keys = [ - "event_id", - "room_id", - "origin", - "origin_server_ts", - "type", - "content", - ] - - # TODO: We need to make this properly load content rather than - # just leaving it as a dict. (OR DO WE?!) - - def __init__(self, destinations=[], prev_events=[], - outlier=False, hashes={}, signatures={}, **kwargs): - super(Pdu, self).__init__( - destinations=destinations, - prev_events=prev_events, - outlier=outlier, - hashes=hashes, - signatures=signatures, - **kwargs - ) - - def __str__(self): - return "(%s, %s)" % (self.__class__.__name__, repr(self.__dict__)) - - def __repr__(self): - return "<%s, %s>" % (self.__class__.__name__, repr(self.__dict__)) - class Edu(JsonEncodedObject): """ An Edu represents a piece of data sent from one homeserver to another. @@ -202,6 +125,6 @@ class Transaction(JsonEncodedObject): for p in pdus: p.transaction_id = kwargs["transaction_id"] - kwargs["pdus"] = [p.get_dict() for p in pdus] + kwargs["pdus"] = [p.get_pdu_json() for p in pdus] return Transaction(**kwargs) diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index fc00128c56..da38f34e6a 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -22,7 +22,6 @@ from synapse.api.errors import AuthError, FederationError, SynapseError from synapse.api.events.room import RoomMemberEvent from synapse.api.constants import Membership from synapse.util.logutils import log_function -from synapse.federation.pdu_codec import PduCodec from synapse.util.async import run_on_reactor from synapse.crypto.event_signing import ( compute_event_signature, check_event_content_hash @@ -69,8 +68,6 @@ class FederationHandler(BaseHandler): self.replication_layer.set_handler(self) - self.pdu_codec = PduCodec(hs) - # When joining a room we need to queue any events for that room up self.room_queues = {} @@ -92,7 +89,7 @@ class FederationHandler(BaseHandler): yield run_on_reactor() - pdu = self.pdu_codec.pdu_from_event(event) + pdu = event if not hasattr(pdu, "destinations") or not pdu.destinations: pdu.destinations = [] @@ -105,7 +102,7 @@ class FederationHandler(BaseHandler): """ Called by the ReplicationLayer when we have a new pdu. We need to do auth checks and put it through the StateHandler. """ - event = self.pdu_codec.event_from_pdu(pdu) + event = pdu logger.debug("Got event: %s", event.event_id) @@ -118,18 +115,15 @@ class FederationHandler(BaseHandler): logger.debug("Processing event: %s", event.event_id) redacted_event = prune_event(event) - redacted_event.origin = pdu.origin - redacted_event.origin_server_ts = pdu.origin_server_ts - redacted_pdu = self.pdu_codec.pdu_from_event(redacted_event) - redacted_pdu_json = redacted_pdu.get_dict() + redacted_pdu_json = redacted_event.get_pdu_json() try: yield self.keyring.verify_json_for_server( event.origin, redacted_pdu_json ) except SynapseError as e: logger.warn("Signature check failed for %s redacted to %s", - encode_canonical_json(pdu.get_dict()), + encode_canonical_json(pdu.get_pdu_json()), encode_canonical_json(redacted_pdu_json), ) raise FederationError( @@ -147,7 +141,7 @@ class FederationHandler(BaseHandler): event = redacted_event if state: - state = [self.pdu_codec.event_from_pdu(p) for p in state] + state = [p for p in state] is_new_state = yield self.state_handler.annotate_event_with_state( event, @@ -239,7 +233,7 @@ class FederationHandler(BaseHandler): events = [] for pdu in pdus: - event = self.pdu_codec.event_from_pdu(pdu) + event = pdu # FIXME (erikj): Not sure this actually works :/ yield self.state_handler.annotate_event_with_state(event) @@ -260,15 +254,15 @@ class FederationHandler(BaseHandler): destination=target_host, context=event.room_id, event_id=event.event_id, - pdu=self.pdu_codec.pdu_from_event(event) + pdu=event ) - defer.returnValue(self.pdu_codec.event_from_pdu(pdu)) + defer.returnValue(pdu) @defer.inlineCallbacks def on_event_auth(self, event_id): auth = yield self.store.get_auth_chain(event_id) - defer.returnValue([self.pdu_codec.pdu_from_event(e) for e in auth]) + defer.returnValue([e for e in auth]) @log_function @defer.inlineCallbacks @@ -292,7 +286,7 @@ class FederationHandler(BaseHandler): logger.debug("Got response to make_join: %s", pdu) - event = self.pdu_codec.event_from_pdu(pdu) + event = pdu # We should assert some things. assert(event.type == RoomMemberEvent.TYPE) @@ -310,10 +304,10 @@ class FederationHandler(BaseHandler): state = yield self.replication_layer.send_join( target_host, - self.pdu_codec.pdu_from_event(event) + event ) - state = [self.pdu_codec.event_from_pdu(p) for p in state] + state = [p for p in state] logger.debug("do_invite_join state: %s", state) @@ -387,7 +381,7 @@ class FederationHandler(BaseHandler): yield self.auth.add_auth_events(event) self.auth.check(event, raises=True) - pdu = self.pdu_codec.pdu_from_event(event) + pdu = event defer.returnValue(pdu) @@ -397,7 +391,7 @@ class FederationHandler(BaseHandler): """ We have received a join event for a room. Fully process it and respond with the current state and auth chains. """ - event = self.pdu_codec.event_from_pdu(pdu) + event = pdu event.outlier = False @@ -429,7 +423,7 @@ class FederationHandler(BaseHandler): "user_joined_room", user=user, room_id=event.room_id ) - new_pdu = self.pdu_codec.pdu_from_event(event) + new_pdu = event destinations = set() @@ -450,17 +444,10 @@ class FederationHandler(BaseHandler): yield self.replication_layer.send_pdu(new_pdu) auth_chain = yield self.store.get_auth_chain(event.event_id) - pdu_auth_chain = [ - self.pdu_codec.pdu_from_event(e) - for e in auth_chain - ] defer.returnValue({ - "state": [ - self.pdu_codec.pdu_from_event(e) - for e in event.state_events.values() - ], - "auth_chain": pdu_auth_chain, + "state": event.state_events.values(), + "auth_chain": auth_chain, }) @defer.inlineCallbacks @@ -469,7 +456,7 @@ class FederationHandler(BaseHandler): Respond with the now signed event. """ - event = self.pdu_codec.event_from_pdu(pdu) + event = pdu event.outlier = True @@ -493,7 +480,7 @@ class FederationHandler(BaseHandler): event, extra_users=[target_user], ) - defer.returnValue(self.pdu_codec.pdu_from_event(event)) + defer.returnValue(event) @defer.inlineCallbacks def get_state_for_pdu(self, origin, room_id, event_id): @@ -524,12 +511,7 @@ class FederationHandler(BaseHandler): else: del results[(event.type, event.state_key)] - defer.returnValue( - [ - self.pdu_codec.pdu_from_event(s) - for s in results.values() - ] - ) + defer.returnValue(results.values()) else: defer.returnValue([]) @@ -546,10 +528,7 @@ class FederationHandler(BaseHandler): limit ) - defer.returnValue([ - self.pdu_codec.pdu_from_event(e) - for e in events - ]) + defer.returnValue(events) @defer.inlineCallbacks @log_function @@ -572,7 +551,7 @@ class FederationHandler(BaseHandler): if not in_room: raise AuthError(403, "Host not in room.") - defer.returnValue(self.pdu_codec.pdu_from_event(event)) + defer.returnValue(event) else: defer.returnValue(None) diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 30e6eac8db..5d4be09a82 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -492,6 +492,14 @@ class SQLBaseStore(object): for n, s in signatures.items() } + hashes = self._get_event_content_hashes_txn( + txn, ev.event_id, + ) + + ev.hashes = { + k: encode_base64(v) for k, v in hashes.items() + } + prevs = self._get_prev_events_and_state(txn, ev.event_id) ev.prev_events = [ diff --git a/tests/federation/test_federation.py b/tests/federation/test_federation.py index ad09fab392..efac4075dc 100644 --- a/tests/federation/test_federation.py +++ b/tests/federation/test_federation.py @@ -23,7 +23,7 @@ from ..utils import MockHttpResource, MockClock, MockKey from synapse.server import HomeServer from synapse.federation import initialize_http_replication -from synapse.federation.units import Pdu +from synapse.api.events import SynapseEvent def make_pdu(prev_pdus=[], **kwargs): @@ -40,7 +40,7 @@ def make_pdu(prev_pdus=[], **kwargs): } pdu_fields.update(kwargs) - return Pdu(prev_pdus=prev_pdus, **pdu_fields) + return SynapseEvent(prev_pdus=prev_pdus, **pdu_fields) class FederationTestCase(unittest.TestCase): @@ -169,7 +169,7 @@ class FederationTestCase(unittest.TestCase): (200, "OK") ) - pdu = Pdu( + pdu = SynapseEvent( event_id="abc123def456", origin="red", room_id="my-context", @@ -189,7 +189,7 @@ class FederationTestCase(unittest.TestCase): "origin_server_ts": 1000000, "origin": "test", "pdus": [ - pdu.get_dict(), + pdu.get_pdu_json(), ], 'pdu_failures': [], }, diff --git a/tests/handlers/test_federation.py b/tests/handlers/test_federation.py index 3f17ca8fb0..e19b073817 100644 --- a/tests/handlers/test_federation.py +++ b/tests/handlers/test_federation.py @@ -19,9 +19,10 @@ from tests import unittest from synapse.api.events.room import ( MessageEvent, ) + +from synapse.api.events import SynapseEvent from synapse.handlers.federation import FederationHandler from synapse.server import HomeServer -from synapse.federation.units import Pdu from mock import NonCallableMock, ANY, Mock @@ -74,7 +75,7 @@ class FederationTestCase(unittest.TestCase): @defer.inlineCallbacks def test_msg(self): - pdu = Pdu( + pdu = SynapseEvent( type=MessageEvent.TYPE, room_id="foo", content={"msgtype": u"fooo"}, -- cgit 1.5.1