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/storage/event_federation.py | 143 ++++++++++++++++++++++++++++++++++++ 1 file changed, 143 insertions(+) create mode 100644 synapse/storage/event_federation.py (limited to 'synapse/storage/event_federation.py') 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 -- cgit 1.4.1 From a10c2ec88d98abe035a60ab0027c1914d4ad7d77 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 28 Oct 2014 17:15:32 +0000 Subject: Don't reference PDU when persisting event --- synapse/storage/__init__.py | 5 +++-- synapse/storage/event_federation.py | 4 ++-- 2 files changed, 5 insertions(+), 4 deletions(-) (limited to 'synapse/storage/event_federation.py') diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index 678de0cf50..f89e518690 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -322,9 +322,10 @@ class DataStore(RoomMemberStore, RoomStore, txn, event.event_id, prev_event_id, alg, hash_bytes ) + # TODO (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._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) diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py index 27ad9aea4d..7688fc550f 100644 --- a/synapse/storage/event_federation.py +++ b/synapse/storage/event_federation.py @@ -14,7 +14,7 @@ # limitations under the License. from ._base import SQLBaseStore -from twisted.internet import defer +from syutil.base64util import encode_base64 import logging @@ -36,7 +36,7 @@ class EventFederationStore(SQLBaseStore): results = [] for pdu_id, origin, depth in txn.fetchall(): - hashes = self._get_pdu_reference_hashes_txn(txn, pdu_id, origin) + 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)) -- cgit 1.4.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/event_federation.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.4.1 From aa80900a8e8cd9e7305a66cec336a8e150c46651 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 30 Oct 2014 10:11:06 +0000 Subject: Fix SQL so that accepts we may want to persist events twice. --- synapse/storage/event_federation.py | 8 +++-- synapse/storage/signatures.py | 64 ++++++++++++++++++++++++------------- 2 files changed, 47 insertions(+), 25 deletions(-) (limited to 'synapse/storage/event_federation.py') diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py index 5f94c31818..88d09d9ba8 100644 --- a/synapse/storage/event_federation.py +++ b/synapse/storage/event_federation.py @@ -99,7 +99,8 @@ class EventFederationStore(SQLBaseStore): "event_id": event_id, "prev_event_id": e_id, "room_id": room_id, - } + }, + or_ignore=True, ) # Update the extremities table if this is not an outlier. @@ -120,7 +121,7 @@ class EventFederationStore(SQLBaseStore): # 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) " + "INSERT OR IGNORE INTO %(table)s (event_id, room_id) " "SELECT ?, ? WHERE NOT EXISTS (" "SELECT 1 FROM %(event_edges)s WHERE " "prev_event_id = ? " @@ -144,7 +145,8 @@ class EventFederationStore(SQLBaseStore): values={ "event_id": e_id, "room_id": room_id, - } + }, + or_ignore=True, ) # Also delete from the backwards extremities table all ones that diff --git a/synapse/storage/signatures.py b/synapse/storage/signatures.py index b8f8fd44cb..5e99174fcd 100644 --- a/synapse/storage/signatures.py +++ b/synapse/storage/signatures.py @@ -181,11 +181,16 @@ class SignatureStore(SQLBaseStore): 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), - }) + self._simple_insert_txn( + txn, + "event_content_hashes", + { + "event_id": event_id, + "algorithm": algorithm, + "hash": buffer(hash_bytes), + }, + or_ignore=True, + ) def _get_event_reference_hashes_txn(self, txn, event_id): """Get all the hashes for a given PDU. @@ -212,11 +217,16 @@ class SignatureStore(SQLBaseStore): 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), - }) + self._simple_insert_txn( + txn, + "event_reference_hashes", + { + "event_id": event_id, + "algorithm": algorithm, + "hash": buffer(hash_bytes), + }, + or_ignore=True, + ) def _get_event_origin_signatures_txn(self, txn, event_id): @@ -245,12 +255,17 @@ class SignatureStore(SQLBaseStore): 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), - }) + self._simple_insert_txn( + txn, + "event_origin_signatures", + { + "event_id": event_id, + "origin": origin, + "key_id": key_id, + "signature": buffer(signature_bytes), + }, + or_ignore=True, + ) def _get_prev_event_hashes_txn(self, txn, event_id): """Get all the hashes for previous PDUs of a PDU @@ -274,9 +289,14 @@ class SignatureStore(SQLBaseStore): 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 + self._simple_insert_txn( + txn, + "event_edge_hashes", + { + "event_id": event_id, + "prev_event_id": prev_event_id, + "algorithm": algorithm, + "hash": buffer(hash_bytes), + }, + or_ignore=True, + ) \ No newline at end of file -- cgit 1.4.1 From f2de2d644af80557baebf43f64f3968b8ab46d0b Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 31 Oct 2014 09:59:02 +0000 Subject: Move the impl of backfill to use events. --- synapse/federation/replication.py | 6 +-- synapse/handlers/federation.py | 27 +++++++++++- synapse/storage/event_federation.py | 86 ++++++++++++++++++++++++++++++++++++- 3 files changed, 114 insertions(+), 5 deletions(-) (limited to 'synapse/storage/event_federation.py') diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py index 000a3081c2..1628a56294 100644 --- a/synapse/federation/replication.py +++ b/synapse/federation/replication.py @@ -181,7 +181,7 @@ class ReplicationLayer(object): @defer.inlineCallbacks @log_function - def backfill(self, dest, context, limit): + def backfill(self, dest, context, limit, extremities): """Requests some more historic PDUs for the given context from the given destination server. @@ -189,12 +189,12 @@ class ReplicationLayer(object): dest (str): The remote home server to ask. context (str): The context to backfill. limit (int): The maximum number of PDUs to return. + extremities (list): List of PDU id and origins of the first pdus + we have seen from the context Returns: Deferred: Results in the received PDUs. """ - extremities = yield self.store.get_oldest_pdus_in_context(context) - logger.debug("backfill extrem=%s", extremities) # If there are no extremeties then we've (probably) reached the start. diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index 1daeee833b..9f457ce292 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -181,7 +181,17 @@ class FederationHandler(BaseHandler): @log_function @defer.inlineCallbacks def backfill(self, dest, room_id, limit): - pdus = yield self.replication_layer.backfill(dest, room_id, limit) + extremities = yield self.store.get_oldest_events_in_room(room_id) + + pdus = yield self.replication_layer.backfill( + dest, + room_id, + limit, + extremities=[ + self.pdu_codec.decode_event_id(e) + for e in extremities + ] + ) events = [] @@ -390,6 +400,21 @@ class FederationHandler(BaseHandler): else: defer.returnValue([]) + @defer.inlineCallbacks + @log_function + def on_backfill_request(self, context, pdu_list, limit): + + events = yield self.store.get_backfill_events( + context, + [self.pdu_codec.encode_event_id(i, o) for i, o in pdu_list], + limit + ) + + defer.returnValue([ + self.pdu_codec.pdu_from_event(e) + for e in events + ]) + @log_function def _on_user_joined(self, user, room_id): waiters = self.waiting_for_join_list.get((user.to_string(), room_id), []) diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py index 88d09d9ba8..438b42c1da 100644 --- a/synapse/storage/event_federation.py +++ b/synapse/storage/event_federation.py @@ -24,6 +24,23 @@ logger = logging.getLogger(__name__) class EventFederationStore(SQLBaseStore): + def get_oldest_events_in_room(self, room_id): + return self.runInteraction( + "get_oldest_events_in_room", + self._get_oldest_events_in_room_txn, + room_id, + ) + + def _get_oldest_events_in_room_txn(self, txn, room_id): + return self._simple_select_onecol_txn( + txn, + table="event_backward_extremities", + keyvalues={ + "room_id": room_id, + }, + retcol="event_id", + ) + def get_latest_events_in_room(self, room_id): return self.runInteraction( "get_latest_events_in_room", @@ -159,4 +176,71 @@ class EventFederationStore(SQLBaseStore): "AND not events.outlier " ")" ) - txn.execute(query) \ No newline at end of file + txn.execute(query) + + + def get_backfill_events(self, room_id, event_list, limit): + """Get a list of Events for a given topic that occured before (and + including) the pdus in pdu_list. Return a list of max size `limit`. + + Args: + txn + room_id (str) + event_list (list) + limit (int) + + Return: + list: A list of PduTuples + """ + return self.runInteraction( + "get_backfill_events", + self._get_backfill_events, room_id, event_list, limit + ) + + def _get_backfill_events(self, txn, room_id, event_list, limit): + logger.debug( + "_get_backfill_events: %s, %s, %s", + room_id, repr(event_list), limit + ) + + # We seed the pdu_results with the things from the pdu_list. + event_results = event_list + + front = event_list + + query = ( + "SELECT prev_event_id FROM event_edges " + "WHERE room_id = ? AND event_id = ? " + "LIMIT ?" + ) + + # We iterate through all event_ids in `front` to select their previous + # events. These are dumped in `new_front`. + # We continue until we reach the limit *or* new_front is empty (i.e., + # we've run out of things to select + while front and len(event_results) < limit: + + new_front = [] + for event_id in front: + logger.debug( + "_backfill_interaction: id=%s", + event_id + ) + + txn.execute( + query, + (room_id, event_id, limit - len(event_results)) + ) + + for row in txn.fetchall(): + logger.debug( + "_backfill_interaction: got id=%s", + *row + ) + new_front.append(row) + + 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) -- cgit 1.4.1 From 21fe249d62deafceca05cc114d5d6bec3e815b8c Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 31 Oct 2014 10:47:34 +0000 Subject: Actually don't store any PDUs --- synapse/federation/replication.py | 27 +++++++++++++-------------- synapse/handlers/federation.py | 22 ++++++++++++++++++++++ synapse/storage/event_federation.py | 7 +++++++ 3 files changed, 42 insertions(+), 14 deletions(-) (limited to 'synapse/storage/event_federation.py') diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py index 89dbf3e2e9..a0bd2e0572 100644 --- a/synapse/federation/replication.py +++ b/synapse/federation/replication.py @@ -106,7 +106,6 @@ class ReplicationLayer(object): self.query_handlers[query_type] = handler - @defer.inlineCallbacks @log_function def send_pdu(self, pdu): """Informs the replication layer about a new PDU generated within the @@ -135,7 +134,7 @@ class ReplicationLayer(object): logger.debug("[%s] Persisting PDU", pdu.pdu_id) # Save *before* trying to send - yield self.store.persist_event(pdu=pdu) + # yield self.store.persist_event(pdu=pdu) logger.debug("[%s] Persisted PDU", pdu.pdu_id) logger.debug("[%s] transaction_layer.enqueue_pdu... ", pdu.pdu_id) @@ -359,12 +358,13 @@ class ReplicationLayer(object): pdu_id, pdu_origin ) else: - results = yield self.store.get_current_state_for_context( - context - ) - pdus = [Pdu.from_pdu_tuple(p) for p in results] - - logger.debug("Context returning %d results", len(pdus)) + raise NotImplementedError("Specify an event") + # results = yield self.store.get_current_state_for_context( + # context + # ) + # pdus = [Pdu.from_pdu_tuple(p) for p in results] + # + # logger.debug("Context returning %d results", len(pdus)) defer.returnValue((200, self._transaction_from_pdus(pdus).get_dict())) @@ -456,7 +456,6 @@ class ReplicationLayer(object): defer.returnValue(pdus) - @defer.inlineCallbacks @log_function def _get_persisted_pdu(self, pdu_id, pdu_origin): """ Get a PDU from the database with given origin and id. @@ -464,9 +463,7 @@ class ReplicationLayer(object): Returns: Deferred: Results in a `Pdu`. """ - pdu_tuple = yield self.store.get_pdu(pdu_id, pdu_origin) - - defer.returnValue(Pdu.from_pdu_tuple(pdu_tuple)) + return self.handler.get_persisted_pdu(pdu_id, pdu_origin) def _transaction_from_pdus(self, pdu_list): """Returns a new Transaction containing the given PDUs suitable for @@ -502,7 +499,9 @@ class ReplicationLayer(object): # Get missing pdus if necessary. if not pdu.outlier: # We only backfill backwards to the min depth. - min_depth = yield self.store.get_min_depth_for_context(pdu.context) + min_depth = yield self.handler.get_min_depth_for_context( + pdu.context + ) if min_depth and pdu.depth > min_depth: for pdu_id, origin, hashes in pdu.prev_pdus: @@ -529,7 +528,7 @@ class ReplicationLayer(object): ) # Persist the Pdu, but don't mark it as processed yet. - yield self.store.persist_event(pdu=pdu) + # yield self.store.persist_event(pdu=pdu) if not backfilled: ret = yield self.handler.on_receive_pdu( diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index 9f457ce292..18cb1d4e97 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -415,6 +415,28 @@ class FederationHandler(BaseHandler): for e in events ]) + @defer.inlineCallbacks + @log_function + def get_persisted_pdu(self, pdu_id, origin): + """ Get a PDU from the database with given origin and id. + + Returns: + Deferred: Results in a `Pdu`. + """ + event = yield self.store.get_event( + self.pdu_codec.encode_event_id(pdu_id, origin), + allow_none=True, + ) + + if event: + defer.returnValue(self.pdu_codec.pdu_from_event(event)) + else: + defer.returnValue(None) + + @log_function + def get_min_depth_for_context(self, context): + return self.store.get_min_depth(context) + @log_function def _on_user_joined(self, user, room_id): waiters = self.waiting_for_join_list.get((user.to_string(), room_id), []) diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py index 438b42c1da..8357071db6 100644 --- a/synapse/storage/event_federation.py +++ b/synapse/storage/event_federation.py @@ -78,6 +78,13 @@ class EventFederationStore(SQLBaseStore): return results + def get_min_depth(self, room_id): + return self.runInteraction( + "get_min_depth", + self._get_min_depth_interaction, + room_id, + ) + def _get_min_depth_interaction(self, txn, room_id): min_depth = self._simple_select_one_onecol_txn( txn, -- cgit 1.4.1 From d30d79b5bed98c7e46852c54875c976d3ac3bc0c Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 31 Oct 2014 15:35:39 +0000 Subject: Make prev_event signing work again. --- synapse/crypto/event_signing.py | 13 ++++++++++++- synapse/storage/__init__.py | 11 +++++------ synapse/storage/event_federation.py | 2 +- 3 files changed, 18 insertions(+), 8 deletions(-) (limited to 'synapse/storage/event_federation.py') diff --git a/synapse/crypto/event_signing.py b/synapse/crypto/event_signing.py index 61edd2c6f9..07e383e221 100644 --- a/synapse/crypto/event_signing.py +++ b/synapse/crypto/event_signing.py @@ -16,11 +16,12 @@ from synapse.federation.units import Pdu -from synapse.api.events.utils import prune_pdu +from synapse.api.events.utils import prune_pdu, prune_event from syutil.jsonutil import encode_canonical_json from syutil.base64util import encode_base64, decode_base64 from syutil.crypto.jsonsign import sign_json, verify_signed_json +import copy import hashlib import logging @@ -69,6 +70,16 @@ def compute_pdu_event_reference_hash(pdu, hash_algorithm=hashlib.sha256): return (hashed.name, hashed.digest()) +def compute_event_reference_hash(event, hash_algorithm=hashlib.sha256): + tmp_event = copy.deepcopy(event) + tmp_event = prune_event(tmp_event) + event_json = tmp_event.get_dict() + event_json.pop("signatures", None) + event_json_bytes = encode_canonical_json(event_json) + hashed = hash_algorithm(event_json_bytes) + return (hashed.name, hashed.digest()) + + def sign_event_pdu(pdu, signature_name, signing_key): tmp_pdu = Pdu(**pdu.get_dict()) tmp_pdu = prune_pdu(tmp_pdu) diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index c2560f6045..31a0022d54 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -46,7 +46,7 @@ from .signatures import SignatureStore from syutil.base64util import decode_base64 -from synapse.crypto.event_signing import compute_pdu_event_reference_hash +from synapse.crypto.event_signing import compute_event_reference_hash import json @@ -271,11 +271,10 @@ class DataStore(RoomMemberStore, RoomStore, txn, event.event_id, prev_event_id, alg, hash_bytes ) - # 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_event_reference_hash(event) + 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) diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py index 8357071db6..dcc116bad2 100644 --- a/synapse/storage/event_federation.py +++ b/synapse/storage/event_federation.py @@ -69,7 +69,7 @@ class EventFederationStore(SQLBaseStore): results = [] for event_id, depth in txn.fetchall(): - hashes = self._get_prev_event_hashes_txn(txn, event_id) + 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" -- cgit 1.4.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/event_federation.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.4.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/event_federation.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.4.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/event_federation.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.4.1 From 8b0e96474b3a18866f28f112de7e56979b401768 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 7 Nov 2014 10:42:31 +0000 Subject: Implement method to get auth_chain from a given event_id --- synapse/storage/event_federation.py | 35 +++++++++++++++++++++++++++++++++++ 1 file changed, 35 insertions(+) (limited to 'synapse/storage/event_federation.py') diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py index 86c68ebf87..7140ea3d57 100644 --- a/synapse/storage/event_federation.py +++ b/synapse/storage/event_federation.py @@ -24,6 +24,41 @@ logger = logging.getLogger(__name__) class EventFederationStore(SQLBaseStore): + def get_auth_chain(self, event_id): + return self.runInteraction( + "get_auth_chain", + self._get_auth_chain_txn, + event_id + ) + + def _get_auth_chain_txn(self, txn, event_id): + results = set([event_id]) + + front = set([event_id]) + while front: + for ev_id in front: + new_front = set() + auth_ids = self._simple_select_onecol_txn( + txn, + table="event_auth", + keyvalues={ + "event_id": ev_id, + }, + retcol="auth_id", + ) + + new_front.update(auth_ids) + front = new_front + new_front.clear() + + sql = "SELECT * FROM events WHERE event_id = ?" + rows = [] + for ev_id in results: + c = txn.execute(sql, (ev_id,)) + rows.extend(self.cursor_to_dict(c)) + + return self._parse_events_txn(txn, rows) + def get_oldest_events_in_room(self, room_id): return self.runInteraction( "get_oldest_events_in_room", -- cgit 1.4.1 From 16a0815fac750863588c3c1f72c5e445d14bbf43 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 7 Nov 2014 11:21:20 +0000 Subject: Fix bug in _get_auth_chain_txn --- synapse/storage/event_federation.py | 25 +++++++++++-------------- 1 file changed, 11 insertions(+), 14 deletions(-) (limited to 'synapse/storage/event_federation.py') diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py index 7140ea3d57..d66a49e9f2 100644 --- a/synapse/storage/event_federation.py +++ b/synapse/storage/event_federation.py @@ -32,24 +32,21 @@ class EventFederationStore(SQLBaseStore): ) def _get_auth_chain_txn(self, txn, event_id): - results = set([event_id]) + results = set() + + base_sql = ( + "SELECT auth_id FROM event_auth WHERE %s" + ) front = set([event_id]) while front: - for ev_id in front: - new_front = set() - auth_ids = self._simple_select_onecol_txn( - txn, - table="event_auth", - keyvalues={ - "event_id": ev_id, - }, - retcol="auth_id", - ) + sql = base_sql % ( + " OR ".join(["event_id=?"] * len(front)), + ) - new_front.update(auth_ids) - front = new_front - new_front.clear() + txn.execute(sql, list(front)) + front = [r[0] for r in txn.fetchall()] + results.update(front) sql = "SELECT * FROM events WHERE event_id = ?" rows = [] -- cgit 1.4.1 From 02c3b1c9e2afa27753e9ce898e5455b6489541b4 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 7 Nov 2014 15:35:53 +0000 Subject: Add '/event_auth/' federation api --- synapse/federation/replication.py | 5 +++++ synapse/federation/transport.py | 26 ++++++++++++++++++++++++++ synapse/handlers/federation.py | 5 +++++ synapse/storage/event_federation.py | 26 +++++++++++++++++++------- 4 files changed, 55 insertions(+), 7 deletions(-) (limited to 'synapse/storage/event_federation.py') diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py index e358de942e..719bfcc42c 100644 --- a/synapse/federation/replication.py +++ b/synapse/federation/replication.py @@ -426,6 +426,11 @@ class ReplicationLayer(object): "auth_chain": [p.get_dict() for p in res_pdus["auth_chain"]], })) + @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.inlineCallbacks def make_join(self, destination, context, user_id): pdu_dict = yield self.transport_layer.make_join( diff --git a/synapse/federation/transport.py b/synapse/federation/transport.py index b9f7d54c71..babe8447eb 100644 --- a/synapse/federation/transport.py +++ b/synapse/federation/transport.py @@ -256,6 +256,21 @@ class TransportLayer(object): defer.returnValue(json.loads(content)) + @defer.inlineCallbacks + @log_function + def get_event_auth(self, destination, context, event_id): + path = PREFIX + "/event_auth/%s/%s" % ( + context, + event_id, + ) + + response = yield self.client.get_json( + destination=destination, + path=path, + ) + + defer.returnValue(response) + @defer.inlineCallbacks def _authenticate_request(self, request): json_request = { @@ -426,6 +441,17 @@ class TransportLayer(object): ) ) + self.server.register_path( + "GET", + re.compile("^" + PREFIX + "/event_auth/([^/]*)/([^/]*)$"), + self._with_authentication( + lambda origin, content, query, context, event_id: + handler.on_event_auth( + origin, context, event_id, + ) + ) + ) + self.server.register_path( "PUT", re.compile("^" + PREFIX + "/send_join/([^/]*)/([^/]*)$"), diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index e6afd95a58..ce65bbcd62 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -224,6 +224,11 @@ class FederationHandler(BaseHandler): defer.returnValue(self.pdu_codec.event_from_pdu(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]) + @log_function @defer.inlineCallbacks def do_invite_join(self, target_host, room_id, joinee, content, snapshot): diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py index d66a49e9f2..06e32d592d 100644 --- a/synapse/storage/event_federation.py +++ b/synapse/storage/event_federation.py @@ -32,6 +32,24 @@ class EventFederationStore(SQLBaseStore): ) def _get_auth_chain_txn(self, txn, event_id): + results = self._get_auth_chain_ids_txn(txn, event_id) + + sql = "SELECT * FROM events WHERE event_id = ?" + rows = [] + for ev_id in results: + c = txn.execute(sql, (ev_id,)) + rows.extend(self.cursor_to_dict(c)) + + return self._parse_events_txn(txn, rows) + + def get_auth_chain_ids(self, event_id): + return self.runInteraction( + "get_auth_chain_ids", + self._get_auth_chain_ids_txn, + event_id + ) + + def _get_auth_chain_ids_txn(self, txn, event_id): results = set() base_sql = ( @@ -48,13 +66,7 @@ class EventFederationStore(SQLBaseStore): front = [r[0] for r in txn.fetchall()] results.update(front) - sql = "SELECT * FROM events WHERE event_id = ?" - rows = [] - for ev_id in results: - c = txn.execute(sql, (ev_id,)) - rows.extend(self.cursor_to_dict(c)) - - return self._parse_events_txn(txn, rows) + return list(results) def get_oldest_events_in_room(self, room_id): return self.runInteraction( -- cgit 1.4.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/event_federation.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.4.1 From 5d439b127ba34b951dfd09a7d3c684c2d50df702 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Mon, 10 Nov 2014 13:46:44 +0000 Subject: PEP8 --- synapse/api/auth.py | 3 +-- synapse/api/events/room.py | 1 + synapse/federation/replication.py | 1 - synapse/federation/transport.py | 9 ++++++--- synapse/federation/units.py | 7 +++---- synapse/handlers/federation.py | 5 ++++- synapse/storage/__init__.py | 7 ++++--- synapse/storage/event_federation.py | 9 +++------ 8 files changed, 22 insertions(+), 20 deletions(-) (limited to 'synapse/storage/event_federation.py') diff --git a/synapse/api/auth.py b/synapse/api/auth.py index 48f9d460a3..a5c6964707 100644 --- a/synapse/api/auth.py +++ b/synapse/api/auth.py @@ -369,7 +369,6 @@ class Auth(object): ] event.auth_events = zip(auth_events, hashes) - @log_function def _can_send_event(self, event): key = (RoomPowerLevelsEvent.TYPE, "", ) @@ -452,7 +451,7 @@ class Auth(object): event.user_id, ) - # Check other levels: + # Check other levels: levels_to_check = [ ("users_default", []), ("events_default", []), diff --git a/synapse/api/events/room.py b/synapse/api/events/room.py index 25bc883706..8c4ac45d02 100644 --- a/synapse/api/events/room.py +++ b/synapse/api/events/room.py @@ -153,6 +153,7 @@ class RoomPowerLevelsEvent(SynapseStateEvent): def get_content_template(self): return {} + class RoomAliasesEvent(SynapseStateEvent): TYPE = "m.room.aliases" diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py index e798304353..bacba36755 100644 --- a/synapse/federation/replication.py +++ b/synapse/federation/replication.py @@ -549,7 +549,6 @@ class ReplicationLayer(object): origin, pdu.room_id, pdu.event_id, ) - if not backfilled: ret = yield self.handler.on_receive_pdu( pdu, diff --git a/synapse/federation/transport.py b/synapse/federation/transport.py index d84a44c211..95c40c6c1b 100644 --- a/synapse/federation/transport.py +++ b/synapse/federation/transport.py @@ -284,7 +284,7 @@ class TransportLayer(object): origin = None if request.method == "PUT": - #TODO: Handle other method types? other content types? + # TODO: Handle other method types? other content types? try: content_bytes = request.content.read() content = json.loads(content_bytes) @@ -296,11 +296,13 @@ class TransportLayer(object): try: params = auth.split(" ")[1].split(",") param_dict = dict(kv.split("=") for kv in params) + def strip_quotes(value): if value.startswith("\""): return value[1:-1] else: return value + origin = strip_quotes(param_dict["origin"]) key = strip_quotes(param_dict["key"]) sig = strip_quotes(param_dict["sig"]) @@ -321,7 +323,7 @@ class TransportLayer(object): if auth.startswith("X-Matrix"): (origin, key, sig) = parse_auth_header(auth) json_request["origin"] = origin - json_request["signatures"].setdefault(origin,{})[key] = sig + json_request["signatures"].setdefault(origin, {})[key] = sig if not json_request["signatures"]: raise SynapseError( @@ -515,7 +517,8 @@ class TransportLayer(object): return try: - code, response = yield self.received_handler.on_incoming_transaction( + handler = self.received_handler + code, response = yield handler.on_incoming_transaction( transaction_data ) except: diff --git a/synapse/federation/units.py b/synapse/federation/units.py index d98014cac7..f4e7b62bd9 100644 --- a/synapse/federation/units.py +++ b/synapse/federation/units.py @@ -192,7 +192,9 @@ class Transaction(JsonEncodedObject): transaction_id and origin_server_ts keys. """ if "origin_server_ts" not in kwargs: - raise KeyError("Require 'origin_server_ts' to construct a Transaction") + raise KeyError( + "Require 'origin_server_ts' to construct a Transaction" + ) if "transaction_id" not in kwargs: raise KeyError( "Require 'transaction_id' to construct a Transaction" @@ -204,6 +206,3 @@ class Transaction(JsonEncodedObject): kwargs["pdus"] = [p.get_dict() for p in pdus] return Transaction(**kwargs) - - - diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index 587fa308c8..e909af6bd8 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -521,6 +521,9 @@ class FederationHandler(BaseHandler): @log_function def _on_user_joined(self, user, room_id): - waiters = self.waiting_for_join_list.get((user.to_string(), room_id), []) + waiters = self.waiting_for_join_list.get( + (user.to_string(), room_id), + [] + ) while waiters: waiters.pop().callback(None) diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index 7d810e6a62..4034437f6b 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -494,11 +494,13 @@ def prepare_database(db_conn): user_version = row[0] if user_version > SCHEMA_VERSION: - raise ValueError("Cannot use this database as it is too " + + raise ValueError( + "Cannot use this database as it is too " + "new for the server to understand" ) elif user_version < SCHEMA_VERSION: - logging.info("Upgrading database from version %d", + logging.info( + "Upgrading database from version %d", user_version ) @@ -520,4 +522,3 @@ def prepare_database(db_conn): c.execute("PRAGMA user_version = %d" % SCHEMA_VERSION) c.close() - diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py index a707030145..a027db3868 100644 --- a/synapse/storage/event_federation.py +++ b/synapse/storage/event_federation.py @@ -215,7 +215,7 @@ class EventFederationStore(SQLBaseStore): min_depth = self._simple_select_one_onecol_txn( txn, table="room_depth", - keyvalues={"room_id": room_id,}, + keyvalues={"room_id": room_id}, retcol="min_depth", allow_none=True, ) @@ -267,10 +267,8 @@ class EventFederationStore(SQLBaseStore): } ) - - - # We only insert as a forward extremity the new pdu if there are no - # other pdus that reference it as a prev pdu + # 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 OR IGNORE INTO %(table)s (event_id, room_id) " "SELECT ?, ? WHERE NOT EXISTS (" @@ -312,7 +310,6 @@ class EventFederationStore(SQLBaseStore): ) txn.execute(query) - def get_backfill_events(self, room_id, event_list, limit): """Get a list of Events for a given topic that occured before (and including) the pdus in pdu_list. Return a list of max size `limit`. -- cgit 1.4.1 From e715741abc42ed39ecaf38048f7b41e1076bfb70 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Wed, 12 Nov 2014 15:02:31 +0000 Subject: Update some of the docs in event_federation --- synapse/storage/event_federation.py | 31 ++++++++++++++++++++----------- 1 file changed, 20 insertions(+), 11 deletions(-) (limited to 'synapse/storage/event_federation.py') diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py index a027db3868..6c559f8f63 100644 --- a/synapse/storage/event_federation.py +++ b/synapse/storage/event_federation.py @@ -23,6 +23,14 @@ logger = logging.getLogger(__name__) class EventFederationStore(SQLBaseStore): + """ Responsible for storing and serving up the various graphs associated + with an event. Including the main event graph and the auth chains for an + event. + + Also has methods for getting the front (latest) and back (oldest) edges + of the event graphs. These are used to generate the parents for new events + and backfilling from another server respectively. + """ def get_auth_chain(self, event_id): return self.runInteraction( @@ -205,6 +213,8 @@ class EventFederationStore(SQLBaseStore): return results def get_min_depth(self, room_id): + """ For hte given room, get the minimum depth we have seen for it. + """ return self.runInteraction( "get_min_depth", self._get_min_depth_interaction, @@ -240,6 +250,10 @@ class EventFederationStore(SQLBaseStore): def _handle_prev_events(self, txn, outlier, event_id, prev_events, room_id): + """ + For the given event, update the event edges table and forward and + backward extremities tables. + """ for e_id, _ in prev_events: # TODO (erikj): This could be done as a bulk insert self._simple_insert_txn( @@ -267,8 +281,8 @@ class EventFederationStore(SQLBaseStore): } ) - # We only insert as a forward extremity the new pdu if there are - # no other pdus that reference it as a prev pdu + # We only insert as a forward extremity the new event if there are + # no other events that reference it as a prev event query = ( "INSERT OR IGNORE INTO %(table)s (event_id, room_id) " "SELECT ?, ? WHERE NOT EXISTS (" @@ -284,7 +298,7 @@ class EventFederationStore(SQLBaseStore): txn.execute(query, (event_id, room_id, event_id)) - # Insert all the prev_pdus as a backwards thing, they'll get + # Insert all the prev_events 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 @@ -299,7 +313,7 @@ class EventFederationStore(SQLBaseStore): ) # Also delete from the backwards extremities table all ones that - # reference pdus that we have already seen + # reference events that we have already seen query = ( "DELETE FROM event_backward_extremities WHERE EXISTS (" "SELECT 1 FROM events " @@ -311,17 +325,14 @@ class EventFederationStore(SQLBaseStore): txn.execute(query) def get_backfill_events(self, room_id, event_list, limit): - """Get a list of Events for a given topic that occured before (and - including) the pdus in pdu_list. Return a list of max size `limit`. + """Get a list of Events for a given topic that occurred before (and + including) the events in event_list. Return a list of max size `limit` Args: txn room_id (str) event_list (list) limit (int) - - Return: - list: A list of PduTuples """ return self.runInteraction( "get_backfill_events", @@ -334,7 +345,6 @@ class EventFederationStore(SQLBaseStore): room_id, repr(event_list), limit ) - # We seed the pdu_results with the things from the pdu_list. event_results = event_list front = event_list @@ -373,5 +383,4 @@ class EventFederationStore(SQLBaseStore): front = new_front event_results += new_front - # We also want to update the `prev_pdus` attributes before returning. return self._get_events_txn(txn, event_results) -- cgit 1.4.1