|
- # Copyright 2014-2016 OpenMarket Ltd
- # Copyright 2017 Vector Creations Ltd
- # Copyright 2018-2019 New Vector Ltd
- # Copyright 2019 The Matrix.org Foundation C.I.C.
- #
- # 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.
-
- """ This module is responsible for getting events from the DB for pagination
- and event streaming.
-
- The order it returns events in depend on whether we are streaming forwards or
- are paginating backwards. We do this because we want to handle out of order
- messages nicely, while still returning them in the correct order when we
- paginate bacwards.
-
- This is implemented by keeping two ordering columns: stream_ordering and
- topological_ordering. Stream ordering is basically insertion/received order
- (except for events from backfill requests). The topological_ordering is a
- weak ordering of events based on the pdu graph.
-
- This means that we have to have two different types of tokens, depending on
- what sort order was used:
- - stream tokens are of the form: "s%d", which maps directly to the column
- - topological tokems: "t%d-%d", where the integers map to the topological
- and stream ordering columns respectively.
- """
-
- import logging
- from typing import (
- TYPE_CHECKING,
- Any,
- Collection,
- Dict,
- List,
- Optional,
- Set,
- Tuple,
- cast,
- overload,
- )
-
- import attr
- from immutabledict import immutabledict
- from typing_extensions import Literal
-
- from twisted.internet import defer
-
- from synapse.api.constants import Direction
- from synapse.api.filtering import Filter
- from synapse.events import EventBase
- from synapse.logging.context import make_deferred_yieldable, run_in_background
- from synapse.logging.opentracing import trace
- from synapse.storage._base import SQLBaseStore
- from synapse.storage.database import (
- DatabasePool,
- LoggingDatabaseConnection,
- LoggingTransaction,
- make_in_list_sql_clause,
- )
- from synapse.storage.databases.main.events_worker import EventsWorkerStore
- from synapse.storage.engines import BaseDatabaseEngine, PostgresEngine, Sqlite3Engine
- from synapse.storage.util.id_generators import MultiWriterIdGenerator
- from synapse.types import PersistedEventPosition, RoomStreamToken
- from synapse.util.caches.descriptors import cached
- from synapse.util.caches.stream_change_cache import StreamChangeCache
- from synapse.util.cancellation import cancellable
-
- if TYPE_CHECKING:
- from synapse.server import HomeServer
-
- logger = logging.getLogger(__name__)
-
-
- MAX_STREAM_SIZE = 1000
-
-
- _STREAM_TOKEN = "stream"
- _TOPOLOGICAL_TOKEN = "topological"
-
-
- # Used as return values for pagination APIs
- @attr.s(slots=True, frozen=True, auto_attribs=True)
- class _EventDictReturn:
- event_id: str
- topological_ordering: Optional[int]
- stream_ordering: int
-
-
- @attr.s(slots=True, frozen=True, auto_attribs=True)
- class _EventsAround:
- events_before: List[EventBase]
- events_after: List[EventBase]
- start: RoomStreamToken
- end: RoomStreamToken
-
-
- def generate_pagination_where_clause(
- direction: Direction,
- column_names: Tuple[str, str],
- from_token: Optional[Tuple[Optional[int], int]],
- to_token: Optional[Tuple[Optional[int], int]],
- engine: BaseDatabaseEngine,
- ) -> str:
- """Creates an SQL expression to bound the columns by the pagination
- tokens.
-
- For example creates an SQL expression like:
-
- (6, 7) >= (topological_ordering, stream_ordering)
- AND (5, 3) < (topological_ordering, stream_ordering)
-
- would be generated for dir=b, from_token=(6, 7) and to_token=(5, 3).
-
- Note that tokens are considered to be after the row they are in, e.g. if
- a row A has a token T, then we consider A to be before T. This convention
- is important when figuring out inequalities for the generated SQL, and
- produces the following result:
- - If paginating forwards then we exclude any rows matching the from
- token, but include those that match the to token.
- - If paginating backwards then we include any rows matching the from
- token, but include those that match the to token.
-
- Args:
- direction: Whether we're paginating backwards or forwards.
- column_names: The column names to bound. Must *not* be user defined as
- these get inserted directly into the SQL statement without escapes.
- from_token: The start point for the pagination. This is an exclusive
- minimum bound if direction is forwards, and an inclusive maximum bound if
- direction is backwards.
- to_token: The endpoint point for the pagination. This is an inclusive
- maximum bound if direction is forwards, and an exclusive minimum bound if
- direction is backwards.
- engine: The database engine to generate the clauses for
-
- Returns:
- The sql expression
- """
-
- where_clause = []
- if from_token:
- where_clause.append(
- _make_generic_sql_bound(
- bound=">=" if direction == Direction.BACKWARDS else "<",
- column_names=column_names,
- values=from_token,
- engine=engine,
- )
- )
-
- if to_token:
- where_clause.append(
- _make_generic_sql_bound(
- bound="<" if direction == Direction.BACKWARDS else ">=",
- column_names=column_names,
- values=to_token,
- engine=engine,
- )
- )
-
- return " AND ".join(where_clause)
-
-
- def generate_pagination_bounds(
- direction: Direction,
- from_token: Optional[RoomStreamToken],
- to_token: Optional[RoomStreamToken],
- ) -> Tuple[
- str, Optional[Tuple[Optional[int], int]], Optional[Tuple[Optional[int], int]]
- ]:
- """
- Generate a start and end point for this page of events.
-
- Args:
- direction: Whether pagination is going forwards or backwards.
- from_token: The token to start pagination at, or None to start at the first value.
- to_token: The token to end pagination at, or None to not limit the end point.
-
- Returns:
- A three tuple of:
-
- ASC or DESC for sorting of the query.
-
- The starting position as a tuple of ints representing
- (topological position, stream position) or None if no from_token was
- provided. The topological position may be None for live tokens.
-
- The end position in the same format as the starting position, or None
- if no to_token was provided.
- """
-
- # Tokens really represent positions between elements, but we use
- # the convention of pointing to the event before the gap. Hence
- # we have a bit of asymmetry when it comes to equalities.
- if direction == Direction.BACKWARDS:
- order = "DESC"
- else:
- order = "ASC"
-
- # The bounds for the stream tokens are complicated by the fact
- # that we need to handle the instance_map part of the tokens. We do this
- # by fetching all events between the min stream token and the maximum
- # stream token (as returned by `RoomStreamToken.get_max_stream_pos`) and
- # then filtering the results.
- from_bound: Optional[Tuple[Optional[int], int]] = None
- if from_token:
- if from_token.topological is not None:
- from_bound = from_token.as_historical_tuple()
- elif direction == Direction.BACKWARDS:
- from_bound = (
- None,
- from_token.get_max_stream_pos(),
- )
- else:
- from_bound = (
- None,
- from_token.stream,
- )
-
- to_bound: Optional[Tuple[Optional[int], int]] = None
- if to_token:
- if to_token.topological is not None:
- to_bound = to_token.as_historical_tuple()
- elif direction == Direction.BACKWARDS:
- to_bound = (
- None,
- to_token.stream,
- )
- else:
- to_bound = (
- None,
- to_token.get_max_stream_pos(),
- )
-
- return order, from_bound, to_bound
-
-
- def generate_next_token(
- direction: Direction, last_topo_ordering: int, last_stream_ordering: int
- ) -> RoomStreamToken:
- """
- Generate the next room stream token based on the currently returned data.
-
- Args:
- direction: Whether pagination is going forwards or backwards.
- last_topo_ordering: The last topological ordering being returned.
- last_stream_ordering: The last stream ordering being returned.
-
- Returns:
- A new RoomStreamToken to return to the client.
- """
- if direction == Direction.BACKWARDS:
- # Tokens are positions between events.
- # This token points *after* the last event in the chunk.
- # We need it to point to the event before it in the chunk
- # when we are going backwards so we subtract one from the
- # stream part.
- last_stream_ordering -= 1
- return RoomStreamToken(last_topo_ordering, last_stream_ordering)
-
-
- def _make_generic_sql_bound(
- bound: str,
- column_names: Tuple[str, str],
- values: Tuple[Optional[int], int],
- engine: BaseDatabaseEngine,
- ) -> str:
- """Create an SQL expression that bounds the given column names by the
- values, e.g. create the equivalent of `(1, 2) < (col1, col2)`.
-
- Only works with two columns.
-
- Older versions of SQLite don't support that syntax so we have to expand it
- out manually.
-
- Args:
- bound: The comparison operator to use. One of ">", "<", ">=",
- "<=", where the values are on the left and columns on the right.
- names: The column names. Must *not* be user defined
- as these get inserted directly into the SQL statement without
- escapes.
- values: The values to bound the columns by. If
- the first value is None then only creates a bound on the second
- column.
- engine: The database engine to generate the SQL for
-
- Returns:
- The SQL statement
- """
-
- assert bound in (">", "<", ">=", "<=")
-
- name1, name2 = column_names
- val1, val2 = values
-
- if val1 is None:
- val2 = int(val2)
- return "(%d %s %s)" % (val2, bound, name2)
-
- val1 = int(val1)
- val2 = int(val2)
-
- if isinstance(engine, PostgresEngine):
- # Postgres doesn't optimise ``(x < a) OR (x=a AND y<b)`` as well
- # as it optimises ``(x,y) < (a,b)`` on multicolumn indexes. So we
- # use the later form when running against postgres.
- return "((%d,%d) %s (%s,%s))" % (val1, val2, bound, name1, name2)
-
- # We want to generate queries of e.g. the form:
- #
- # (val1 < name1 OR (val1 = name1 AND val2 <= name2))
- #
- # which is equivalent to (val1, val2) < (name1, name2)
-
- return """(
- {val1:d} {strict_bound} {name1}
- OR ({val1:d} = {name1} AND {val2:d} {bound} {name2})
- )""".format(
- name1=name1,
- val1=val1,
- name2=name2,
- val2=val2,
- strict_bound=bound[0], # The first bound must always be strict equality here
- bound=bound,
- )
-
-
- def _filter_results(
- lower_token: Optional[RoomStreamToken],
- upper_token: Optional[RoomStreamToken],
- instance_name: str,
- topological_ordering: int,
- stream_ordering: int,
- ) -> bool:
- """Returns True if the event persisted by the given instance at the given
- topological/stream_ordering falls between the two tokens (taking a None
- token to mean unbounded).
-
- Used to filter results from fetching events in the DB against the given
- tokens. This is necessary to handle the case where the tokens include
- position maps, which we handle by fetching more than necessary from the DB
- and then filtering (rather than attempting to construct a complicated SQL
- query).
- """
-
- event_historical_tuple = (
- topological_ordering,
- stream_ordering,
- )
-
- if lower_token:
- if lower_token.topological is not None:
- # If these are historical tokens we compare the `(topological, stream)`
- # tuples.
- if event_historical_tuple <= lower_token.as_historical_tuple():
- return False
-
- else:
- # If these are live tokens we compare the stream ordering against the
- # writers stream position.
- if stream_ordering <= lower_token.get_stream_pos_for_instance(
- instance_name
- ):
- return False
-
- if upper_token:
- if upper_token.topological is not None:
- if upper_token.as_historical_tuple() < event_historical_tuple:
- return False
- else:
- if upper_token.get_stream_pos_for_instance(instance_name) < stream_ordering:
- return False
-
- return True
-
-
- def filter_to_clause(event_filter: Optional[Filter]) -> Tuple[str, List[str]]:
- # NB: This may create SQL clauses that don't optimise well (and we don't
- # have indices on all possible clauses). E.g. it may create
- # "room_id == X AND room_id != X", which postgres doesn't optimise.
-
- if not event_filter:
- return "", []
-
- clauses = []
- args = []
-
- if event_filter.types:
- clauses.append(
- "(%s)" % " OR ".join("event.type = ?" for _ in event_filter.types)
- )
- args.extend(event_filter.types)
-
- for typ in event_filter.not_types:
- clauses.append("event.type != ?")
- args.append(typ)
-
- if event_filter.senders:
- clauses.append(
- "(%s)" % " OR ".join("event.sender = ?" for _ in event_filter.senders)
- )
- args.extend(event_filter.senders)
-
- for sender in event_filter.not_senders:
- clauses.append("event.sender != ?")
- args.append(sender)
-
- if event_filter.rooms:
- clauses.append(
- "(%s)" % " OR ".join("event.room_id = ?" for _ in event_filter.rooms)
- )
- args.extend(event_filter.rooms)
-
- for room_id in event_filter.not_rooms:
- clauses.append("event.room_id != ?")
- args.append(room_id)
-
- if event_filter.contains_url:
- clauses.append("event.contains_url = ?")
- args.append(event_filter.contains_url)
-
- # We're only applying the "labels" filter on the database query, because applying the
- # "not_labels" filter via a SQL query is non-trivial. Instead, we let
- # event_filter.check_fields apply it, which is not as efficient but makes the
- # implementation simpler.
- if event_filter.labels:
- clauses.append("(%s)" % " OR ".join("label = ?" for _ in event_filter.labels))
- args.extend(event_filter.labels)
-
- # Filter on relation_senders / relation types from the joined tables.
- if event_filter.related_by_senders:
- clauses.append(
- "(%s)"
- % " OR ".join(
- "related_event.sender = ?" for _ in event_filter.related_by_senders
- )
- )
- args.extend(event_filter.related_by_senders)
-
- if event_filter.related_by_rel_types:
- clauses.append(
- "(%s)"
- % " OR ".join(
- "relation_type = ?" for _ in event_filter.related_by_rel_types
- )
- )
- args.extend(event_filter.related_by_rel_types)
-
- if event_filter.rel_types:
- clauses.append(
- "(%s)"
- % " OR ".join(
- "event_relation.relation_type = ?" for _ in event_filter.rel_types
- )
- )
- args.extend(event_filter.rel_types)
-
- if event_filter.not_rel_types:
- clauses.append(
- "((%s) OR event_relation.relation_type IS NULL)"
- % " AND ".join(
- "event_relation.relation_type != ?" for _ in event_filter.not_rel_types
- )
- )
- args.extend(event_filter.not_rel_types)
-
- return " AND ".join(clauses), args
-
-
- class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
- def __init__(
- self,
- database: DatabasePool,
- db_conn: LoggingDatabaseConnection,
- hs: "HomeServer",
- ):
- super().__init__(database, db_conn, hs)
-
- self._instance_name = hs.get_instance_name()
- self._send_federation = hs.should_send_federation()
- self._federation_shard_config = hs.config.worker.federation_shard_config
-
- # If we're a process that sends federation we may need to reset the
- # `federation_stream_position` table to match the current sharding
- # config. We don't do this now as otherwise two processes could conflict
- # during startup which would cause one to die.
- self._need_to_reset_federation_stream_positions = self._send_federation
-
- events_max = self.get_room_max_stream_ordering()
- event_cache_prefill, min_event_val = self.db_pool.get_cache_dict(
- db_conn,
- "events",
- entity_column="room_id",
- stream_column="stream_ordering",
- max_value=events_max,
- )
- self._events_stream_cache = StreamChangeCache(
- "EventsRoomStreamChangeCache",
- min_event_val,
- prefilled_cache=event_cache_prefill,
- )
- self._membership_stream_cache = StreamChangeCache(
- "MembershipStreamChangeCache", events_max
- )
-
- self._stream_order_on_start = self.get_room_max_stream_ordering()
- self._min_stream_order_on_start = self.get_room_min_stream_ordering()
-
- def get_room_max_stream_ordering(self) -> int:
- """Get the stream_ordering of regular events that we have committed up to
-
- Returns the maximum stream id such that all stream ids less than or
- equal to it have been successfully persisted.
- """
- return self._stream_id_gen.get_current_token()
-
- def get_room_min_stream_ordering(self) -> int:
- """Get the stream_ordering of backfilled events that we have committed up to
-
- Backfilled events use *negative* stream orderings, so this returns the
- minimum negative stream id such that all stream ids greater than or
- equal to it have been successfully persisted.
- """
- return self._backfill_id_gen.get_current_token()
-
- def get_room_max_token(self) -> RoomStreamToken:
- """Get a `RoomStreamToken` that marks the current maximum persisted
- position of the events stream. Useful to get a token that represents
- "now".
-
- The token returned is a "live" token that may have an instance_map
- component.
- """
-
- min_pos = self._stream_id_gen.get_current_token()
-
- positions = {}
- if isinstance(self._stream_id_gen, MultiWriterIdGenerator):
- # The `min_pos` is the minimum position that we know all instances
- # have finished persisting to, so we only care about instances whose
- # positions are ahead of that. (Instance positions can be behind the
- # min position as there are times we can work out that the minimum
- # position is ahead of the naive minimum across all current
- # positions. See MultiWriterIdGenerator for details)
- positions = {
- i: p
- for i, p in self._stream_id_gen.get_positions().items()
- if p > min_pos
- }
-
- return RoomStreamToken(None, min_pos, immutabledict(positions))
-
- async def get_room_events_stream_for_rooms(
- self,
- room_ids: Collection[str],
- from_key: RoomStreamToken,
- to_key: RoomStreamToken,
- limit: int = 0,
- order: str = "DESC",
- ) -> Dict[str, Tuple[List[EventBase], RoomStreamToken]]:
- """Get new room events in stream ordering since `from_key`.
-
- Args:
- room_ids
- from_key: Token from which no events are returned before
- to_key: Token from which no events are returned after. (This
- is typically the current stream token)
- limit: Maximum number of events to return
- order: Either "DESC" or "ASC". Determines which events are
- returned when the result is limited. If "DESC" then the most
- recent `limit` events are returned, otherwise returns the
- oldest `limit` events.
-
- Returns:
- A map from room id to a tuple containing:
- - list of recent events in the room
- - stream ordering key for the start of the chunk of events returned.
- """
- room_ids = self._events_stream_cache.get_entities_changed(
- room_ids, from_key.stream
- )
-
- if not room_ids:
- return {}
-
- results = {}
- room_ids = list(room_ids)
- for rm_ids in (room_ids[i : i + 20] for i in range(0, len(room_ids), 20)):
- res = await make_deferred_yieldable(
- defer.gatherResults(
- [
- run_in_background(
- self.get_room_events_stream_for_room,
- room_id,
- from_key,
- to_key,
- limit,
- order=order,
- )
- for room_id in rm_ids
- ],
- consumeErrors=True,
- )
- )
- results.update(dict(zip(rm_ids, res)))
-
- return results
-
- def get_rooms_that_changed(
- self, room_ids: Collection[str], from_key: RoomStreamToken
- ) -> Set[str]:
- """Given a list of rooms and a token, return rooms where there may have
- been changes.
- """
- from_id = from_key.stream
- return {
- room_id
- for room_id in room_ids
- if self._events_stream_cache.has_entity_changed(room_id, from_id)
- }
-
- async def get_room_events_stream_for_room(
- self,
- room_id: str,
- from_key: RoomStreamToken,
- to_key: RoomStreamToken,
- limit: int = 0,
- order: str = "DESC",
- ) -> Tuple[List[EventBase], RoomStreamToken]:
- """Get new room events in stream ordering since `from_key`.
-
- Args:
- room_id
- from_key: Token from which no events are returned before
- to_key: Token from which no events are returned after. (This
- is typically the current stream token)
- limit: Maximum number of events to return
- order: Either "DESC" or "ASC". Determines which events are
- returned when the result is limited. If "DESC" then the most
- recent `limit` events are returned, otherwise returns the
- oldest `limit` events.
-
- Returns:
- The list of events (in ascending stream order) and the token from the start
- of the chunk of events returned.
- """
- if from_key == to_key:
- return [], from_key
-
- has_changed = self._events_stream_cache.has_entity_changed(
- room_id, from_key.stream
- )
-
- if not has_changed:
- return [], from_key
-
- def f(txn: LoggingTransaction) -> List[_EventDictReturn]:
- # To handle tokens with a non-empty instance_map we fetch more
- # results than necessary and then filter down
- min_from_id = from_key.stream
- max_to_id = to_key.get_max_stream_pos()
-
- sql = """
- SELECT event_id, instance_name, topological_ordering, stream_ordering
- FROM events
- WHERE
- room_id = ?
- AND not outlier
- AND stream_ordering > ? AND stream_ordering <= ?
- ORDER BY stream_ordering %s LIMIT ?
- """ % (
- order,
- )
- txn.execute(sql, (room_id, min_from_id, max_to_id, 2 * limit))
-
- rows = [
- _EventDictReturn(event_id, None, stream_ordering)
- for event_id, instance_name, topological_ordering, stream_ordering in txn
- if _filter_results(
- from_key,
- to_key,
- instance_name,
- topological_ordering,
- stream_ordering,
- )
- ][:limit]
- return rows
-
- rows = await self.db_pool.runInteraction("get_room_events_stream_for_room", f)
-
- ret = await self.get_events_as_list(
- [r.event_id for r in rows], get_prev_content=True
- )
-
- self._set_before_and_after(ret, rows, topo_order=False)
-
- if order.lower() == "desc":
- ret.reverse()
-
- if rows:
- key = RoomStreamToken(None, min(r.stream_ordering for r in rows))
- else:
- # Assume we didn't get anything because there was nothing to
- # get.
- key = from_key
-
- return ret, key
-
- @cancellable
- async def get_membership_changes_for_user(
- self,
- user_id: str,
- from_key: RoomStreamToken,
- to_key: RoomStreamToken,
- excluded_rooms: Optional[List[str]] = None,
- ) -> List[EventBase]:
- """Fetch membership events for a given user.
-
- All such events whose stream ordering `s` lies in the range
- `from_key < s <= to_key` are returned. Events are ordered by ascending stream
- order.
- """
- # Start by ruling out cases where a DB query is not necessary.
- if from_key == to_key:
- return []
-
- if from_key:
- has_changed = self._membership_stream_cache.has_entity_changed(
- user_id, int(from_key.stream)
- )
- if not has_changed:
- return []
-
- def f(txn: LoggingTransaction) -> List[_EventDictReturn]:
- # To handle tokens with a non-empty instance_map we fetch more
- # results than necessary and then filter down
- min_from_id = from_key.stream
- max_to_id = to_key.get_max_stream_pos()
-
- args: List[Any] = [user_id, min_from_id, max_to_id]
-
- ignore_room_clause = ""
- if excluded_rooms is not None and len(excluded_rooms) > 0:
- ignore_room_clause = "AND e.room_id NOT IN (%s)" % ",".join(
- "?" for _ in excluded_rooms
- )
- args = args + excluded_rooms
-
- sql = """
- SELECT m.event_id, instance_name, topological_ordering, stream_ordering
- FROM events AS e, room_memberships AS m
- WHERE e.event_id = m.event_id
- AND m.user_id = ?
- AND e.stream_ordering > ? AND e.stream_ordering <= ?
- %s
- ORDER BY e.stream_ordering ASC
- """ % (
- ignore_room_clause,
- )
-
- txn.execute(sql, args)
-
- rows = [
- _EventDictReturn(event_id, None, stream_ordering)
- for event_id, instance_name, topological_ordering, stream_ordering in txn
- if _filter_results(
- from_key,
- to_key,
- instance_name,
- topological_ordering,
- stream_ordering,
- )
- ]
-
- return rows
-
- rows = await self.db_pool.runInteraction("get_membership_changes_for_user", f)
-
- ret = await self.get_events_as_list(
- [r.event_id for r in rows], get_prev_content=True
- )
-
- self._set_before_and_after(ret, rows, topo_order=False)
-
- return ret
-
- async def get_recent_events_for_room(
- self, room_id: str, limit: int, end_token: RoomStreamToken
- ) -> Tuple[List[EventBase], RoomStreamToken]:
- """Get the most recent events in the room in topological ordering.
-
- Args:
- room_id
- limit
- end_token: The stream token representing now.
-
- Returns:
- A list of events and a token pointing to the start of the returned
- events. The events returned are in ascending topological order.
- """
-
- rows, token = await self.get_recent_event_ids_for_room(
- room_id, limit, end_token
- )
-
- events = await self.get_events_as_list(
- [r.event_id for r in rows], get_prev_content=True
- )
-
- self._set_before_and_after(events, rows)
-
- return events, token
-
- async def get_recent_event_ids_for_room(
- self, room_id: str, limit: int, end_token: RoomStreamToken
- ) -> Tuple[List[_EventDictReturn], RoomStreamToken]:
- """Get the most recent events in the room in topological ordering.
-
- Args:
- room_id
- limit
- end_token: The stream token representing now.
-
- Returns:
- A list of _EventDictReturn and a token pointing to the start of the
- returned events. The events returned are in ascending order.
- """
- # Allow a zero limit here, and no-op.
- if limit == 0:
- return [], end_token
-
- rows, token = await self.db_pool.runInteraction(
- "get_recent_event_ids_for_room",
- self._paginate_room_events_txn,
- room_id,
- from_token=end_token,
- limit=limit,
- )
-
- # We want to return the results in ascending order.
- rows.reverse()
-
- return rows, token
-
- async def get_room_event_before_stream_ordering(
- self, room_id: str, stream_ordering: int
- ) -> Optional[Tuple[int, int, str]]:
- """Gets details of the first event in a room at or before a stream ordering
-
- Args:
- room_id:
- stream_ordering:
-
- Returns:
- A tuple of (stream ordering, topological ordering, event_id)
- """
-
- def _f(txn: LoggingTransaction) -> Optional[Tuple[int, int, str]]:
- sql = """
- SELECT stream_ordering, topological_ordering, event_id
- FROM events
- LEFT JOIN rejections USING (event_id)
- WHERE room_id = ?
- AND stream_ordering <= ?
- AND NOT outlier
- AND rejections.event_id IS NULL
- ORDER BY stream_ordering DESC
- LIMIT 1
- """
- txn.execute(sql, (room_id, stream_ordering))
- return cast(Optional[Tuple[int, int, str]], txn.fetchone())
-
- return await self.db_pool.runInteraction(
- "get_room_event_before_stream_ordering", _f
- )
-
- async def get_last_event_in_room_before_stream_ordering(
- self,
- room_id: str,
- end_token: RoomStreamToken,
- ) -> Optional[str]:
- """Returns the ID of the last event in a room at or before a stream ordering
-
- Args:
- room_id
- end_token: The token used to stream from
-
- Returns:
- The ID of the most recent event, or None if there are no events in the room
- before this stream ordering.
- """
-
- def get_last_event_in_room_before_stream_ordering_txn(
- txn: LoggingTransaction,
- ) -> Optional[str]:
- # We need to handle the fact that the stream tokens can be vector
- # clocks. We do this by getting all rows between the minimum and
- # maximum stream ordering in the token, plus one row less than the
- # minimum stream ordering. We then filter the results against the
- # token and return the first row that matches.
-
- sql = """
- SELECT * FROM (
- SELECT instance_name, stream_ordering, topological_ordering, event_id
- FROM events
- LEFT JOIN rejections USING (event_id)
- WHERE room_id = ?
- AND ? < stream_ordering AND stream_ordering <= ?
- AND NOT outlier
- AND rejections.event_id IS NULL
- ORDER BY stream_ordering DESC
- ) AS a
- UNION
- SELECT * FROM (
- SELECT instance_name, stream_ordering, topological_ordering, event_id
- FROM events
- LEFT JOIN rejections USING (event_id)
- WHERE room_id = ?
- AND stream_ordering <= ?
- AND NOT outlier
- AND rejections.event_id IS NULL
- ORDER BY stream_ordering DESC
- LIMIT 1
- ) AS b
- """
- txn.execute(
- sql,
- (
- room_id,
- end_token.stream,
- end_token.get_max_stream_pos(),
- room_id,
- end_token.stream,
- ),
- )
-
- for instance_name, stream_ordering, topological_ordering, event_id in txn:
- if _filter_results(
- lower_token=None,
- upper_token=end_token,
- instance_name=instance_name,
- topological_ordering=topological_ordering,
- stream_ordering=stream_ordering,
- ):
- return event_id
-
- return None
-
- return await self.db_pool.runInteraction(
- "get_last_event_in_room_before_stream_ordering",
- get_last_event_in_room_before_stream_ordering_txn,
- )
-
- async def get_current_room_stream_token_for_room_id(
- self, room_id: str
- ) -> RoomStreamToken:
- """Returns the current position of the rooms stream (historic token)."""
- stream_ordering = self.get_room_max_stream_ordering()
- topo = await self.db_pool.runInteraction(
- "_get_max_topological_txn", self._get_max_topological_txn, room_id
- )
- return RoomStreamToken(topo, stream_ordering)
-
- @overload
- def get_stream_id_for_event_txn(
- self,
- txn: LoggingTransaction,
- event_id: str,
- allow_none: Literal[False] = False,
- ) -> int:
- ...
-
- @overload
- def get_stream_id_for_event_txn(
- self,
- txn: LoggingTransaction,
- event_id: str,
- allow_none: bool = False,
- ) -> Optional[int]:
- ...
-
- def get_stream_id_for_event_txn(
- self,
- txn: LoggingTransaction,
- event_id: str,
- allow_none: bool = False,
- ) -> Optional[int]:
- # Type ignore: we pass keyvalues a Dict[str, str]; the function wants
- # Dict[str, Any]. I think mypy is unhappy because Dict is invariant?
- return self.db_pool.simple_select_one_onecol_txn( # type: ignore[call-overload]
- txn=txn,
- table="events",
- keyvalues={"event_id": event_id},
- retcol="stream_ordering",
- allow_none=allow_none,
- )
-
- async def get_position_for_event(self, event_id: str) -> PersistedEventPosition:
- """Get the persisted position for an event"""
- row = await self.db_pool.simple_select_one(
- table="events",
- keyvalues={"event_id": event_id},
- retcols=("stream_ordering", "instance_name"),
- desc="get_position_for_event",
- )
-
- return PersistedEventPosition(
- row["instance_name"] or "master", row["stream_ordering"]
- )
-
- async def get_topological_token_for_event(self, event_id: str) -> RoomStreamToken:
- """The stream token for an event
- Args:
- event_id: The id of the event to look up a stream token for.
- Raises:
- StoreError if the event wasn't in the database.
- Returns:
- A `RoomStreamToken` topological token.
- """
- row = await self.db_pool.simple_select_one(
- table="events",
- keyvalues={"event_id": event_id},
- retcols=("stream_ordering", "topological_ordering"),
- desc="get_topological_token_for_event",
- )
- return RoomStreamToken(row["topological_ordering"], row["stream_ordering"])
-
- async def get_current_topological_token(self, room_id: str, stream_key: int) -> int:
- """Gets the topological token in a room after or at the given stream
- ordering.
-
- Args:
- room_id
- stream_key
- """
- if isinstance(self.database_engine, PostgresEngine):
- min_function = "LEAST"
- elif isinstance(self.database_engine, Sqlite3Engine):
- min_function = "MIN"
- else:
- raise RuntimeError(f"Unknown database engine {self.database_engine}")
-
- # This query used to be
- # SELECT COALESCE(MIN(topological_ordering), 0) FROM events
- # WHERE room_id = ? and events.stream_ordering >= {stream_key}
- # which returns 0 if the stream_key is newer than any event in
- # the room. That's not wrong, but it seems to interact oddly with backfill,
- # requiring a second call to /messages to actually backfill from a remote
- # homeserver.
- #
- # Instead, rollback the stream ordering to that after the most recent event in
- # this room.
- sql = f"""
- WITH fallback(max_stream_ordering) AS (
- SELECT MAX(stream_ordering)
- FROM events
- WHERE room_id = ?
- )
- SELECT COALESCE(MIN(topological_ordering), 0) FROM events
- WHERE
- room_id = ?
- AND events.stream_ordering >= {min_function}(
- ?,
- (SELECT max_stream_ordering FROM fallback)
- )
- """
-
- row = await self.db_pool.execute(
- "get_current_topological_token", None, sql, room_id, room_id, stream_key
- )
- return row[0][0] if row else 0
-
- def _get_max_topological_txn(self, txn: LoggingTransaction, room_id: str) -> int:
- txn.execute(
- "SELECT MAX(topological_ordering) FROM events WHERE room_id = ?",
- (room_id,),
- )
-
- rows = txn.fetchall()
- # An aggregate function like MAX() will always return one row per group
- # so we can safely rely on the lookup here. For example, when a we
- # lookup a `room_id` which does not exist, `rows` will look like
- # `[(None,)]`
- return rows[0][0] if rows[0][0] is not None else 0
-
- @staticmethod
- def _set_before_and_after(
- events: List[EventBase], rows: List[_EventDictReturn], topo_order: bool = True
- ) -> None:
- """Inserts ordering information to events' internal metadata from
- the DB rows.
-
- Args:
- events
- rows
- topo_order: Whether the events were ordered topologically or by stream
- ordering. If true then all rows should have a non null
- topological_ordering.
- """
- for event, row in zip(events, rows):
- stream = row.stream_ordering
- if topo_order and row.topological_ordering:
- topo: Optional[int] = row.topological_ordering
- else:
- topo = None
- internal = event.internal_metadata
- internal.before = RoomStreamToken(topo, stream - 1)
- internal.after = RoomStreamToken(topo, stream)
- internal.order = (int(topo) if topo else 0, int(stream))
-
- async def get_events_around(
- self,
- room_id: str,
- event_id: str,
- before_limit: int,
- after_limit: int,
- event_filter: Optional[Filter] = None,
- ) -> _EventsAround:
- """Retrieve events and pagination tokens around a given event in a
- room.
- """
-
- results = await self.db_pool.runInteraction(
- "get_events_around",
- self._get_events_around_txn,
- room_id,
- event_id,
- before_limit,
- after_limit,
- event_filter,
- )
-
- events_before = await self.get_events_as_list(
- list(results["before"]["event_ids"]), get_prev_content=True
- )
-
- events_after = await self.get_events_as_list(
- list(results["after"]["event_ids"]), get_prev_content=True
- )
-
- return _EventsAround(
- events_before=events_before,
- events_after=events_after,
- start=results["before"]["token"],
- end=results["after"]["token"],
- )
-
- def _get_events_around_txn(
- self,
- txn: LoggingTransaction,
- room_id: str,
- event_id: str,
- before_limit: int,
- after_limit: int,
- event_filter: Optional[Filter],
- ) -> dict:
- """Retrieves event_ids and pagination tokens around a given event in a
- room.
-
- Args:
- room_id
- event_id
- before_limit
- after_limit
- event_filter
-
- Returns:
- dict
- """
-
- results = self.db_pool.simple_select_one_txn(
- txn,
- "events",
- keyvalues={"event_id": event_id, "room_id": room_id},
- retcols=["stream_ordering", "topological_ordering"],
- )
-
- # This cannot happen as `allow_none=False`.
- assert results is not None
-
- # Paginating backwards includes the event at the token, but paginating
- # forward doesn't.
- before_token = RoomStreamToken(
- results["topological_ordering"] - 1, results["stream_ordering"]
- )
-
- after_token = RoomStreamToken(
- results["topological_ordering"], results["stream_ordering"]
- )
-
- rows, start_token = self._paginate_room_events_txn(
- txn,
- room_id,
- before_token,
- direction=Direction.BACKWARDS,
- limit=before_limit,
- event_filter=event_filter,
- )
- events_before = [r.event_id for r in rows]
-
- rows, end_token = self._paginate_room_events_txn(
- txn,
- room_id,
- after_token,
- direction=Direction.FORWARDS,
- limit=after_limit,
- event_filter=event_filter,
- )
- events_after = [r.event_id for r in rows]
-
- return {
- "before": {"event_ids": events_before, "token": start_token},
- "after": {"event_ids": events_after, "token": end_token},
- }
-
- async def get_all_new_event_ids_stream(
- self,
- from_id: int,
- current_id: int,
- limit: int,
- ) -> Tuple[int, Dict[str, Optional[int]]]:
- """Get all new events
-
- Returns all event ids with from_id < stream_ordering <= current_id.
-
- Args:
- from_id: the stream_ordering of the last event we processed
- current_id: the stream_ordering of the most recently processed event
- limit: the maximum number of events to return
-
- Returns:
- A tuple of (next_id, event_to_received_ts), where `next_id`
- is the next value to pass as `from_id` (it will either be the
- stream_ordering of the last returned event, or, if fewer than `limit`
- events were found, the `current_id`). The `event_to_received_ts` is
- a dictionary mapping event ID to the event `received_ts`, sorted by ascending
- stream_ordering.
- """
-
- def get_all_new_event_ids_stream_txn(
- txn: LoggingTransaction,
- ) -> Tuple[int, Dict[str, Optional[int]]]:
- sql = (
- "SELECT e.stream_ordering, e.event_id, e.received_ts"
- " FROM events AS e"
- " WHERE"
- " ? < e.stream_ordering AND e.stream_ordering <= ?"
- " ORDER BY e.stream_ordering ASC"
- " LIMIT ?"
- )
-
- txn.execute(sql, (from_id, current_id, limit))
- rows = txn.fetchall()
-
- upper_bound = current_id
- if len(rows) == limit:
- upper_bound = rows[-1][0]
-
- event_to_received_ts: Dict[str, Optional[int]] = {
- row[1]: row[2] for row in rows
- }
- return upper_bound, event_to_received_ts
-
- upper_bound, event_to_received_ts = await self.db_pool.runInteraction(
- "get_all_new_event_ids_stream", get_all_new_event_ids_stream_txn
- )
-
- return upper_bound, event_to_received_ts
-
- async def get_federation_out_pos(self, typ: str) -> int:
- if self._need_to_reset_federation_stream_positions:
- await self.db_pool.runInteraction(
- "_reset_federation_positions_txn", self._reset_federation_positions_txn
- )
- self._need_to_reset_federation_stream_positions = False
-
- return await self.db_pool.simple_select_one_onecol(
- table="federation_stream_position",
- retcol="stream_id",
- keyvalues={"type": typ, "instance_name": self._instance_name},
- desc="get_federation_out_pos",
- )
-
- async def update_federation_out_pos(self, typ: str, stream_id: int) -> None:
- if self._need_to_reset_federation_stream_positions:
- await self.db_pool.runInteraction(
- "_reset_federation_positions_txn", self._reset_federation_positions_txn
- )
- self._need_to_reset_federation_stream_positions = False
-
- await self.db_pool.simple_update_one(
- table="federation_stream_position",
- keyvalues={"type": typ, "instance_name": self._instance_name},
- updatevalues={"stream_id": stream_id},
- desc="update_federation_out_pos",
- )
-
- def _reset_federation_positions_txn(self, txn: LoggingTransaction) -> None:
- """Fiddles with the `federation_stream_position` table to make it match
- the configured federation sender instances during start up.
- """
-
- # The federation sender instances may have changed, so we need to
- # massage the `federation_stream_position` table to have a row per type
- # per instance sending federation. If there is a mismatch we update the
- # table with the correct rows using the *minimum* stream ID seen. This
- # may result in resending of events/EDUs to remote servers, but that is
- # preferable to dropping them.
-
- if not self._send_federation:
- return
-
- # Pull out the configured instances. If we don't have a shard config then
- # we assume that we're the only instance sending.
- configured_instances = self._federation_shard_config.instances
- if not configured_instances:
- configured_instances = [self._instance_name]
- elif self._instance_name not in configured_instances:
- return
-
- instances_in_table = self.db_pool.simple_select_onecol_txn(
- txn,
- table="federation_stream_position",
- keyvalues={},
- retcol="instance_name",
- )
-
- if set(instances_in_table) == set(configured_instances):
- # Nothing to do
- return
-
- sql = """
- SELECT type, MIN(stream_id) FROM federation_stream_position
- GROUP BY type
- """
- txn.execute(sql)
- min_positions = {typ: pos for typ, pos in txn} # Map from type -> min position
-
- # Ensure we do actually have some values here
- assert set(min_positions) == {"federation", "events"}
-
- sql = """
- DELETE FROM federation_stream_position
- WHERE NOT (%s)
- """
- clause, args = make_in_list_sql_clause(
- txn.database_engine, "instance_name", configured_instances
- )
- txn.execute(sql % (clause,), args)
-
- for typ, stream_id in min_positions.items():
- self.db_pool.simple_upsert_txn(
- txn,
- table="federation_stream_position",
- keyvalues={"type": typ, "instance_name": self._instance_name},
- values={"stream_id": stream_id},
- )
-
- def has_room_changed_since(self, room_id: str, stream_id: int) -> bool:
- return self._events_stream_cache.has_entity_changed(room_id, stream_id)
-
- def _paginate_room_events_txn(
- self,
- txn: LoggingTransaction,
- room_id: str,
- from_token: RoomStreamToken,
- to_token: Optional[RoomStreamToken] = None,
- direction: Direction = Direction.BACKWARDS,
- limit: int = -1,
- event_filter: Optional[Filter] = None,
- ) -> Tuple[List[_EventDictReturn], RoomStreamToken]:
- """Returns list of events before or after a given token.
-
- Args:
- txn
- room_id
- from_token: The token used to stream from
- to_token: A token which if given limits the results to only those before
- direction: Indicates whether we are paginating forwards or backwards
- from `from_key`.
- limit: The maximum number of events to return.
- event_filter: If provided filters the events to
- those that match the filter.
-
- Returns:
- A list of _EventDictReturn and a token that points to the end of the
- result set. If no events are returned then the end of the stream has
- been reached (i.e. there are no events between `from_token` and
- `to_token`), or `limit` is zero.
- """
-
- args = [False, room_id]
-
- order, from_bound, to_bound = generate_pagination_bounds(
- direction, from_token, to_token
- )
-
- bounds = generate_pagination_where_clause(
- direction=direction,
- column_names=("event.topological_ordering", "event.stream_ordering"),
- from_token=from_bound,
- to_token=to_bound,
- engine=self.database_engine,
- )
-
- filter_clause, filter_args = filter_to_clause(event_filter)
-
- if filter_clause:
- bounds += " AND " + filter_clause
- args.extend(filter_args)
-
- # We fetch more events as we'll filter the result set
- args.append(int(limit) * 2)
-
- select_keywords = "SELECT"
- join_clause = ""
- # Using DISTINCT in this SELECT query is quite expensive, because it
- # requires the engine to sort on the entire (not limited) result set,
- # i.e. the entire events table. Only use it in scenarios that could result
- # in the same event ID occurring multiple times in the results.
- needs_distinct = False
- if event_filter and event_filter.labels:
- # If we're not filtering on a label, then joining on event_labels will
- # return as many row for a single event as the number of labels it has. To
- # avoid this, only join if we're filtering on at least one label.
- join_clause += """
- LEFT JOIN event_labels
- USING (event_id, room_id, topological_ordering)
- """
- if len(event_filter.labels) > 1:
- # Multiple labels could cause the same event to appear multiple times.
- needs_distinct = True
-
- # If there is a relation_senders and relation_types filter join to the
- # relations table to get events related to the current event.
- if event_filter and (
- event_filter.related_by_senders or event_filter.related_by_rel_types
- ):
- # Filtering by relations could cause the same event to appear multiple
- # times (since there's no limit on the number of relations to an event).
- needs_distinct = True
- join_clause += """
- LEFT JOIN event_relations AS relation ON (event.event_id = relation.relates_to_id)
- """
- if event_filter.related_by_senders:
- join_clause += """
- LEFT JOIN events AS related_event ON (relation.event_id = related_event.event_id)
- """
-
- # If there is a not_rel_types filter join to the relations table to get
- # the event's relation information.
- if event_filter and (event_filter.rel_types or event_filter.not_rel_types):
- join_clause += """
- LEFT JOIN event_relations AS event_relation USING (event_id)
- """
-
- if needs_distinct:
- select_keywords += " DISTINCT"
-
- sql = """
- %(select_keywords)s
- event.event_id, event.instance_name,
- event.topological_ordering, event.stream_ordering
- FROM events AS event
- %(join_clause)s
- WHERE event.outlier = ? AND event.room_id = ? AND %(bounds)s
- ORDER BY event.topological_ordering %(order)s,
- event.stream_ordering %(order)s LIMIT ?
- """ % {
- "select_keywords": select_keywords,
- "join_clause": join_clause,
- "bounds": bounds,
- "order": order,
- }
-
- txn.execute(sql, args)
-
- # Filter the result set.
- rows = [
- _EventDictReturn(event_id, topological_ordering, stream_ordering)
- for event_id, instance_name, topological_ordering, stream_ordering in txn
- if _filter_results(
- lower_token=to_token
- if direction == Direction.BACKWARDS
- else from_token,
- upper_token=from_token
- if direction == Direction.BACKWARDS
- else to_token,
- instance_name=instance_name,
- topological_ordering=topological_ordering,
- stream_ordering=stream_ordering,
- )
- ][:limit]
-
- if rows:
- assert rows[-1].topological_ordering is not None
- next_token = generate_next_token(
- direction, rows[-1].topological_ordering, rows[-1].stream_ordering
- )
- else:
- # TODO (erikj): We should work out what to do here instead.
- next_token = to_token if to_token else from_token
-
- return rows, next_token
-
- @trace
- async def paginate_room_events(
- self,
- room_id: str,
- from_key: RoomStreamToken,
- to_key: Optional[RoomStreamToken] = None,
- direction: Direction = Direction.BACKWARDS,
- limit: int = -1,
- event_filter: Optional[Filter] = None,
- ) -> Tuple[List[EventBase], RoomStreamToken]:
- """Returns list of events before or after a given token.
-
- Args:
- room_id
- from_key: The token used to stream from
- to_key: A token which if given limits the results to only those before
- direction: Indicates whether we are paginating forwards or backwards
- from `from_key`.
- limit: The maximum number of events to return.
- event_filter: If provided filters the events to those that match the filter.
-
- Returns:
- The results as a list of events and a token that points to the end
- of the result set. If no events are returned then the end of the
- stream has been reached (i.e. there are no events between `from_key`
- and `to_key`).
- """
-
- rows, token = await self.db_pool.runInteraction(
- "paginate_room_events",
- self._paginate_room_events_txn,
- room_id,
- from_key,
- to_key,
- direction,
- limit,
- event_filter,
- )
-
- events = await self.get_events_as_list(
- [r.event_id for r in rows], get_prev_content=True
- )
-
- self._set_before_and_after(events, rows)
-
- return events, token
-
- @cached()
- async def get_id_for_instance(self, instance_name: str) -> int:
- """Get a unique, immutable ID that corresponds to the given Synapse worker instance."""
-
- def _get_id_for_instance_txn(txn: LoggingTransaction) -> int:
- instance_id = self.db_pool.simple_select_one_onecol_txn(
- txn,
- table="instance_map",
- keyvalues={"instance_name": instance_name},
- retcol="instance_id",
- allow_none=True,
- )
- if instance_id is not None:
- return instance_id
-
- # If we don't have an entry upsert one.
- #
- # We could do this before the first check, and rely on the cache for
- # efficiency, but each UPSERT causes the next ID to increment which
- # can quickly bloat the size of the generated IDs for new instances.
- self.db_pool.simple_upsert_txn(
- txn,
- table="instance_map",
- keyvalues={"instance_name": instance_name},
- values={},
- )
-
- return self.db_pool.simple_select_one_onecol_txn(
- txn,
- table="instance_map",
- keyvalues={"instance_name": instance_name},
- retcol="instance_id",
- )
-
- return await self.db_pool.runInteraction(
- "get_id_for_instance", _get_id_for_instance_txn
- )
-
- @cached()
- async def get_name_from_instance_id(self, instance_id: int) -> str:
- """Get the instance name from an ID previously returned by
- `get_id_for_instance`.
- """
-
- return await self.db_pool.simple_select_one_onecol(
- table="instance_map",
- keyvalues={"instance_id": instance_id},
- retcol="instance_name",
- desc="get_name_from_instance_id",
- )
|