Move storage classes into a main "data store".

This is in preparation for having multiple data stores that offer
different functionality, e.g. splitting out state or event storage.

1 files changed, 0 insertions, 672 deletions

diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py
deleted file mode 100644
index 47cc10d32a..0000000000
--- a/synapse/storage/event_federation.py
+++ /dev/null
@@ -1,672 +0,0 @@
-# -*- coding: utf-8 -*-
-# Copyright 2014-2016 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.
-import itertools
-import logging
-import random
-
-from six.moves import range
-from six.moves.queue import Empty, PriorityQueue
-
-from unpaddedbase64 import encode_base64
-
-from twisted.internet import defer
-
-from synapse.api.errors import StoreError
-from synapse.metrics.background_process_metrics import run_as_background_process
-from synapse.storage._base import SQLBaseStore, make_in_list_sql_clause
-from synapse.storage.events_worker import EventsWorkerStore
-from synapse.storage.signatures import SignatureWorkerStore
-from synapse.util.caches.descriptors import cached
-
-logger = logging.getLogger(__name__)
-
-
-class EventFederationWorkerStore(EventsWorkerStore, SignatureWorkerStore, SQLBaseStore):
-    def get_auth_chain(self, event_ids, include_given=False):
-        """Get auth events for given event_ids. The events *must* be state events.
-
-        Args:
-            event_ids (list): state events
-            include_given (bool): include the given events in result
-
-        Returns:
-            list of events
-        """
-        return self.get_auth_chain_ids(
-            event_ids, include_given=include_given
-        ).addCallback(self.get_events_as_list)
-
-    def get_auth_chain_ids(self, event_ids, include_given=False):
-        """Get auth events for given event_ids. The events *must* be state events.
-
-        Args:
-            event_ids (list): state events
-            include_given (bool): include the given events in result
-
-        Returns:
-            list of event_ids
-        """
-        return self.runInteraction(
-            "get_auth_chain_ids", self._get_auth_chain_ids_txn, event_ids, include_given
-        )
-
-    def _get_auth_chain_ids_txn(self, txn, event_ids, include_given):
-        if include_given:
-            results = set(event_ids)
-        else:
-            results = set()
-
-        base_sql = "SELECT auth_id FROM event_auth WHERE "
-
-        front = set(event_ids)
-        while front:
-            new_front = set()
-            front_list = list(front)
-            chunks = [front_list[x : x + 100] for x in range(0, len(front), 100)]
-            for chunk in chunks:
-                clause, args = make_in_list_sql_clause(
-                    txn.database_engine, "event_id", chunk
-                )
-                txn.execute(base_sql + clause, list(args))
-                new_front.update([r[0] for r in txn])
-
-            new_front -= results
-
-            front = new_front
-            results.update(front)
-
-        return list(results)
-
-    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_with_depth_in_room(self, room_id):
-        return self.runInteraction(
-            "get_oldest_events_with_depth_in_room",
-            self.get_oldest_events_with_depth_in_room_txn,
-            room_id,
-        )
-
-    def get_oldest_events_with_depth_in_room_txn(self, txn, room_id):
-        sql = (
-            "SELECT b.event_id, MAX(e.depth) FROM events as e"
-            " INNER JOIN event_edges as g"
-            " ON g.event_id = e.event_id"
-            " INNER JOIN event_backward_extremities as b"
-            " ON g.prev_event_id = b.event_id"
-            " WHERE b.room_id = ? AND g.is_state is ?"
-            " GROUP BY b.event_id"
-        )
-
-        txn.execute(sql, (room_id, False))
-
-        return dict(txn)
-
-    @defer.inlineCallbacks
-    def get_max_depth_of(self, event_ids):
-        """Returns the max depth of a set of event IDs
-
-        Args:
-            event_ids (list[str])
-
-        Returns
-            Deferred[int]
-        """
-        rows = yield self._simple_select_many_batch(
-            table="events",
-            column="event_id",
-            iterable=event_ids,
-            retcols=("depth",),
-            desc="get_max_depth_of",
-        )
-
-        if not rows:
-            return 0
-        else:
-            return max(row["depth"] for row in rows)
-
-    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",
-        )
-
-    @defer.inlineCallbacks
-    def get_prev_events_for_room(self, room_id):
-        """
-        Gets a subset of the current forward extremities in the given room.
-
-        Limits the result to 10 extremities, so that we can avoid creating
-        events which refer to hundreds of prev_events.
-
-        Args:
-            room_id (str): room_id
-
-        Returns:
-            Deferred[list[(str, dict[str, str], int)]]
-                for each event, a tuple of (event_id, hashes, depth)
-                where *hashes* is a map from algorithm to hash.
-        """
-        res = yield self.get_latest_event_ids_and_hashes_in_room(room_id)
-        if len(res) > 10:
-            # Sort by reverse depth, so we point to the most recent.
-            res.sort(key=lambda a: -a[2])
-
-            # we use half of the limit for the actual most recent events, and
-            # the other half to randomly point to some of the older events, to
-            # make sure that we don't completely ignore the older events.
-            res = res[0:5] + random.sample(res[5:], 5)
-
-        return res
-
-    def get_latest_event_ids_and_hashes_in_room(self, room_id):
-        """
-        Gets the current forward extremities in the given room
-
-        Args:
-            room_id (str): room_id
-
-        Returns:
-            Deferred[list[(str, dict[str, str], int)]]
-                for each event, a tuple of (event_id, hashes, depth)
-                where *hashes* is a map from algorithm to hash.
-        """
-
-        return self.runInteraction(
-            "get_latest_event_ids_and_hashes_in_room",
-            self._get_latest_event_ids_and_hashes_in_room,
-            room_id,
-        )
-
-    def get_rooms_with_many_extremities(self, min_count, limit, room_id_filter):
-        """Get the top rooms with at least N extremities.
-
-        Args:
-            min_count (int): The minimum number of extremities
-            limit (int): The maximum number of rooms to return.
-            room_id_filter (iterable[str]): room_ids to exclude from the results
-
-        Returns:
-            Deferred[list]: At most `limit` room IDs that have at least
-            `min_count` extremities, sorted by extremity count.
-        """
-
-        def _get_rooms_with_many_extremities_txn(txn):
-            where_clause = "1=1"
-            if room_id_filter:
-                where_clause = "room_id NOT IN (%s)" % (
-                    ",".join("?" for _ in room_id_filter),
-                )
-
-            sql = """
-                SELECT room_id FROM event_forward_extremities
-                WHERE %s
-                GROUP BY room_id
-                HAVING count(*) > ?
-                ORDER BY count(*) DESC
-                LIMIT ?
-            """ % (
-                where_clause,
-            )
-
-            query_args = list(itertools.chain(room_id_filter, [min_count, limit]))
-            txn.execute(sql, query_args)
-            return [room_id for room_id, in txn]
-
-        return self.runInteraction(
-            "get_rooms_with_many_extremities", _get_rooms_with_many_extremities_txn
-        )
-
-    @cached(max_entries=5000, iterable=True)
-    def get_latest_event_ids_in_room(self, room_id):
-        return self._simple_select_onecol(
-            table="event_forward_extremities",
-            keyvalues={"room_id": room_id},
-            retcol="event_id",
-            desc="get_latest_event_ids_in_room",
-        )
-
-    def _get_latest_event_ids_and_hashes_in_room(self, txn, room_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 "
-            "AND e.room_id = f.room_id "
-            "WHERE f.room_id = ?"
-        )
-
-        txn.execute(sql, (room_id,))
-
-        results = []
-        for event_id, depth in txn.fetchall():
-            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"
-            }
-            results.append((event_id, prev_hashes, depth))
-
-        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, room_id
-        )
-
-    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 get_forward_extremeties_for_room(self, room_id, stream_ordering):
-        """For a given room_id and stream_ordering, return the forward
-        extremeties of the room at that point in "time".
-
-        Throws a StoreError if we have since purged the index for
-        stream_orderings from that point.
-
-        Args:
-            room_id (str):
-            stream_ordering (int):
-
-        Returns:
-            deferred, which resolves to a list of event_ids
-        """
-        # We want to make the cache more effective, so we clamp to the last
-        # change before the given ordering.
-        last_change = self._events_stream_cache.get_max_pos_of_last_change(room_id)
-
-        # We don't always have a full stream_to_exterm_id table, e.g. after
-        # the upgrade that introduced it, so we make sure we never ask for a
-        # stream_ordering from before a restart
-        last_change = max(self._stream_order_on_start, last_change)
-
-        # provided the last_change is recent enough, we now clamp the requested
-        # stream_ordering to it.
-        if last_change > self.stream_ordering_month_ago:
-            stream_ordering = min(last_change, stream_ordering)
-
-        return self._get_forward_extremeties_for_room(room_id, stream_ordering)
-
-    @cached(max_entries=5000, num_args=2)
-    def _get_forward_extremeties_for_room(self, room_id, stream_ordering):
-        """For a given room_id and stream_ordering, return the forward
-        extremeties of the room at that point in "time".
-
-        Throws a StoreError if we have since purged the index for
-        stream_orderings from that point.
-        """
-
-        if stream_ordering <= self.stream_ordering_month_ago:
-            raise StoreError(400, "stream_ordering too old")
-
-        sql = """
-                SELECT event_id FROM stream_ordering_to_exterm
-                INNER JOIN (
-                    SELECT room_id, MAX(stream_ordering) AS stream_ordering
-                    FROM stream_ordering_to_exterm
-                    WHERE stream_ordering <= ? GROUP BY room_id
-                ) AS rms USING (room_id, stream_ordering)
-                WHERE room_id = ?
-        """
-
-        def get_forward_extremeties_for_room_txn(txn):
-            txn.execute(sql, (stream_ordering, room_id))
-            return [event_id for event_id, in txn]
-
-        return self.runInteraction(
-            "get_forward_extremeties_for_room", get_forward_extremeties_for_room_txn
-        )
-
-    def get_backfill_events(self, room_id, event_list, 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 (
-            self.runInteraction(
-                "get_backfill_events",
-                self._get_backfill_events,
-                room_id,
-                event_list,
-                limit,
-            )
-            .addCallback(self.get_events_as_list)
-            .addCallback(lambda l: sorted(l, key=lambda e: -e.depth))
-        )
-
-    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
-        )
-
-        event_results = set()
-
-        # We want to make sure that we do a breadth-first, "depth" ordered
-        # search.
-
-        query = (
-            "SELECT depth, prev_event_id FROM event_edges"
-            " INNER JOIN events"
-            " ON prev_event_id = events.event_id"
-            " WHERE event_edges.event_id = ?"
-            " AND event_edges.is_state = ?"
-            " LIMIT ?"
-        )
-
-        queue = PriorityQueue()
-
-        for event_id in event_list:
-            depth = self._simple_select_one_onecol_txn(
-                txn,
-                table="events",
-                keyvalues={"event_id": event_id, "room_id": room_id},
-                retcol="depth",
-                allow_none=True,
-            )
-
-            if depth:
-                queue.put((-depth, event_id))
-
-        while not queue.empty() and len(event_results) < limit:
-            try:
-                _, event_id = queue.get_nowait()
-            except Empty:
-                break
-
-            if event_id in event_results:
-                continue
-
-            event_results.add(event_id)
-
-            txn.execute(query, (event_id, False, limit - len(event_results)))
-
-            for row in txn:
-                if row[1] not in event_results:
-                    queue.put((-row[0], row[1]))
-
-        return event_results
-
-    @defer.inlineCallbacks
-    def get_missing_events(self, room_id, earliest_events, latest_events, limit):
-        ids = yield self.runInteraction(
-            "get_missing_events",
-            self._get_missing_events,
-            room_id,
-            earliest_events,
-            latest_events,
-            limit,
-        )
-        events = yield self.get_events_as_list(ids)
-        return events
-
-    def _get_missing_events(self, txn, room_id, earliest_events, latest_events, limit):
-
-        seen_events = set(earliest_events)
-        front = set(latest_events) - seen_events
-        event_results = []
-
-        query = (
-            "SELECT prev_event_id FROM event_edges "
-            "WHERE room_id = ? AND event_id = ? AND is_state = ? "
-            "LIMIT ?"
-        )
-
-        while front and len(event_results) < limit:
-            new_front = set()
-            for event_id in front:
-                txn.execute(
-                    query, (room_id, event_id, False, limit - len(event_results))
-                )
-
-                new_results = set(t[0] for t in txn) - seen_events
-
-                new_front |= new_results
-                seen_events |= new_results
-                event_results.extend(new_results)
-
-            front = new_front
-
-        # we built the list working backwards from latest_events; we now need to
-        # reverse it so that the events are approximately chronological.
-        event_results.reverse()
-        return event_results
-
-    @defer.inlineCallbacks
-    def get_successor_events(self, event_ids):
-        """Fetch all events that have the given events as a prev event
-
-        Args:
-            event_ids (iterable[str])
-
-        Returns:
-            Deferred[list[str]]
-        """
-        rows = yield self._simple_select_many_batch(
-            table="event_edges",
-            column="prev_event_id",
-            iterable=event_ids,
-            retcols=("event_id",),
-            desc="get_successor_events",
-        )
-
-        return [row["event_id"] for row in rows]
-
-
-class EventFederationStore(EventFederationWorkerStore):
-    """ 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.
-    """
-
-    EVENT_AUTH_STATE_ONLY = "event_auth_state_only"
-
-    def __init__(self, db_conn, hs):
-        super(EventFederationStore, self).__init__(db_conn, hs)
-
-        self.register_background_update_handler(
-            self.EVENT_AUTH_STATE_ONLY, self._background_delete_non_state_event_auth
-        )
-
-        hs.get_clock().looping_call(
-            self._delete_old_forward_extrem_cache, 60 * 60 * 1000
-        )
-
-    def _update_min_depth_for_room_txn(self, txn, room_id, depth):
-        min_depth = self._get_min_depth_interaction(txn, room_id)
-
-        if min_depth and depth >= min_depth:
-            return
-
-        self._simple_upsert_txn(
-            txn,
-            table="room_depth",
-            keyvalues={"room_id": room_id},
-            values={"min_depth": depth},
-        )
-
-    def _handle_mult_prev_events(self, txn, events):
-        """
-        For the given event, update the event edges table and forward and
-        backward extremities tables.
-        """
-        self._simple_insert_many_txn(
-            txn,
-            table="event_edges",
-            values=[
-                {
-                    "event_id": ev.event_id,
-                    "prev_event_id": e_id,
-                    "room_id": ev.room_id,
-                    "is_state": False,
-                }
-                for ev in events
-                for e_id in ev.prev_event_ids()
-            ],
-        )
-
-        self._update_backward_extremeties(txn, events)
-
-    def _update_backward_extremeties(self, txn, events):
-        """Updates the event_backward_extremities tables based on the new/updated
-        events being persisted.
-
-        This is called for new events *and* for events that were outliers, but
-        are now being persisted as non-outliers.
-
-        Forward extremities are handled when we first start persisting the events.
-        """
-        events_by_room = {}
-        for ev in events:
-            events_by_room.setdefault(ev.room_id, []).append(ev)
-
-        query = (
-            "INSERT INTO event_backward_extremities (event_id, room_id)"
-            " SELECT ?, ? WHERE NOT EXISTS ("
-            " SELECT 1 FROM event_backward_extremities"
-            " WHERE event_id = ? AND room_id = ?"
-            " )"
-            " AND NOT EXISTS ("
-            " SELECT 1 FROM events WHERE event_id = ? AND room_id = ? "
-            " AND outlier = ?"
-            " )"
-        )
-
-        txn.executemany(
-            query,
-            [
-                (e_id, ev.room_id, e_id, ev.room_id, e_id, ev.room_id, False)
-                for ev in events
-                for e_id in ev.prev_event_ids()
-                if not ev.internal_metadata.is_outlier()
-            ],
-        )
-
-        query = (
-            "DELETE FROM event_backward_extremities"
-            " WHERE event_id = ? AND room_id = ?"
-        )
-        txn.executemany(
-            query,
-            [
-                (ev.event_id, ev.room_id)
-                for ev in events
-                if not ev.internal_metadata.is_outlier()
-            ],
-        )
-
-    def _delete_old_forward_extrem_cache(self):
-        def _delete_old_forward_extrem_cache_txn(txn):
-            # Delete entries older than a month, while making sure we don't delete
-            # the only entries for a room.
-            sql = """
-                DELETE FROM stream_ordering_to_exterm
-                WHERE
-                room_id IN (
-                    SELECT room_id
-                    FROM stream_ordering_to_exterm
-                    WHERE stream_ordering > ?
-                ) AND stream_ordering < ?
-            """
-            txn.execute(
-                sql, (self.stream_ordering_month_ago, self.stream_ordering_month_ago)
-            )
-
-        return run_as_background_process(
-            "delete_old_forward_extrem_cache",
-            self.runInteraction,
-            "_delete_old_forward_extrem_cache",
-            _delete_old_forward_extrem_cache_txn,
-        )
-
-    def clean_room_for_join(self, room_id):
-        return self.runInteraction(
-            "clean_room_for_join", self._clean_room_for_join_txn, room_id
-        )
-
-    def _clean_room_for_join_txn(self, txn, room_id):
-        query = "DELETE FROM event_forward_extremities WHERE room_id = ?"
-
-        txn.execute(query, (room_id,))
-        txn.call_after(self.get_latest_event_ids_in_room.invalidate, (room_id,))
-
-    @defer.inlineCallbacks
-    def _background_delete_non_state_event_auth(self, progress, batch_size):
-        def delete_event_auth(txn):
-            target_min_stream_id = progress.get("target_min_stream_id_inclusive")
-            max_stream_id = progress.get("max_stream_id_exclusive")
-
-            if not target_min_stream_id or not max_stream_id:
-                txn.execute("SELECT COALESCE(MIN(stream_ordering), 0) FROM events")
-                rows = txn.fetchall()
-                target_min_stream_id = rows[0][0]
-
-                txn.execute("SELECT COALESCE(MAX(stream_ordering), 0) FROM events")
-                rows = txn.fetchall()
-                max_stream_id = rows[0][0]
-
-            min_stream_id = max_stream_id - batch_size
-
-            sql = """
-                DELETE FROM event_auth
-                WHERE event_id IN (
-                    SELECT event_id FROM events
-                    LEFT JOIN state_events USING (room_id, event_id)
-                    WHERE ? <= stream_ordering AND stream_ordering < ?
-                        AND state_key IS null
-                )
-            """
-
-            txn.execute(sql, (min_stream_id, max_stream_id))
-
-            new_progress = {
-                "target_min_stream_id_inclusive": target_min_stream_id,
-                "max_stream_id_exclusive": min_stream_id,
-            }
-
-            self._background_update_progress_txn(
-                txn, self.EVENT_AUTH_STATE_ONLY, new_progress
-            )
-
-            return min_stream_id >= target_min_stream_id
-
-        result = yield self.runInteraction(
-            self.EVENT_AUTH_STATE_ONLY, delete_event_auth
-        )
-
-        if not result:
-            yield self._end_background_update(self.EVENT_AUTH_STATE_ONLY)
-
-        return batch_size