2014-10-28 16:42:35 +00:00
|
|
|
# -*- coding: utf-8 -*-
|
2016-01-07 04:26:29 +00:00
|
|
|
# Copyright 2014-2016 OpenMarket Ltd
|
2014-10-28 16:42:35 +00:00
|
|
|
#
|
|
|
|
# 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.
|
2019-09-26 10:47:53 +00:00
|
|
|
import itertools
|
2018-07-09 06:09:20 +00:00
|
|
|
import logging
|
2020-02-19 15:04:47 +00:00
|
|
|
from typing import List, Optional, Set
|
2014-10-28 16:42:35 +00:00
|
|
|
|
2018-07-09 06:09:20 +00:00
|
|
|
from six.moves import range
|
|
|
|
from six.moves.queue import Empty, PriorityQueue
|
|
|
|
|
2015-05-14 12:45:48 +00:00
|
|
|
from twisted.internet import defer
|
|
|
|
|
2018-07-09 06:09:20 +00:00
|
|
|
from synapse.api.errors import StoreError
|
2018-07-25 08:41:12 +00:00
|
|
|
from synapse.metrics.background_process_metrics import run_as_background_process
|
2019-10-02 18:07:07 +00:00
|
|
|
from synapse.storage._base import SQLBaseStore, make_in_list_sql_clause
|
2019-10-21 11:56:42 +00:00
|
|
|
from synapse.storage.data_stores.main.events_worker import EventsWorkerStore
|
|
|
|
from synapse.storage.data_stores.main.signatures import SignatureWorkerStore
|
2019-12-06 13:08:40 +00:00
|
|
|
from synapse.storage.database import Database
|
2015-08-11 16:59:32 +00:00
|
|
|
from synapse.util.caches.descriptors import cached
|
2014-10-28 16:42:35 +00:00
|
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
|
|
|
2019-04-03 09:07:29 +00:00
|
|
|
class EventFederationWorkerStore(EventsWorkerStore, SignatureWorkerStore, SQLBaseStore):
|
2017-05-24 13:22:41 +00:00
|
|
|
def get_auth_chain(self, event_ids, include_given=False):
|
|
|
|
"""Get auth events for given event_ids. The events *must* be state events.
|
2014-11-07 15:35:53 +00:00
|
|
|
|
2017-05-24 13:22:41 +00:00
|
|
|
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(
|
2019-04-03 09:07:29 +00:00
|
|
|
event_ids, include_given=include_given
|
2019-05-14 12:37:44 +00:00
|
|
|
).addCallback(self.get_events_as_list)
|
2017-05-24 13:22:41 +00:00
|
|
|
|
2020-02-19 15:04:47 +00:00
|
|
|
def get_auth_chain_ids(
|
|
|
|
self,
|
|
|
|
event_ids: List[str],
|
|
|
|
include_given: bool = False,
|
|
|
|
ignore_events: Optional[Set[str]] = None,
|
|
|
|
):
|
2017-05-24 13:22:41 +00:00
|
|
|
"""Get auth events for given event_ids. The events *must* be state events.
|
|
|
|
|
|
|
|
Args:
|
2020-02-19 15:04:47 +00:00
|
|
|
event_ids: state events
|
|
|
|
include_given: include the given events in result
|
|
|
|
ignore_events: Set of events to exclude from the returned auth
|
|
|
|
chain. This is useful if the caller will just discard the
|
|
|
|
given events anyway, and saves us from figuring out their auth
|
|
|
|
chains if not required.
|
2017-05-24 13:22:41 +00:00
|
|
|
|
|
|
|
Returns:
|
|
|
|
list of event_ids
|
|
|
|
"""
|
2019-12-04 13:52:46 +00:00
|
|
|
return self.db.runInteraction(
|
2020-02-19 15:04:47 +00:00
|
|
|
"get_auth_chain_ids",
|
|
|
|
self._get_auth_chain_ids_txn,
|
|
|
|
event_ids,
|
|
|
|
include_given,
|
|
|
|
ignore_events,
|
2014-11-07 15:35:53 +00:00
|
|
|
)
|
|
|
|
|
2020-02-19 15:04:47 +00:00
|
|
|
def _get_auth_chain_ids_txn(self, txn, event_ids, include_given, ignore_events):
|
|
|
|
if ignore_events is None:
|
|
|
|
ignore_events = set()
|
|
|
|
|
2020-02-19 09:39:27 +00:00
|
|
|
if include_given:
|
|
|
|
results = set(event_ids)
|
|
|
|
else:
|
|
|
|
results = set()
|
|
|
|
|
2019-10-02 18:07:07 +00:00
|
|
|
base_sql = "SELECT auth_id FROM event_auth WHERE "
|
2014-11-07 10:42:31 +00:00
|
|
|
|
2014-12-16 18:57:36 +00:00
|
|
|
front = set(event_ids)
|
2014-11-07 10:42:31 +00:00
|
|
|
while front:
|
2015-02-12 14:39:31 +00:00
|
|
|
new_front = set()
|
2015-06-23 10:01:04 +00:00
|
|
|
front_list = list(front)
|
2019-04-03 09:07:29 +00:00
|
|
|
chunks = [front_list[x : x + 100] for x in range(0, len(front), 100)]
|
2015-06-23 10:01:04 +00:00
|
|
|
for chunk in chunks:
|
2019-10-02 18:07:07 +00:00
|
|
|
clause, args = make_in_list_sql_clause(
|
|
|
|
txn.database_engine, "event_id", chunk
|
|
|
|
)
|
|
|
|
txn.execute(base_sql + clause, list(args))
|
2017-03-23 17:53:49 +00:00
|
|
|
new_front.update([r[0] for r in txn])
|
2015-02-19 17:24:14 +00:00
|
|
|
|
2020-02-19 15:04:47 +00:00
|
|
|
new_front -= ignore_events
|
2015-02-19 17:24:14 +00:00
|
|
|
new_front -= results
|
|
|
|
|
2015-02-12 14:39:31 +00:00
|
|
|
front = new_front
|
2014-11-07 11:21:20 +00:00
|
|
|
results.update(front)
|
2014-11-07 10:42:31 +00:00
|
|
|
|
2014-11-07 15:35:53 +00:00
|
|
|
return list(results)
|
2014-11-07 10:42:31 +00:00
|
|
|
|
2014-10-31 09:59:02 +00:00
|
|
|
def get_oldest_events_in_room(self, room_id):
|
2019-12-04 13:52:46 +00:00
|
|
|
return self.db.runInteraction(
|
2019-04-03 09:07:29 +00:00
|
|
|
"get_oldest_events_in_room", self._get_oldest_events_in_room_txn, room_id
|
2014-10-31 09:59:02 +00:00
|
|
|
)
|
|
|
|
|
2015-05-11 17:01:31 +00:00
|
|
|
def get_oldest_events_with_depth_in_room(self, room_id):
|
2019-12-04 13:52:46 +00:00
|
|
|
return self.db.runInteraction(
|
2015-05-11 17:01:31 +00:00
|
|
|
"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"
|
2018-07-26 12:19:08 +00:00
|
|
|
" ON g.event_id = e.event_id"
|
2015-05-11 17:01:31 +00:00
|
|
|
" INNER JOIN event_backward_extremities as b"
|
2018-07-26 12:19:08 +00:00
|
|
|
" ON g.prev_event_id = b.event_id"
|
2015-05-11 17:01:31 +00:00
|
|
|
" WHERE b.room_id = ? AND g.is_state is ?"
|
|
|
|
" GROUP BY b.event_id"
|
|
|
|
)
|
|
|
|
|
2019-04-03 09:07:29 +00:00
|
|
|
txn.execute(sql, (room_id, False))
|
2015-05-11 17:01:31 +00:00
|
|
|
|
2017-03-23 17:53:49 +00:00
|
|
|
return dict(txn)
|
2015-05-11 17:01:31 +00:00
|
|
|
|
2019-01-25 17:19:31 +00:00
|
|
|
@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]
|
|
|
|
"""
|
2019-12-04 13:52:46 +00:00
|
|
|
rows = yield self.db.simple_select_many_batch(
|
2019-01-25 17:19:31 +00:00
|
|
|
table="events",
|
|
|
|
column="event_id",
|
|
|
|
iterable=event_ids,
|
|
|
|
retcols=("depth",),
|
|
|
|
desc="get_max_depth_of",
|
|
|
|
)
|
|
|
|
|
|
|
|
if not rows:
|
2019-07-23 13:00:55 +00:00
|
|
|
return 0
|
2019-01-25 17:19:31 +00:00
|
|
|
else:
|
2019-07-23 13:00:55 +00:00
|
|
|
return max(row["depth"] for row in rows)
|
2019-01-25 17:19:31 +00:00
|
|
|
|
2014-10-31 09:59:02 +00:00
|
|
|
def _get_oldest_events_in_room_txn(self, txn, room_id):
|
2019-12-04 13:52:46 +00:00
|
|
|
return self.db.simple_select_onecol_txn(
|
2014-10-31 09:59:02 +00:00
|
|
|
txn,
|
|
|
|
table="event_backward_extremities",
|
2019-04-03 09:07:29 +00:00
|
|
|
keyvalues={"room_id": room_id},
|
2014-10-31 09:59:02 +00:00
|
|
|
retcol="event_id",
|
|
|
|
)
|
|
|
|
|
2020-01-03 16:09:24 +00:00
|
|
|
def get_prev_events_for_room(self, room_id: str):
|
|
|
|
"""
|
|
|
|
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]]: the event ids of the forward extremites
|
|
|
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
return self.db.runInteraction(
|
|
|
|
"get_prev_events_for_room", self._get_prev_events_for_room_txn, room_id
|
|
|
|
)
|
|
|
|
|
|
|
|
def _get_prev_events_for_room_txn(self, txn, room_id: str):
|
|
|
|
# we just use the 10 newest events. Older events will become
|
|
|
|
# prev_events of future events.
|
|
|
|
|
|
|
|
sql = """
|
|
|
|
SELECT e.event_id FROM event_forward_extremities AS f
|
|
|
|
INNER JOIN events AS e USING (event_id)
|
|
|
|
WHERE f.room_id = ?
|
|
|
|
ORDER BY e.depth DESC
|
|
|
|
LIMIT 10
|
|
|
|
"""
|
|
|
|
|
|
|
|
txn.execute(sql, (room_id,))
|
|
|
|
|
|
|
|
return [row[0] for row in txn]
|
|
|
|
|
2019-09-26 10:47:53 +00:00
|
|
|
def get_rooms_with_many_extremities(self, min_count, limit, room_id_filter):
|
2019-06-17 17:04:42 +00:00
|
|
|
"""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.
|
2019-09-26 10:47:53 +00:00
|
|
|
room_id_filter (iterable[str]): room_ids to exclude from the results
|
2019-06-17 17:04:42 +00:00
|
|
|
|
|
|
|
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):
|
2019-09-26 10:47:53 +00:00
|
|
|
where_clause = "1=1"
|
|
|
|
if room_id_filter:
|
|
|
|
where_clause = "room_id NOT IN (%s)" % (
|
|
|
|
",".join("?" for _ in room_id_filter),
|
|
|
|
)
|
|
|
|
|
2019-06-17 17:04:42 +00:00
|
|
|
sql = """
|
|
|
|
SELECT room_id FROM event_forward_extremities
|
2019-09-26 10:47:53 +00:00
|
|
|
WHERE %s
|
2019-06-17 17:04:42 +00:00
|
|
|
GROUP BY room_id
|
|
|
|
HAVING count(*) > ?
|
|
|
|
ORDER BY count(*) DESC
|
|
|
|
LIMIT ?
|
2019-09-26 10:47:53 +00:00
|
|
|
""" % (
|
|
|
|
where_clause,
|
|
|
|
)
|
2019-06-17 17:04:42 +00:00
|
|
|
|
2019-09-26 10:47:53 +00:00
|
|
|
query_args = list(itertools.chain(room_id_filter, [min_count, limit]))
|
|
|
|
txn.execute(sql, query_args)
|
2019-06-17 17:04:42 +00:00
|
|
|
return [room_id for room_id, in txn]
|
|
|
|
|
2019-12-04 13:52:46 +00:00
|
|
|
return self.db.runInteraction(
|
2019-06-20 09:32:02 +00:00
|
|
|
"get_rooms_with_many_extremities", _get_rooms_with_many_extremities_txn
|
2019-06-17 17:04:42 +00:00
|
|
|
)
|
|
|
|
|
2017-02-01 10:39:41 +00:00
|
|
|
@cached(max_entries=5000, iterable=True)
|
2015-05-01 09:17:19 +00:00
|
|
|
def get_latest_event_ids_in_room(self, room_id):
|
2019-12-04 13:52:46 +00:00
|
|
|
return self.db.simple_select_onecol(
|
2015-05-01 09:17:19 +00:00
|
|
|
table="event_forward_extremities",
|
2019-04-03 09:07:29 +00:00
|
|
|
keyvalues={"room_id": room_id},
|
2015-05-01 09:17:19 +00:00
|
|
|
retcol="event_id",
|
2015-05-05 09:24:10 +00:00
|
|
|
desc="get_latest_event_ids_in_room",
|
2015-05-01 09:17:19 +00:00
|
|
|
)
|
|
|
|
|
2014-10-31 10:47:34 +00:00
|
|
|
def get_min_depth(self, room_id):
|
2014-11-12 15:02:31 +00:00
|
|
|
""" For hte given room, get the minimum depth we have seen for it.
|
|
|
|
"""
|
2019-12-04 13:52:46 +00:00
|
|
|
return self.db.runInteraction(
|
2019-04-03 09:07:29 +00:00
|
|
|
"get_min_depth", self._get_min_depth_interaction, room_id
|
2014-10-31 10:47:34 +00:00
|
|
|
)
|
|
|
|
|
2014-10-28 16:42:35 +00:00
|
|
|
def _get_min_depth_interaction(self, txn, room_id):
|
2019-12-04 13:52:46 +00:00
|
|
|
min_depth = self.db.simple_select_one_onecol_txn(
|
2014-10-28 16:42:35 +00:00
|
|
|
txn,
|
|
|
|
table="room_depth",
|
2014-11-10 13:46:44 +00:00
|
|
|
keyvalues={"room_id": room_id},
|
2014-10-28 16:42:35 +00:00
|
|
|
retcol="min_depth",
|
|
|
|
allow_none=True,
|
|
|
|
)
|
|
|
|
|
|
|
|
return int(min_depth) if min_depth is not None else None
|
|
|
|
|
2016-09-14 15:09:32 +00:00
|
|
|
def get_forward_extremeties_for_room(self, room_id, stream_ordering):
|
2017-02-14 13:59:50 +00:00
|
|
|
"""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
|
|
|
|
"""
|
2016-09-15 13:27:15 +00:00
|
|
|
# We want to make the cache more effective, so we clamp to the last
|
|
|
|
# change before the given ordering.
|
2016-09-15 14:12:07 +00:00
|
|
|
last_change = self._events_stream_cache.get_max_pos_of_last_change(room_id)
|
2016-09-15 16:34:59 +00:00
|
|
|
|
|
|
|
# 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
|
2017-02-14 13:59:50 +00:00
|
|
|
# stream_ordering from before a restart
|
2016-09-15 16:34:59 +00:00
|
|
|
last_change = max(self._stream_order_on_start, last_change)
|
|
|
|
|
2017-02-14 13:59:50 +00:00
|
|
|
# provided the last_change is recent enough, we now clamp the requested
|
|
|
|
# stream_ordering to it.
|
2016-09-15 16:34:59 +00:00
|
|
|
if last_change > self.stream_ordering_month_ago:
|
|
|
|
stream_ordering = min(last_change, stream_ordering)
|
2016-09-15 13:27:15 +00:00
|
|
|
|
|
|
|
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):
|
2016-09-14 15:09:32 +00:00
|
|
|
"""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")
|
|
|
|
|
2019-04-03 09:07:29 +00:00
|
|
|
sql = """
|
2016-09-14 15:09:32 +00:00
|
|
|
SELECT event_id FROM stream_ordering_to_exterm
|
|
|
|
INNER JOIN (
|
|
|
|
SELECT room_id, MAX(stream_ordering) AS stream_ordering
|
|
|
|
FROM stream_ordering_to_exterm
|
2016-09-16 09:19:32 +00:00
|
|
|
WHERE stream_ordering <= ? GROUP BY room_id
|
2016-09-14 15:09:32 +00:00
|
|
|
) AS rms USING (room_id, stream_ordering)
|
|
|
|
WHERE room_id = ?
|
2019-04-03 09:07:29 +00:00
|
|
|
"""
|
2016-09-14 15:09:32 +00:00
|
|
|
|
|
|
|
def get_forward_extremeties_for_room_txn(txn):
|
2016-09-14 16:01:02 +00:00
|
|
|
txn.execute(sql, (stream_ordering, room_id))
|
2017-03-23 17:53:49 +00:00
|
|
|
return [event_id for event_id, in txn]
|
2016-09-14 15:09:32 +00:00
|
|
|
|
2019-12-04 13:52:46 +00:00
|
|
|
return self.db.runInteraction(
|
2019-04-03 09:07:29 +00:00
|
|
|
"get_forward_extremeties_for_room", get_forward_extremeties_for_room_txn
|
2016-09-14 15:09:32 +00:00
|
|
|
)
|
|
|
|
|
2014-10-31 09:59:02 +00:00
|
|
|
def get_backfill_events(self, room_id, event_list, limit):
|
2014-11-12 15:02:31 +00:00
|
|
|
"""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`
|
2014-10-31 09:59:02 +00:00
|
|
|
|
|
|
|
Args:
|
|
|
|
txn
|
|
|
|
room_id (str)
|
|
|
|
event_list (list)
|
|
|
|
limit (int)
|
|
|
|
"""
|
2019-04-03 09:07:29 +00:00
|
|
|
return (
|
2019-12-04 13:52:46 +00:00
|
|
|
self.db.runInteraction(
|
2019-04-03 09:07:29 +00:00
|
|
|
"get_backfill_events",
|
|
|
|
self._get_backfill_events,
|
|
|
|
room_id,
|
|
|
|
event_list,
|
|
|
|
limit,
|
|
|
|
)
|
2019-05-14 12:37:44 +00:00
|
|
|
.addCallback(self.get_events_as_list)
|
2019-04-03 09:07:29 +00:00
|
|
|
.addCallback(lambda l: sorted(l, key=lambda e: -e.depth))
|
2015-05-21 14:57:35 +00:00
|
|
|
)
|
2014-10-31 09:59:02 +00:00
|
|
|
|
|
|
|
def _get_backfill_events(self, txn, room_id, event_list, limit):
|
2019-10-24 17:43:13 +00:00
|
|
|
logger.debug("_get_backfill_events: %s, %r, %s", room_id, event_list, limit)
|
2014-10-31 09:59:02 +00:00
|
|
|
|
2015-05-21 14:44:05 +00:00
|
|
|
event_results = set()
|
2014-10-31 09:59:02 +00:00
|
|
|
|
2015-05-20 11:57:00 +00:00
|
|
|
# We want to make sure that we do a breadth-first, "depth" ordered
|
|
|
|
# search.
|
2014-10-31 09:59:02 +00:00
|
|
|
|
|
|
|
query = (
|
2015-05-20 11:57:00 +00:00
|
|
|
"SELECT depth, prev_event_id FROM event_edges"
|
|
|
|
" INNER JOIN events"
|
|
|
|
" ON prev_event_id = events.event_id"
|
2018-07-26 12:19:08 +00:00
|
|
|
" WHERE event_edges.event_id = ?"
|
2015-05-21 14:52:29 +00:00
|
|
|
" AND event_edges.is_state = ?"
|
2015-05-20 11:57:00 +00:00
|
|
|
" LIMIT ?"
|
2014-10-31 09:59:02 +00:00
|
|
|
)
|
|
|
|
|
2015-05-20 11:57:00 +00:00
|
|
|
queue = PriorityQueue()
|
2014-10-31 09:59:02 +00:00
|
|
|
|
2015-05-20 11:57:00 +00:00
|
|
|
for event_id in event_list:
|
2019-12-04 13:52:46 +00:00
|
|
|
depth = self.db.simple_select_one_onecol_txn(
|
2015-05-21 14:46:07 +00:00
|
|
|
txn,
|
|
|
|
table="events",
|
2019-04-03 09:07:29 +00:00
|
|
|
keyvalues={"event_id": event_id, "room_id": room_id},
|
2015-07-06 08:31:40 +00:00
|
|
|
retcol="depth",
|
|
|
|
allow_none=True,
|
2015-05-21 14:46:07 +00:00
|
|
|
)
|
|
|
|
|
2015-07-06 08:31:40 +00:00
|
|
|
if depth:
|
|
|
|
queue.put((-depth, event_id))
|
2014-10-31 09:59:02 +00:00
|
|
|
|
2015-05-20 11:57:00 +00:00
|
|
|
while not queue.empty() and len(event_results) < limit:
|
2015-05-21 14:37:43 +00:00
|
|
|
try:
|
|
|
|
_, event_id = queue.get_nowait()
|
|
|
|
except Empty:
|
|
|
|
break
|
2014-10-31 09:59:02 +00:00
|
|
|
|
2015-05-21 14:48:56 +00:00
|
|
|
if event_id in event_results:
|
|
|
|
continue
|
|
|
|
|
2015-05-20 11:57:00 +00:00
|
|
|
event_results.add(event_id)
|
|
|
|
|
2019-04-03 09:07:29 +00:00
|
|
|
txn.execute(query, (event_id, False, limit - len(event_results)))
|
2015-05-20 11:57:00 +00:00
|
|
|
|
2017-03-23 17:53:49 +00:00
|
|
|
for row in txn:
|
2015-05-21 14:40:22 +00:00
|
|
|
if row[1] not in event_results:
|
|
|
|
queue.put((-row[0], row[1]))
|
2014-10-31 09:59:02 +00:00
|
|
|
|
2015-05-14 12:45:48 +00:00
|
|
|
return event_results
|
2015-02-19 17:24:14 +00:00
|
|
|
|
2015-05-14 12:45:48 +00:00
|
|
|
@defer.inlineCallbacks
|
2019-04-03 09:07:29 +00:00
|
|
|
def get_missing_events(self, room_id, earliest_events, latest_events, limit):
|
2019-12-04 13:52:46 +00:00
|
|
|
ids = yield self.db.runInteraction(
|
2015-02-19 17:24:14 +00:00
|
|
|
"get_missing_events",
|
|
|
|
self._get_missing_events,
|
2019-04-03 09:07:29 +00:00
|
|
|
room_id,
|
|
|
|
earliest_events,
|
|
|
|
latest_events,
|
|
|
|
limit,
|
2015-02-19 17:24:14 +00:00
|
|
|
)
|
2019-05-14 12:37:44 +00:00
|
|
|
events = yield self.get_events_as_list(ids)
|
2019-07-23 13:00:55 +00:00
|
|
|
return events
|
2015-05-14 12:45:48 +00:00
|
|
|
|
2019-04-03 09:07:29 +00:00
|
|
|
def _get_missing_events(self, txn, room_id, earliest_events, latest_events, limit):
|
2015-02-19 17:24:14 +00:00
|
|
|
|
2018-10-16 19:37:16 +00:00
|
|
|
seen_events = set(earliest_events)
|
|
|
|
front = set(latest_events) - seen_events
|
|
|
|
event_results = []
|
2015-02-19 17:24:14 +00:00
|
|
|
|
|
|
|
query = (
|
|
|
|
"SELECT prev_event_id FROM event_edges "
|
2018-10-16 19:37:16 +00:00
|
|
|
"WHERE room_id = ? AND event_id = ? AND is_state = ? "
|
2015-02-19 17:24:14 +00:00
|
|
|
"LIMIT ?"
|
|
|
|
)
|
|
|
|
|
|
|
|
while front and len(event_results) < limit:
|
|
|
|
new_front = set()
|
|
|
|
for event_id in front:
|
|
|
|
txn.execute(
|
2019-04-03 09:07:29 +00:00
|
|
|
query, (room_id, event_id, False, limit - len(event_results))
|
2015-02-19 17:24:14 +00:00
|
|
|
)
|
|
|
|
|
2018-10-16 19:37:16 +00:00
|
|
|
new_results = set(t[0] for t in txn) - seen_events
|
2015-02-19 17:24:14 +00:00
|
|
|
|
2018-10-16 19:37:16 +00:00
|
|
|
new_front |= new_results
|
|
|
|
seen_events |= new_results
|
|
|
|
event_results.extend(new_results)
|
2015-02-19 17:24:14 +00:00
|
|
|
|
|
|
|
front = new_front
|
|
|
|
|
2018-10-16 19:37:16 +00:00
|
|
|
# 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()
|
2015-05-14 12:45:48 +00:00
|
|
|
return event_results
|
2015-03-18 11:19:47 +00:00
|
|
|
|
2019-02-20 16:54:35 +00:00
|
|
|
@defer.inlineCallbacks
|
2019-03-04 11:56:03 +00:00
|
|
|
def get_successor_events(self, event_ids):
|
2019-02-20 16:54:35 +00:00
|
|
|
"""Fetch all events that have the given events as a prev event
|
|
|
|
|
|
|
|
Args:
|
|
|
|
event_ids (iterable[str])
|
|
|
|
|
|
|
|
Returns:
|
|
|
|
Deferred[list[str]]
|
|
|
|
"""
|
2019-12-04 13:52:46 +00:00
|
|
|
rows = yield self.db.simple_select_many_batch(
|
2019-02-20 16:54:35 +00:00
|
|
|
table="event_edges",
|
|
|
|
column="prev_event_id",
|
|
|
|
iterable=event_ids,
|
|
|
|
retcols=("event_id",),
|
2019-04-03 09:07:29 +00:00
|
|
|
desc="get_successor_events",
|
2019-02-20 16:54:35 +00:00
|
|
|
)
|
|
|
|
|
2019-07-23 13:00:55 +00:00
|
|
|
return [row["event_id"] for row in rows]
|
2019-02-20 16:54:35 +00:00
|
|
|
|
2018-03-01 14:16:02 +00:00
|
|
|
|
|
|
|
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"
|
|
|
|
|
2019-12-06 13:08:40 +00:00
|
|
|
def __init__(self, database: Database, db_conn, hs):
|
|
|
|
super(EventFederationStore, self).__init__(database, db_conn, hs)
|
2018-03-01 14:16:02 +00:00
|
|
|
|
2019-12-04 15:09:36 +00:00
|
|
|
self.db.updates.register_background_update_handler(
|
2019-04-03 09:07:29 +00:00
|
|
|
self.EVENT_AUTH_STATE_ONLY, self._background_delete_non_state_event_auth
|
2018-03-01 14:16:02 +00:00
|
|
|
)
|
|
|
|
|
|
|
|
hs.get_clock().looping_call(
|
2019-04-03 09:07:29 +00:00
|
|
|
self._delete_old_forward_extrem_cache, 60 * 60 * 1000
|
2018-03-01 14:16:02 +00:00
|
|
|
)
|
|
|
|
|
|
|
|
def _update_min_depth_for_room_txn(self, txn, room_id, depth):
|
|
|
|
min_depth = self._get_min_depth_interaction(txn, room_id)
|
|
|
|
|
2020-01-07 12:08:58 +00:00
|
|
|
if min_depth is not None and depth >= min_depth:
|
2018-03-01 14:16:02 +00:00
|
|
|
return
|
|
|
|
|
2019-12-04 13:52:46 +00:00
|
|
|
self.db.simple_upsert_txn(
|
2018-03-01 14:16:02 +00:00
|
|
|
txn,
|
|
|
|
table="room_depth",
|
2019-04-03 09:07:29 +00:00
|
|
|
keyvalues={"room_id": room_id},
|
|
|
|
values={"min_depth": depth},
|
2018-03-01 14:16:02 +00:00
|
|
|
)
|
|
|
|
|
|
|
|
def _handle_mult_prev_events(self, txn, events):
|
|
|
|
"""
|
|
|
|
For the given event, update the event edges table and forward and
|
|
|
|
backward extremities tables.
|
|
|
|
"""
|
2019-12-04 13:52:46 +00:00
|
|
|
self.db.simple_insert_many_txn(
|
2018-03-01 14:16:02 +00:00
|
|
|
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
|
2018-11-05 13:35:15 +00:00
|
|
|
for e_id in ev.prev_event_ids()
|
2018-03-01 14:16:02 +00:00
|
|
|
],
|
|
|
|
)
|
|
|
|
|
|
|
|
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 = ?"
|
|
|
|
" )"
|
|
|
|
)
|
|
|
|
|
2019-04-03 09:07:29 +00:00
|
|
|
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()
|
|
|
|
],
|
|
|
|
)
|
2018-03-01 14:16:02 +00:00
|
|
|
|
|
|
|
query = (
|
|
|
|
"DELETE FROM event_backward_extremities"
|
|
|
|
" WHERE event_id = ? AND room_id = ?"
|
|
|
|
)
|
|
|
|
txn.executemany(
|
|
|
|
query,
|
|
|
|
[
|
2019-04-03 09:07:29 +00:00
|
|
|
(ev.event_id, ev.room_id)
|
|
|
|
for ev in events
|
2018-03-01 14:16:02 +00:00
|
|
|
if not ev.internal_metadata.is_outlier()
|
2019-04-03 09:07:29 +00:00
|
|
|
],
|
2018-03-01 14:16:02 +00:00
|
|
|
)
|
|
|
|
|
|
|
|
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.
|
2019-04-03 09:07:29 +00:00
|
|
|
sql = """
|
2018-03-01 14:16:02 +00:00
|
|
|
DELETE FROM stream_ordering_to_exterm
|
|
|
|
WHERE
|
|
|
|
room_id IN (
|
|
|
|
SELECT room_id
|
|
|
|
FROM stream_ordering_to_exterm
|
|
|
|
WHERE stream_ordering > ?
|
|
|
|
) AND stream_ordering < ?
|
2019-04-03 09:07:29 +00:00
|
|
|
"""
|
2018-03-01 14:16:02 +00:00
|
|
|
txn.execute(
|
2019-04-03 09:07:29 +00:00
|
|
|
sql, (self.stream_ordering_month_ago, self.stream_ordering_month_ago)
|
2018-03-01 14:16:02 +00:00
|
|
|
)
|
2019-04-03 09:07:29 +00:00
|
|
|
|
2018-07-26 10:44:26 +00:00
|
|
|
return run_as_background_process(
|
2018-07-25 08:41:12 +00:00
|
|
|
"delete_old_forward_extrem_cache",
|
2019-12-04 13:52:46 +00:00
|
|
|
self.db.runInteraction,
|
2018-03-01 14:16:02 +00:00
|
|
|
"_delete_old_forward_extrem_cache",
|
2018-07-25 08:41:12 +00:00
|
|
|
_delete_old_forward_extrem_cache_txn,
|
2018-03-01 14:16:02 +00:00
|
|
|
)
|
|
|
|
|
2015-03-18 11:19:47 +00:00
|
|
|
def clean_room_for_join(self, room_id):
|
2019-12-04 13:52:46 +00:00
|
|
|
return self.db.runInteraction(
|
2019-04-03 09:07:29 +00:00
|
|
|
"clean_room_for_join", self._clean_room_for_join_txn, room_id
|
2015-03-18 11:19:47 +00:00
|
|
|
)
|
|
|
|
|
|
|
|
def _clean_room_for_join_txn(self, txn, room_id):
|
|
|
|
query = "DELETE FROM event_forward_extremities WHERE room_id = ?"
|
|
|
|
|
|
|
|
txn.execute(query, (room_id,))
|
2015-08-07 10:52:21 +00:00
|
|
|
txn.call_after(self.get_latest_event_ids_in_room.invalidate, (room_id,))
|
2017-05-24 13:58:13 +00:00
|
|
|
|
|
|
|
@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
|
|
|
|
)
|
|
|
|
"""
|
|
|
|
|
2019-04-03 09:07:29 +00:00
|
|
|
txn.execute(sql, (min_stream_id, max_stream_id))
|
2017-05-24 13:58:13 +00:00
|
|
|
|
|
|
|
new_progress = {
|
|
|
|
"target_min_stream_id_inclusive": target_min_stream_id,
|
|
|
|
"max_stream_id_exclusive": min_stream_id,
|
|
|
|
}
|
|
|
|
|
2019-12-04 15:09:36 +00:00
|
|
|
self.db.updates._background_update_progress_txn(
|
2017-05-24 13:58:13 +00:00
|
|
|
txn, self.EVENT_AUTH_STATE_ONLY, new_progress
|
|
|
|
)
|
|
|
|
|
|
|
|
return min_stream_id >= target_min_stream_id
|
|
|
|
|
2019-12-04 13:52:46 +00:00
|
|
|
result = yield self.db.runInteraction(
|
2017-05-24 13:58:13 +00:00
|
|
|
self.EVENT_AUTH_STATE_ONLY, delete_event_auth
|
|
|
|
)
|
|
|
|
|
|
|
|
if not result:
|
2019-12-04 15:09:36 +00:00
|
|
|
yield self.db.updates._end_background_update(self.EVENT_AUTH_STATE_ONLY)
|
2017-05-24 13:58:13 +00:00
|
|
|
|
2019-07-23 13:00:55 +00:00
|
|
|
return batch_size
|