diff --git a/synapse/storage/controllers/persist_events.py b/synapse/storage/controllers/persist_events.py
index d0e015bf19..ac0919340b 100644
--- a/synapse/storage/controllers/persist_events.py
+++ b/synapse/storage/controllers/persist_events.py
@@ -502,8 +502,15 @@ class EventsPersistenceStorageController:
"""
state = await self._calculate_current_state(room_id)
delta = await self._calculate_state_delta(room_id, state)
+ sliding_sync_table_changes = (
+ await self.persist_events_store._calculate_sliding_sync_table_changes(
+ room_id, [], delta
+ )
+ )
- await self.persist_events_store.update_current_state(room_id, delta)
+ await self.persist_events_store.update_current_state(
+ room_id, delta, sliding_sync_table_changes
+ )
async def _calculate_current_state(self, room_id: str) -> StateMap[str]:
"""Calculate the current state of a room, based on the forward extremities
diff --git a/synapse/storage/database.py b/synapse/storage/database.py
index 569f618193..d666039120 100644
--- a/synapse/storage/database.py
+++ b/synapse/storage/database.py
@@ -35,6 +35,7 @@ from typing import (
Iterable,
Iterator,
List,
+ Mapping,
Optional,
Sequence,
Tuple,
@@ -1254,9 +1255,9 @@ class DatabasePool:
self,
txn: LoggingTransaction,
table: str,
- keyvalues: Dict[str, Any],
- values: Dict[str, Any],
- insertion_values: Optional[Dict[str, Any]] = None,
+ keyvalues: Mapping[str, Any],
+ values: Mapping[str, Any],
+ insertion_values: Optional[Mapping[str, Any]] = None,
where_clause: Optional[str] = None,
) -> bool:
"""
@@ -1299,9 +1300,9 @@ class DatabasePool:
self,
txn: LoggingTransaction,
table: str,
- keyvalues: Dict[str, Any],
- values: Dict[str, Any],
- insertion_values: Optional[Dict[str, Any]] = None,
+ keyvalues: Mapping[str, Any],
+ values: Mapping[str, Any],
+ insertion_values: Optional[Mapping[str, Any]] = None,
where_clause: Optional[str] = None,
lock: bool = True,
) -> bool:
@@ -1322,7 +1323,7 @@ class DatabasePool:
if lock:
# We need to lock the table :(
- self.engine.lock_table(txn, table)
+ txn.database_engine.lock_table(txn, table)
def _getwhere(key: str) -> str:
# If the value we're passing in is None (aka NULL), we need to use
@@ -1376,13 +1377,13 @@ class DatabasePool:
# successfully inserted
return True
+ @staticmethod
def simple_upsert_txn_native_upsert(
- self,
txn: LoggingTransaction,
table: str,
- keyvalues: Dict[str, Any],
- values: Dict[str, Any],
- insertion_values: Optional[Dict[str, Any]] = None,
+ keyvalues: Mapping[str, Any],
+ values: Mapping[str, Any],
+ insertion_values: Optional[Mapping[str, Any]] = None,
where_clause: Optional[str] = None,
) -> bool:
"""
@@ -1535,8 +1536,8 @@ class DatabasePool:
self.simple_upsert_txn_emulated(txn, table, _keys, _vals, lock=False)
+ @staticmethod
def simple_upsert_many_txn_native_upsert(
- self,
txn: LoggingTransaction,
table: str,
key_names: Collection[str],
@@ -1966,8 +1967,8 @@ class DatabasePool:
def simple_update_txn(
txn: LoggingTransaction,
table: str,
- keyvalues: Dict[str, Any],
- updatevalues: Dict[str, Any],
+ keyvalues: Mapping[str, Any],
+ updatevalues: Mapping[str, Any],
) -> int:
"""
Update rows in the given database table.
diff --git a/synapse/storage/databases/main/events.py b/synapse/storage/databases/main/events.py
index 1f7acdb859..60c92e5804 100644
--- a/synapse/storage/databases/main/events.py
+++ b/synapse/storage/databases/main/events.py
@@ -32,6 +32,7 @@ from typing import (
Iterable,
List,
Optional,
+ Sequence,
Set,
Tuple,
cast,
@@ -39,19 +40,27 @@ from typing import (
import attr
from prometheus_client import Counter
+from typing_extensions import TypedDict
import synapse.metrics
-from synapse.api.constants import EventContentFields, EventTypes, RelationTypes
+from synapse.api.constants import (
+ EventContentFields,
+ EventTypes,
+ Membership,
+ RelationTypes,
+)
from synapse.api.errors import PartialStateConflictError
from synapse.api.room_versions import RoomVersions
-from synapse.events import EventBase, relation_from_event
+from synapse.events import EventBase, StrippedStateEvent, relation_from_event
from synapse.events.snapshot import EventContext
+from synapse.events.utils import parse_stripped_state_event
from synapse.logging.opentracing import trace
from synapse.storage._base import db_to_json, make_in_list_sql_clause
from synapse.storage.database import (
DatabasePool,
LoggingDatabaseConnection,
LoggingTransaction,
+ make_tuple_in_list_sql_clause,
)
from synapse.storage.databases.main.event_federation import EventFederationStore
from synapse.storage.databases.main.events_worker import EventCacheEntry
@@ -59,7 +68,15 @@ from synapse.storage.databases.main.search import SearchEntry
from synapse.storage.engines import PostgresEngine
from synapse.storage.util.id_generators import AbstractStreamIdGenerator
from synapse.storage.util.sequence import SequenceGenerator
-from synapse.types import JsonDict, StateMap, StrCollection, get_domain_from_id
+from synapse.types import (
+ JsonDict,
+ MutableStateMap,
+ StateMap,
+ StrCollection,
+ get_domain_from_id,
+)
+from synapse.types.handlers import SLIDING_SYNC_DEFAULT_BUMP_EVENT_TYPES
+from synapse.types.state import StateFilter
from synapse.util import json_encoder
from synapse.util.iterutils import batch_iter, sorted_topologically
from synapse.util.stringutils import non_null_str_or_none
@@ -78,6 +95,19 @@ event_counter = Counter(
["type", "origin_type", "origin_entity"],
)
+# State event type/key pairs that we need to gather to fill in the
+# `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` tables.
+SLIDING_SYNC_RELEVANT_STATE_SET = (
+ # So we can fill in the `room_type` column
+ (EventTypes.Create, ""),
+ # So we can fill in the `is_encrypted` column
+ (EventTypes.RoomEncryption, ""),
+ # So we can fill in the `room_name` column
+ (EventTypes.Name, ""),
+ # So we can fill in the `tombstone_successor_room_id` column
+ (EventTypes.Tombstone, ""),
+)
+
@attr.s(slots=True, auto_attribs=True)
class DeltaState:
@@ -99,6 +129,82 @@ class DeltaState:
return not self.to_delete and not self.to_insert and not self.no_longer_in_room
+# We want `total=False` because we want to allow values to be unset.
+class SlidingSyncStateInsertValues(TypedDict, total=False):
+ """
+ Insert values relevant for the `sliding_sync_joined_rooms` and
+ `sliding_sync_membership_snapshots` database tables.
+ """
+
+ room_type: Optional[str]
+ is_encrypted: Optional[bool]
+ room_name: Optional[str]
+ tombstone_successor_room_id: Optional[str]
+
+
+class SlidingSyncMembershipSnapshotSharedInsertValues(
+ SlidingSyncStateInsertValues, total=False
+):
+ """
+ Insert values for `sliding_sync_membership_snapshots` that we can share across
+ multiple memberships
+ """
+
+ has_known_state: Optional[bool]
+
+
+@attr.s(slots=True, auto_attribs=True)
+class SlidingSyncMembershipInfo:
+ """
+ Values unique to each membership
+ """
+
+ user_id: str
+ sender: str
+ membership_event_id: str
+ membership: str
+ membership_event_stream_ordering: int
+ membership_event_instance_name: str
+
+
+@attr.s(slots=True, auto_attribs=True)
+class SlidingSyncTableChanges:
+ room_id: str
+ # `stream_ordering` of the most recent event being persisted in the room. This doesn't
+ # need to be perfect, we just need *some* answer that points to a real event in the
+ # room in case we are the first ones inserting into the `sliding_sync_joined_rooms`
+ # table because of the `NON NULL` constraint on `event_stream_ordering`. In reality,
+ # `_update_sliding_sync_tables_with_new_persisted_events_txn()` is run after
+ # `_update_current_state_txn()` whenever a new event is persisted to update it to the
+ # correct latest value.
+ #
+ # This should be *some* value that points to a real event in the room if we are
+ # still joined to the room and some state is changing (`to_insert` or `to_delete`).
+ joined_room_best_effort_most_recent_stream_ordering: Optional[int]
+ # If the row doesn't exist in the `sliding_sync_joined_rooms` table, we need to
+ # fully-insert it which means we also need to include a `bump_stamp` value to use
+ # for the row. This should only be populated when we're trying to fully-insert a
+ # row.
+ #
+ # FIXME: This can be removed once we bump `SCHEMA_COMPAT_VERSION` and run the
+ # foreground update for
+ # `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` (tracked by
+ # https://github.com/element-hq/synapse/issues/17623)
+ joined_room_bump_stamp_to_fully_insert: Optional[int]
+ # Values to upsert into `sliding_sync_joined_rooms`
+ joined_room_updates: SlidingSyncStateInsertValues
+
+ # Shared values to upsert into `sliding_sync_membership_snapshots` for each
+ # `to_insert_membership_snapshots`
+ membership_snapshot_shared_insert_values: (
+ SlidingSyncMembershipSnapshotSharedInsertValues
+ )
+ # List of membership to insert into `sliding_sync_membership_snapshots`
+ to_insert_membership_snapshots: List[SlidingSyncMembershipInfo]
+ # List of user_id to delete from `sliding_sync_membership_snapshots`
+ to_delete_membership_snapshots: List[str]
+
+
@attr.s(slots=True, auto_attribs=True)
class NewEventChainLinks:
"""Information about new auth chain links that need to be added to the DB.
@@ -226,6 +332,14 @@ class PersistEventsStore:
event.internal_metadata.stream_ordering = stream
event.internal_metadata.instance_name = self._instance_name
+ sliding_sync_table_changes = None
+ if state_delta_for_room is not None:
+ sliding_sync_table_changes = (
+ await self._calculate_sliding_sync_table_changes(
+ room_id, events_and_contexts, state_delta_for_room
+ )
+ )
+
await self.db_pool.runInteraction(
"persist_events",
self._persist_events_txn,
@@ -235,6 +349,7 @@ class PersistEventsStore:
state_delta_for_room=state_delta_for_room,
new_forward_extremities=new_forward_extremities,
new_event_links=new_event_links,
+ sliding_sync_table_changes=sliding_sync_table_changes,
)
persist_event_counter.inc(len(events_and_contexts))
@@ -261,6 +376,341 @@ class PersistEventsStore:
(room_id,), frozenset(new_forward_extremities)
)
+ async def _calculate_sliding_sync_table_changes(
+ self,
+ room_id: str,
+ events_and_contexts: Sequence[Tuple[EventBase, EventContext]],
+ delta_state: DeltaState,
+ ) -> SlidingSyncTableChanges:
+ """
+ Calculate the changes to the `sliding_sync_membership_snapshots` and
+ `sliding_sync_joined_rooms` tables given the deltas that are going to be used to
+ update the `current_state_events` table.
+
+ Just a bunch of pre-processing so we so we don't need to spend time in the
+ transaction itself gathering all of this info. It's also easier to deal with
+ redactions outside of a transaction.
+
+ Args:
+ room_id: The room ID currently being processed.
+ events_and_contexts: List of tuples of (event, context) being persisted.
+ This is completely optional (you can pass an empty list) and will just
+ save us from fetching the events from the database if we already have
+ them. We assume the list is sorted ascending by `stream_ordering`. We
+ don't care about the sort when the events are backfilled (with negative
+ `stream_ordering`).
+ delta_state: Deltas that are going to be used to update the
+ `current_state_events` table. Changes to the current state of the room.
+ """
+ to_insert = delta_state.to_insert
+ to_delete = delta_state.to_delete
+
+ # If no state is changing, we don't need to do anything. This can happen when a
+ # partial-stated room is re-syncing the current state.
+ if not to_insert and not to_delete:
+ return SlidingSyncTableChanges(
+ room_id=room_id,
+ joined_room_best_effort_most_recent_stream_ordering=None,
+ joined_room_bump_stamp_to_fully_insert=None,
+ joined_room_updates={},
+ membership_snapshot_shared_insert_values={},
+ to_insert_membership_snapshots=[],
+ to_delete_membership_snapshots=[],
+ )
+
+ event_map = {event.event_id: event for event, _ in events_and_contexts}
+
+ # Handle gathering info for the `sliding_sync_membership_snapshots` table
+ #
+ # This would only happen if someone was state reset out of the room
+ user_ids_to_delete_membership_snapshots = [
+ state_key
+ for event_type, state_key in to_delete
+ if event_type == EventTypes.Member and self.is_mine_id(state_key)
+ ]
+
+ membership_snapshot_shared_insert_values: (
+ SlidingSyncMembershipSnapshotSharedInsertValues
+ ) = {}
+ membership_infos_to_insert_membership_snapshots: List[
+ SlidingSyncMembershipInfo
+ ] = []
+ if to_insert:
+ membership_event_id_to_user_id_map: Dict[str, str] = {}
+ for state_key, event_id in to_insert.items():
+ if state_key[0] == EventTypes.Member and self.is_mine_id(state_key[1]):
+ membership_event_id_to_user_id_map[event_id] = state_key[1]
+
+ membership_event_map: Dict[str, EventBase] = {}
+ # In normal event persist scenarios, we should be able to find the
+ # membership events in the `events_and_contexts` given to us but it's
+ # possible a state reset happened which added us to the room without a
+ # corresponding new membership event (reset back to a previous membership).
+ missing_membership_event_ids: Set[str] = set()
+ for membership_event_id in membership_event_id_to_user_id_map.keys():
+ membership_event = event_map.get(membership_event_id)
+ if membership_event:
+ membership_event_map[membership_event_id] = membership_event
+ else:
+ missing_membership_event_ids.add(membership_event_id)
+
+ # Otherwise, we need to find a couple events that we were reset to.
+ if missing_membership_event_ids:
+ remaining_events = await self.store.get_events(
+ missing_membership_event_ids
+ )
+ # There shouldn't be any missing events
+ assert (
+ remaining_events.keys() == missing_membership_event_ids
+ ), missing_membership_event_ids.difference(remaining_events.keys())
+ membership_event_map.update(remaining_events)
+
+ for (
+ membership_event_id,
+ user_id,
+ ) in membership_event_id_to_user_id_map.items():
+ # We should only be seeing events with `stream_ordering`/`instance_name` assigned by this point
+ membership_event_stream_ordering = membership_event_map[
+ membership_event_id
+ ].internal_metadata.stream_ordering
+ assert membership_event_stream_ordering is not None
+ membership_event_instance_name = membership_event_map[
+ membership_event_id
+ ].internal_metadata.instance_name
+ assert membership_event_instance_name is not None
+
+ membership_infos_to_insert_membership_snapshots.append(
+ SlidingSyncMembershipInfo(
+ user_id=user_id,
+ sender=membership_event_map[membership_event_id].sender,
+ membership_event_id=membership_event_id,
+ membership=membership_event_map[membership_event_id].membership,
+ membership_event_stream_ordering=membership_event_stream_ordering,
+ membership_event_instance_name=membership_event_instance_name,
+ )
+ )
+
+ if membership_infos_to_insert_membership_snapshots:
+ current_state_ids_map: MutableStateMap[str] = dict(
+ await self.store.get_partial_filtered_current_state_ids(
+ room_id,
+ state_filter=StateFilter.from_types(
+ SLIDING_SYNC_RELEVANT_STATE_SET
+ ),
+ )
+ )
+ # Since we fetched the current state before we took `to_insert`/`to_delete`
+ # into account, we need to do a couple fixups.
+ #
+ # Update the current_state_map with what we have `to_delete`
+ for state_key in to_delete:
+ current_state_ids_map.pop(state_key, None)
+ # Update the current_state_map with what we have `to_insert`
+ for state_key, event_id in to_insert.items():
+ if state_key in SLIDING_SYNC_RELEVANT_STATE_SET:
+ current_state_ids_map[state_key] = event_id
+
+ current_state_map: MutableStateMap[EventBase] = {}
+ # In normal event persist scenarios, we probably won't be able to find
+ # these state events in `events_and_contexts` since we don't generally
+ # batch up local membership changes with other events, but it can
+ # happen.
+ missing_state_event_ids: Set[str] = set()
+ for state_key, event_id in current_state_ids_map.items():
+ event = event_map.get(event_id)
+ if event:
+ current_state_map[state_key] = event
+ else:
+ missing_state_event_ids.add(event_id)
+
+ # Otherwise, we need to find a couple events
+ if missing_state_event_ids:
+ remaining_events = await self.store.get_events(
+ missing_state_event_ids
+ )
+ # There shouldn't be any missing events
+ assert (
+ remaining_events.keys() == missing_state_event_ids
+ ), missing_state_event_ids.difference(remaining_events.keys())
+ for event in remaining_events.values():
+ current_state_map[(event.type, event.state_key)] = event
+
+ if current_state_map:
+ state_insert_values = PersistEventsStore._get_sliding_sync_insert_values_from_state_map(
+ current_state_map
+ )
+ membership_snapshot_shared_insert_values.update(state_insert_values)
+ # We have current state to work from
+ membership_snapshot_shared_insert_values["has_known_state"] = True
+ else:
+ # We don't have any `current_state_events` anymore (previously
+ # cleared out because of `no_longer_in_room`). This can happen if
+ # one user is joined and another is invited (some non-join
+ # membership). If the joined user leaves, we are `no_longer_in_room`
+ # and `current_state_events` is cleared out. When the invited user
+ # rejects the invite (leaves the room), we will end up here.
+ #
+ # In these cases, we should inherit the meta data from the previous
+ # snapshot so we shouldn't update any of the state values. When
+ # using sliding sync filters, this will prevent the room from
+ # disappearing/appearing just because you left the room.
+ #
+ # Ideally, we could additionally assert that we're only here for
+ # valid non-join membership transitions.
+ assert delta_state.no_longer_in_room
+
+ # Handle gathering info for the `sliding_sync_joined_rooms` table
+ #
+ # We only deal with
+ # updating the state related columns. The
+ # `event_stream_ordering`/`bump_stamp` are updated elsewhere in the event
+ # persisting stack (see
+ # `_update_sliding_sync_tables_with_new_persisted_events_txn()`)
+ #
+ joined_room_updates: SlidingSyncStateInsertValues = {}
+ best_effort_most_recent_stream_ordering: Optional[int] = None
+ bump_stamp_to_fully_insert: Optional[int] = None
+ if not delta_state.no_longer_in_room:
+ current_state_ids_map = {}
+
+ # Always fully-insert rows if they don't already exist in the
+ # `sliding_sync_joined_rooms` table. This way we can rely on a row if it
+ # exists in the table.
+ #
+ # FIXME: This can be removed once we bump `SCHEMA_COMPAT_VERSION` and run the
+ # foreground update for
+ # `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` (tracked by
+ # https://github.com/element-hq/synapse/issues/17623)
+ existing_row_in_table = await self.store.db_pool.simple_select_one_onecol(
+ table="sliding_sync_joined_rooms",
+ keyvalues={"room_id": room_id},
+ retcol="room_id",
+ allow_none=True,
+ )
+ if not existing_row_in_table:
+ most_recent_bump_event_pos_results = (
+ await self.store.get_last_event_pos_in_room(
+ room_id,
+ event_types=SLIDING_SYNC_DEFAULT_BUMP_EVENT_TYPES,
+ )
+ )
+ bump_stamp_to_fully_insert = (
+ most_recent_bump_event_pos_results[1].stream
+ if most_recent_bump_event_pos_results is not None
+ else None
+ )
+
+ current_state_ids_map = dict(
+ await self.store.get_partial_filtered_current_state_ids(
+ room_id,
+ state_filter=StateFilter.from_types(
+ SLIDING_SYNC_RELEVANT_STATE_SET
+ ),
+ )
+ )
+
+ # Look through the items we're going to insert into the current state to see
+ # if there is anything that we care about and should also update in the
+ # `sliding_sync_joined_rooms` table.
+ for state_key, event_id in to_insert.items():
+ if state_key in SLIDING_SYNC_RELEVANT_STATE_SET:
+ current_state_ids_map[state_key] = event_id
+
+ # Get the full event objects for the current state events
+ #
+ # In normal event persist scenarios, we should be able to find the state
+ # events in the `events_and_contexts` given to us but it's possible a state
+ # reset happened which that reset back to a previous state.
+ current_state_map = {}
+ missing_event_ids: Set[str] = set()
+ for state_key, event_id in current_state_ids_map.items():
+ event = event_map.get(event_id)
+ if event:
+ current_state_map[state_key] = event
+ else:
+ missing_event_ids.add(event_id)
+
+ # Otherwise, we need to find a couple events that we were reset to.
+ if missing_event_ids:
+ remaining_events = await self.store.get_events(
+ current_state_ids_map.values()
+ )
+ # There shouldn't be any missing events
+ assert (
+ remaining_events.keys() == missing_event_ids
+ ), missing_event_ids.difference(remaining_events.keys())
+ for event in remaining_events.values():
+ current_state_map[(event.type, event.state_key)] = event
+
+ joined_room_updates = (
+ PersistEventsStore._get_sliding_sync_insert_values_from_state_map(
+ current_state_map
+ )
+ )
+
+ # If something is being deleted from the state, we need to clear it out
+ for state_key in to_delete:
+ if state_key == (EventTypes.Create, ""):
+ joined_room_updates["room_type"] = None
+ elif state_key == (EventTypes.RoomEncryption, ""):
+ joined_room_updates["is_encrypted"] = False
+ elif state_key == (EventTypes.Name, ""):
+ joined_room_updates["room_name"] = None
+
+ # Figure out `best_effort_most_recent_stream_ordering`. This doesn't need to
+ # be perfect, we just need *some* answer that points to a real event in the
+ # room in case we are the first ones inserting into the
+ # `sliding_sync_joined_rooms` table because of the `NON NULL` constraint on
+ # `event_stream_ordering`. In reality,
+ # `_update_sliding_sync_tables_with_new_persisted_events_txn()` is run after
+ # `_update_current_state_txn()` whenever a new event is persisted to update
+ # it to the correct latest value.
+ #
+ if len(events_and_contexts) > 0:
+ # Since the list is sorted ascending by `stream_ordering`, the last event
+ # should have the highest `stream_ordering`.
+ best_effort_most_recent_stream_ordering = events_and_contexts[-1][
+ 0
+ ].internal_metadata.stream_ordering
+ else:
+ # If there are no `events_and_contexts`, we assume it's one of two scenarios:
+ # 1. If there are new state `to_insert` but no `events_and_contexts`,
+ # then it's a state reset.
+ # 2. Otherwise, it's some partial-state room re-syncing the current state and
+ # going through un-partial process.
+ #
+ # Either way, we assume no new events are being persisted and we can
+ # find the latest already in the database. Since this is a best-effort
+ # value, we don't need to be perfect although I think we're pretty close
+ # here.
+ most_recent_event_pos_results = (
+ await self.store.get_last_event_pos_in_room(
+ room_id, event_types=None
+ )
+ )
+ assert most_recent_event_pos_results, (
+ f"We should not be seeing `None` here because we are still in the room ({room_id}) and "
+ + "it should at-least have a join membership event that's keeping us here."
+ )
+ best_effort_most_recent_stream_ordering = most_recent_event_pos_results[
+ 1
+ ].stream
+
+ # We should have found a value if we are still in the room
+ assert best_effort_most_recent_stream_ordering is not None
+
+ return SlidingSyncTableChanges(
+ room_id=room_id,
+ # For `sliding_sync_joined_rooms`
+ joined_room_best_effort_most_recent_stream_ordering=best_effort_most_recent_stream_ordering,
+ joined_room_bump_stamp_to_fully_insert=bump_stamp_to_fully_insert,
+ joined_room_updates=joined_room_updates,
+ # For `sliding_sync_membership_snapshots`
+ membership_snapshot_shared_insert_values=membership_snapshot_shared_insert_values,
+ to_insert_membership_snapshots=membership_infos_to_insert_membership_snapshots,
+ to_delete_membership_snapshots=user_ids_to_delete_membership_snapshots,
+ )
+
async def calculate_chain_cover_index_for_events(
self, room_id: str, events: Collection[EventBase]
) -> Dict[str, NewEventChainLinks]:
@@ -458,6 +908,7 @@ class PersistEventsStore:
state_delta_for_room: Optional[DeltaState],
new_forward_extremities: Optional[Set[str]],
new_event_links: Dict[str, NewEventChainLinks],
+ sliding_sync_table_changes: Optional[SlidingSyncTableChanges],
) -> None:
"""Insert some number of room events into the necessary database tables.
@@ -478,9 +929,14 @@ class PersistEventsStore:
delete_existing True to purge existing table rows for the events
from the database. This is useful when retrying due to
IntegrityError.
- state_delta_for_room: The current-state delta for the room.
+ state_delta_for_room: Deltas that are going to be used to update the
+ `current_state_events` table. Changes to the current state of the room.
new_forward_extremities: The new forward extremities for the room:
a set of the event ids which are the forward extremities.
+ sliding_sync_table_changes: Changes to the
+ `sliding_sync_membership_snapshots` and `sliding_sync_joined_rooms` tables
+ derived from the given `delta_state` (see
+ `_calculate_sliding_sync_table_changes(...)`)
Raises:
PartialStateConflictError: if attempting to persist a partial state event in
@@ -590,10 +1046,22 @@ class PersistEventsStore:
# room_memberships, where applicable.
# NB: This function invalidates all state related caches
if state_delta_for_room:
+ # If the state delta exists, the sliding sync table changes should also exist
+ assert sliding_sync_table_changes is not None
+
self._update_current_state_txn(
- txn, room_id, state_delta_for_room, min_stream_order
+ txn,
+ room_id,
+ state_delta_for_room,
+ min_stream_order,
+ sliding_sync_table_changes,
)
+ # We only update the sliding sync tables for non-backfilled events.
+ self._update_sliding_sync_tables_with_new_persisted_events_txn(
+ txn, room_id, events_and_contexts
+ )
+
def _persist_event_auth_chain_txn(
self,
txn: LoggingTransaction,
@@ -1128,8 +1596,20 @@ class PersistEventsStore:
self,
room_id: str,
state_delta: DeltaState,
+ sliding_sync_table_changes: SlidingSyncTableChanges,
) -> None:
- """Update the current state stored in the datatabase for the given room"""
+ """
+ Update the current state stored in the datatabase for the given room
+
+ Args:
+ room_id
+ state_delta: Deltas that are going to be used to update the
+ `current_state_events` table. Changes to the current state of the room.
+ sliding_sync_table_changes: Changes to the
+ `sliding_sync_membership_snapshots` and `sliding_sync_joined_rooms` tables
+ derived from the given `delta_state` (see
+ `_calculate_sliding_sync_table_changes(...)`)
+ """
if state_delta.is_noop():
return
@@ -1141,6 +1621,7 @@ class PersistEventsStore:
room_id,
delta_state=state_delta,
stream_id=stream_ordering,
+ sliding_sync_table_changes=sliding_sync_table_changes,
)
def _update_current_state_txn(
@@ -1149,16 +1630,34 @@ class PersistEventsStore:
room_id: str,
delta_state: DeltaState,
stream_id: int,
+ sliding_sync_table_changes: SlidingSyncTableChanges,
) -> None:
+ """
+ Handles updating tables that track the current state of a room.
+
+ Args:
+ txn
+ room_id
+ delta_state: Deltas that are going to be used to update the
+ `current_state_events` table. Changes to the current state of the room.
+ stream_id: TODO
+ sliding_sync_table_changes: Changes to the
+ `sliding_sync_membership_snapshots` and `sliding_sync_joined_rooms` tables
+ derived from the given `delta_state` (see
+ `_calculate_sliding_sync_table_changes(...)`)
+ """
to_delete = delta_state.to_delete
to_insert = delta_state.to_insert
+ # Sanity check we're processing the same thing
+ assert room_id == sliding_sync_table_changes.room_id
+
# Figure out the changes of membership to invalidate the
# `get_rooms_for_user` cache.
# We find out which membership events we may have deleted
# and which we have added, then we invalidate the caches for all
# those users.
- members_changed = {
+ members_to_cache_bust = {
state_key
for ev_type, state_key in itertools.chain(to_delete, to_insert)
if ev_type == EventTypes.Member
@@ -1182,16 +1681,22 @@ class PersistEventsStore:
"""
txn.execute(sql, (stream_id, self._instance_name, room_id))
+ # Grab the list of users before we clear out the current state
+ users_in_room = self.store.get_users_in_room_txn(txn, room_id)
# We also want to invalidate the membership caches for users
# that were in the room.
- users_in_room = self.store.get_users_in_room_txn(txn, room_id)
- members_changed.update(users_in_room)
+ members_to_cache_bust.update(users_in_room)
self.db_pool.simple_delete_txn(
txn,
table="current_state_events",
keyvalues={"room_id": room_id},
)
+ self.db_pool.simple_delete_txn(
+ txn,
+ table="sliding_sync_joined_rooms",
+ keyvalues={"room_id": room_id},
+ )
else:
# We're still in the room, so we update the current state as normal.
@@ -1260,6 +1765,41 @@ class PersistEventsStore:
],
)
+ # Handle updating the `sliding_sync_joined_rooms` table. We only deal with
+ # updating the state related columns. The
+ # `event_stream_ordering`/`bump_stamp` are updated elsewhere in the event
+ # persisting stack (see
+ # `_update_sliding_sync_tables_with_new_persisted_events_txn()`)
+ #
+ # We only need to update when one of the relevant state values has changed
+ if sliding_sync_table_changes.joined_room_updates:
+ # This should be *some* value that points to a real event in the room if
+ # we are still joined to the room.
+ assert (
+ sliding_sync_table_changes.joined_room_best_effort_most_recent_stream_ordering
+ is not None
+ )
+
+ self.db_pool.simple_upsert_txn(
+ txn,
+ table="sliding_sync_joined_rooms",
+ keyvalues={"room_id": room_id},
+ values=sliding_sync_table_changes.joined_room_updates,
+ insertion_values={
+ # The reason we're only *inserting* (not *updating*)
+ # `event_stream_ordering` here is because the column has a `NON
+ # NULL` constraint and we need *some* answer. And if the row
+ # already exists, it already has the correct value and it's
+ # better to just rely on
+ # `_update_sliding_sync_tables_with_new_persisted_events_txn()`
+ # to do the right thing (same for `bump_stamp`).
+ "event_stream_ordering": sliding_sync_table_changes.joined_room_best_effort_most_recent_stream_ordering,
+ # If we're trying to fully-insert a row, we need to provide a
+ # value for `bump_stamp` if it exists for the room.
+ "bump_stamp": sliding_sync_table_changes.joined_room_bump_stamp_to_fully_insert,
+ },
+ )
+
# We now update `local_current_membership`. We do this regardless
# of whether we're still in the room or not to handle the case where
# e.g. we just got banned (where we need to record that fact here).
@@ -1296,6 +1836,60 @@ class PersistEventsStore:
],
)
+ # Handle updating the `sliding_sync_membership_snapshots` table
+ #
+ # This would only happen if someone was state reset out of the room
+ if sliding_sync_table_changes.to_delete_membership_snapshots:
+ self.db_pool.simple_delete_many_txn(
+ txn,
+ table="sliding_sync_membership_snapshots",
+ column="user_id",
+ values=sliding_sync_table_changes.to_delete_membership_snapshots,
+ keyvalues={"room_id": room_id},
+ )
+
+ # We do this regardless of whether the server is `no_longer_in_room` or not
+ # because we still want a row if a local user was just left/kicked or got banned
+ # from the room.
+ if sliding_sync_table_changes.to_insert_membership_snapshots:
+ # Update the `sliding_sync_membership_snapshots` table
+ #
+ # We need to insert/update regardless of whether we have `sliding_sync_snapshot_keys`
+ # because there are other fields in the `ON CONFLICT` upsert to run (see
+ # inherit case above for more context when this happens).
+ self.db_pool.simple_upsert_many_txn(
+ txn=txn,
+ table="sliding_sync_membership_snapshots",
+ key_names=("room_id", "user_id"),
+ key_values=[
+ (room_id, membership_info.user_id)
+ for membership_info in sliding_sync_table_changes.to_insert_membership_snapshots
+ ],
+ value_names=[
+ "sender",
+ "membership_event_id",
+ "membership",
+ "event_stream_ordering",
+ "event_instance_name",
+ ]
+ + list(
+ sliding_sync_table_changes.membership_snapshot_shared_insert_values.keys()
+ ),
+ value_values=[
+ [
+ membership_info.sender,
+ membership_info.membership_event_id,
+ membership_info.membership,
+ membership_info.membership_event_stream_ordering,
+ membership_info.membership_event_instance_name,
+ ]
+ + list(
+ sliding_sync_table_changes.membership_snapshot_shared_insert_values.values()
+ )
+ for membership_info in sliding_sync_table_changes.to_insert_membership_snapshots
+ ],
+ )
+
txn.call_after(
self.store._curr_state_delta_stream_cache.entity_has_changed,
room_id,
@@ -1303,13 +1897,302 @@ class PersistEventsStore:
)
# Invalidate the various caches
- self.store._invalidate_state_caches_and_stream(txn, room_id, members_changed)
+ self.store._invalidate_state_caches_and_stream(
+ txn, room_id, members_to_cache_bust
+ )
# Check if any of the remote membership changes requires us to
# unsubscribe from their device lists.
self.store.handle_potentially_left_users_txn(
- txn, {m for m in members_changed if not self.hs.is_mine_id(m)}
+ txn, {m for m in members_to_cache_bust if not self.hs.is_mine_id(m)}
+ )
+
+ @classmethod
+ def _get_relevant_sliding_sync_current_state_event_ids_txn(
+ cls, txn: LoggingTransaction, room_id: str
+ ) -> MutableStateMap[str]:
+ """
+ Fetch the current state event IDs for the relevant (to the
+ `sliding_sync_joined_rooms` table) state types for the given room.
+
+ Returns:
+ A tuple of:
+ 1. StateMap of event IDs necessary to to fetch the relevant state values
+ needed to insert into the
+ `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots`.
+ 2. The corresponding latest `stream_id` in the
+ `current_state_delta_stream` table. This is useful to compare against
+ the `current_state_delta_stream` table later so you can check whether
+ the current state has changed since you last fetched the current
+ state.
+ """
+ # Fetch the current state event IDs from the database
+ (
+ event_type_and_state_key_in_list_clause,
+ event_type_and_state_key_args,
+ ) = make_tuple_in_list_sql_clause(
+ txn.database_engine,
+ ("type", "state_key"),
+ SLIDING_SYNC_RELEVANT_STATE_SET,
+ )
+ txn.execute(
+ f"""
+ SELECT c.event_id, c.type, c.state_key
+ FROM current_state_events AS c
+ WHERE
+ c.room_id = ?
+ AND {event_type_and_state_key_in_list_clause}
+ """,
+ [room_id] + event_type_and_state_key_args,
+ )
+ current_state_map: MutableStateMap[str] = {
+ (event_type, state_key): event_id for event_id, event_type, state_key in txn
+ }
+
+ return current_state_map
+
+ @classmethod
+ def _get_sliding_sync_insert_values_from_state_map(
+ cls, state_map: StateMap[EventBase]
+ ) -> SlidingSyncStateInsertValues:
+ """
+ Extract the relevant state values from the `state_map` needed to insert into the
+ `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` tables.
+
+ Returns:
+ Map from column names (`room_type`, `is_encrypted`, `room_name`) to relevant
+ state values needed to insert into
+ the `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` tables.
+ """
+ # Map of values to insert/update in the `sliding_sync_membership_snapshots` table
+ sliding_sync_insert_map: SlidingSyncStateInsertValues = {}
+
+ # Parse the raw event JSON
+ for state_key, event in state_map.items():
+ if state_key == (EventTypes.Create, ""):
+ room_type = event.content.get(EventContentFields.ROOM_TYPE)
+ # Scrutinize JSON values
+ if room_type is None or isinstance(room_type, str):
+ sliding_sync_insert_map["room_type"] = room_type
+ elif state_key == (EventTypes.RoomEncryption, ""):
+ encryption_algorithm = event.content.get(
+ EventContentFields.ENCRYPTION_ALGORITHM
+ )
+ is_encrypted = encryption_algorithm is not None
+ sliding_sync_insert_map["is_encrypted"] = is_encrypted
+ elif state_key == (EventTypes.Name, ""):
+ room_name = event.content.get(EventContentFields.ROOM_NAME)
+ # Scrutinize JSON values
+ if room_name is None or isinstance(room_name, str):
+ sliding_sync_insert_map["room_name"] = room_name
+ elif state_key == (EventTypes.Tombstone, ""):
+ successor_room_id = event.content.get(
+ EventContentFields.TOMBSTONE_SUCCESSOR_ROOM
+ )
+ # Scrutinize JSON values
+ if successor_room_id is None or isinstance(successor_room_id, str):
+ sliding_sync_insert_map["tombstone_successor_room_id"] = (
+ successor_room_id
+ )
+ else:
+ # We only expect to see events according to the
+ # `SLIDING_SYNC_RELEVANT_STATE_SET`.
+ raise AssertionError(
+ "Unexpected event (we should not be fetching extra events or this "
+ + "piece of code needs to be updated to handle a new event type added "
+ + "to `SLIDING_SYNC_RELEVANT_STATE_SET`): {state_key} {event.event_id}"
+ )
+
+ return sliding_sync_insert_map
+
+ @classmethod
+ def _get_sliding_sync_insert_values_from_stripped_state(
+ cls, unsigned_stripped_state_events: Any
+ ) -> SlidingSyncMembershipSnapshotSharedInsertValues:
+ """
+ Pull out the relevant state values from the stripped state on an invite or knock
+ membership event needed to insert into the `sliding_sync_membership_snapshots`
+ tables.
+
+ Returns:
+ Map from column names (`room_type`, `is_encrypted`, `room_name`) to relevant
+ state values needed to insert into the `sliding_sync_membership_snapshots` tables.
+ """
+ # Map of values to insert/update in the `sliding_sync_membership_snapshots` table
+ sliding_sync_insert_map: SlidingSyncMembershipSnapshotSharedInsertValues = {}
+
+ if unsigned_stripped_state_events is not None:
+ stripped_state_map: MutableStateMap[StrippedStateEvent] = {}
+ if isinstance(unsigned_stripped_state_events, list):
+ for raw_stripped_event in unsigned_stripped_state_events:
+ stripped_state_event = parse_stripped_state_event(
+ raw_stripped_event
+ )
+ if stripped_state_event is not None:
+ stripped_state_map[
+ (
+ stripped_state_event.type,
+ stripped_state_event.state_key,
+ )
+ ] = stripped_state_event
+
+ # If there is some stripped state, we assume the remote server passed *all*
+ # of the potential stripped state events for the room.
+ create_stripped_event = stripped_state_map.get((EventTypes.Create, ""))
+ # Sanity check that we at-least have the create event
+ if create_stripped_event is not None:
+ sliding_sync_insert_map["has_known_state"] = True
+
+ # XXX: Keep this up-to-date with `SLIDING_SYNC_RELEVANT_STATE_SET`
+
+ # Find the room_type
+ sliding_sync_insert_map["room_type"] = (
+ create_stripped_event.content.get(EventContentFields.ROOM_TYPE)
+ if create_stripped_event is not None
+ else None
+ )
+
+ # Find whether the room is_encrypted
+ encryption_stripped_event = stripped_state_map.get(
+ (EventTypes.RoomEncryption, "")
+ )
+ encryption = (
+ encryption_stripped_event.content.get(
+ EventContentFields.ENCRYPTION_ALGORITHM
+ )
+ if encryption_stripped_event is not None
+ else None
+ )
+ sliding_sync_insert_map["is_encrypted"] = encryption is not None
+
+ # Find the room_name
+ room_name_stripped_event = stripped_state_map.get((EventTypes.Name, ""))
+ sliding_sync_insert_map["room_name"] = (
+ room_name_stripped_event.content.get(EventContentFields.ROOM_NAME)
+ if room_name_stripped_event is not None
+ else None
+ )
+
+ # Find the tombstone_successor_room_id
+ # Note: This isn't one of the stripped state events according to the spec
+ # but seems like there is no reason not to support this kind of thing.
+ tombstone_stripped_event = stripped_state_map.get(
+ (EventTypes.Tombstone, "")
+ )
+ sliding_sync_insert_map["tombstone_successor_room_id"] = (
+ tombstone_stripped_event.content.get(
+ EventContentFields.TOMBSTONE_SUCCESSOR_ROOM
+ )
+ if tombstone_stripped_event is not None
+ else None
+ )
+
+ else:
+ # No stripped state provided
+ sliding_sync_insert_map["has_known_state"] = False
+ sliding_sync_insert_map["room_type"] = None
+ sliding_sync_insert_map["room_name"] = None
+ sliding_sync_insert_map["is_encrypted"] = False
+ else:
+ # No stripped state provided
+ sliding_sync_insert_map["has_known_state"] = False
+ sliding_sync_insert_map["room_type"] = None
+ sliding_sync_insert_map["room_name"] = None
+ sliding_sync_insert_map["is_encrypted"] = False
+
+ return sliding_sync_insert_map
+
+ def _update_sliding_sync_tables_with_new_persisted_events_txn(
+ self,
+ txn: LoggingTransaction,
+ room_id: str,
+ events_and_contexts: List[Tuple[EventBase, EventContext]],
+ ) -> None:
+ """
+ Update the latest `event_stream_ordering`/`bump_stamp` columns in the
+ `sliding_sync_joined_rooms` table for the room with new events.
+
+ This function assumes that `_store_event_txn()` (to persist the event) and
+ `_update_current_state_txn(...)` (so that `sliding_sync_joined_rooms` table has
+ been updated with rooms that were joined) have already been run.
+
+ Args:
+ txn
+ room_id: The room that all of the events belong to
+ events_and_contexts: The events being persisted. We assume the list is
+ sorted ascending by `stream_ordering`. We don't care about the sort when the
+ events are backfilled (with negative `stream_ordering`).
+ """
+
+ # Nothing to do if there are no events
+ if len(events_and_contexts) == 0:
+ return
+
+ # We only update the sliding sync tables for non-backfilled events.
+ #
+ # Check if the first event is a backfilled event (with a negative
+ # `stream_ordering`). If one event is backfilled, we assume this whole batch was
+ # backfilled.
+ first_event_stream_ordering = events_and_contexts[0][
+ 0
+ ].internal_metadata.stream_ordering
+ # This should exist for persisted events
+ assert first_event_stream_ordering is not None
+ if first_event_stream_ordering < 0:
+ return
+
+ # Since the list is sorted ascending by `stream_ordering`, the last event should
+ # have the highest `stream_ordering`.
+ max_stream_ordering = events_and_contexts[-1][
+ 0
+ ].internal_metadata.stream_ordering
+ max_bump_stamp = None
+ for event, _ in reversed(events_and_contexts):
+ # Sanity check that all events belong to the same room
+ assert event.room_id == room_id
+
+ if event.type in SLIDING_SYNC_DEFAULT_BUMP_EVENT_TYPES:
+ # This should exist for persisted events
+ assert event.internal_metadata.stream_ordering is not None
+
+ max_bump_stamp = event.internal_metadata.stream_ordering
+
+ # Since we're iterating in reverse, we can break as soon as we find a
+ # matching bump event which should have the highest `stream_ordering`.
+ break
+
+ # We should have exited earlier if there were no events
+ assert (
+ max_stream_ordering is not None
+ ), "Expected to have a stream_ordering if we have events"
+
+ # Handle updating the `sliding_sync_joined_rooms` table.
+ #
+ txn.execute(
+ """
+ UPDATE sliding_sync_joined_rooms
+ SET
+ event_stream_ordering = CASE
+ WHEN event_stream_ordering IS NULL OR event_stream_ordering < ?
+ THEN ?
+ ELSE event_stream_ordering
+ END,
+ bump_stamp = CASE
+ WHEN bump_stamp IS NULL OR bump_stamp < ?
+ THEN ?
+ ELSE bump_stamp
+ END
+ WHERE room_id = ?
+ """,
+ (
+ max_stream_ordering,
+ max_stream_ordering,
+ max_bump_stamp,
+ max_bump_stamp,
+ room_id,
+ ),
)
+ # This may or may not update any rows depending if we are `no_longer_in_room`
def _upsert_room_version_txn(self, txn: LoggingTransaction, room_id: str) -> None:
"""Update the room version in the database based off current state
@@ -1931,7 +2814,9 @@ class PersistEventsStore:
)
for event in events:
+ # Sanity check that we're working with persisted events
assert event.internal_metadata.stream_ordering is not None
+ assert event.internal_metadata.instance_name is not None
# We update the local_current_membership table only if the event is
# "current", i.e., its something that has just happened.
@@ -1945,6 +2830,16 @@ class PersistEventsStore:
and event.internal_metadata.is_outlier()
and event.internal_metadata.is_out_of_band_membership()
):
+ # The only sort of out-of-band-membership events we expect to see here
+ # are remote invites/knocks and LEAVE events corresponding to
+ # rejected/retracted invites and rescinded knocks.
+ assert event.type == EventTypes.Member
+ assert event.membership in (
+ Membership.INVITE,
+ Membership.KNOCK,
+ Membership.LEAVE,
+ )
+
self.db_pool.simple_upsert_txn(
txn,
table="local_current_membership",
@@ -1956,6 +2851,56 @@ class PersistEventsStore:
},
)
+ # Handle updating the `sliding_sync_membership_snapshots` table
+ # (out-of-band membership events only)
+ #
+ raw_stripped_state_events = None
+ if event.membership == Membership.INVITE:
+ invite_room_state = event.unsigned.get("invite_room_state")
+ raw_stripped_state_events = invite_room_state
+ elif event.membership == Membership.KNOCK:
+ knock_room_state = event.unsigned.get("knock_room_state")
+ raw_stripped_state_events = knock_room_state
+
+ insert_values = {
+ "sender": event.sender,
+ "membership_event_id": event.event_id,
+ "membership": event.membership,
+ "event_stream_ordering": event.internal_metadata.stream_ordering,
+ "event_instance_name": event.internal_metadata.instance_name,
+ }
+ if event.membership == Membership.LEAVE:
+ # Inherit the meta data from the remote invite/knock. When using
+ # sliding sync filters, this will prevent the room from
+ # disappearing/appearing just because you left the room.
+ pass
+ elif event.membership in (Membership.INVITE, Membership.KNOCK):
+ extra_insert_values = (
+ self._get_sliding_sync_insert_values_from_stripped_state(
+ raw_stripped_state_events
+ )
+ )
+ insert_values.update(extra_insert_values)
+ else:
+ # We don't know how to handle this type of membership yet
+ #
+ # FIXME: We should use `assert_never` here but for some reason
+ # the exhaustive matching doesn't recognize the `Never` here.
+ # assert_never(event.membership)
+ raise AssertionError(
+ f"Unexpected out-of-band membership {event.membership} ({event.event_id}) that we don't know how to handle yet"
+ )
+
+ self.db_pool.simple_upsert_txn(
+ txn,
+ table="sliding_sync_membership_snapshots",
+ keyvalues={
+ "room_id": event.room_id,
+ "user_id": event.state_key,
+ },
+ values=insert_values,
+ )
+
def _handle_event_relations(
self, txn: LoggingTransaction, event: EventBase
) -> None:
diff --git a/synapse/storage/databases/main/events_bg_updates.py b/synapse/storage/databases/main/events_bg_updates.py
index 64d303e330..88ff5aa2df 100644
--- a/synapse/storage/databases/main/events_bg_updates.py
+++ b/synapse/storage/databases/main/events_bg_updates.py
@@ -24,9 +24,9 @@ from typing import TYPE_CHECKING, Dict, List, Optional, Set, Tuple, cast
import attr
-from synapse.api.constants import EventContentFields, RelationTypes
+from synapse.api.constants import EventContentFields, Membership, RelationTypes
from synapse.api.room_versions import KNOWN_ROOM_VERSIONS
-from synapse.events import make_event_from_dict
+from synapse.events import EventBase, make_event_from_dict
from synapse.storage._base import SQLBaseStore, db_to_json, make_in_list_sql_clause
from synapse.storage.database import (
DatabasePool,
@@ -34,9 +34,21 @@ from synapse.storage.database import (
LoggingTransaction,
make_tuple_comparison_clause,
)
-from synapse.storage.databases.main.events import PersistEventsStore
+from synapse.storage.databases.main.events import (
+ SLIDING_SYNC_RELEVANT_STATE_SET,
+ PersistEventsStore,
+ SlidingSyncMembershipInfo,
+ SlidingSyncMembershipSnapshotSharedInsertValues,
+ SlidingSyncStateInsertValues,
+)
+from synapse.storage.databases.main.state_deltas import StateDeltasStore
+from synapse.storage.databases.main.stream import StreamWorkerStore
from synapse.storage.types import Cursor
-from synapse.types import JsonDict, StrCollection
+from synapse.types import JsonDict, RoomStreamToken, StateMap, StrCollection
+from synapse.types.handlers import SLIDING_SYNC_DEFAULT_BUMP_EVENT_TYPES
+from synapse.types.state import StateFilter
+from synapse.util import json_encoder
+from synapse.util.iterutils import batch_iter
if TYPE_CHECKING:
from synapse.server import HomeServer
@@ -78,6 +90,14 @@ class _BackgroundUpdates:
EVENTS_JUMP_TO_DATE_INDEX = "events_jump_to_date_index"
+ SLIDING_SYNC_PREFILL_JOINED_ROOMS_TO_RECALCULATE_TABLE_BG_UPDATE = (
+ "sliding_sync_prefill_joined_rooms_to_recalculate_table_bg_update"
+ )
+ SLIDING_SYNC_JOINED_ROOMS_BG_UPDATE = "sliding_sync_joined_rooms_bg_update"
+ SLIDING_SYNC_MEMBERSHIP_SNAPSHOTS_BG_UPDATE = (
+ "sliding_sync_membership_snapshots_bg_update"
+ )
+
@attr.s(slots=True, frozen=True, auto_attribs=True)
class _CalculateChainCover:
@@ -97,7 +117,19 @@ class _CalculateChainCover:
finished_room_map: Dict[str, Tuple[int, int]]
-class EventsBackgroundUpdatesStore(SQLBaseStore):
+@attr.s(slots=True, frozen=True, auto_attribs=True)
+class _JoinedRoomStreamOrderingUpdate:
+ """
+ Intermediate container class used in `SLIDING_SYNC_JOINED_ROOMS_BG_UPDATE`
+ """
+
+ # The most recent event stream_ordering for the room
+ most_recent_event_stream_ordering: int
+ # The most recent event `bump_stamp` for the room
+ most_recent_bump_stamp: Optional[int]
+
+
+class EventsBackgroundUpdatesStore(StreamWorkerStore, StateDeltasStore, SQLBaseStore):
def __init__(
self,
database: DatabasePool,
@@ -279,6 +311,34 @@ class EventsBackgroundUpdatesStore(SQLBaseStore):
where_clause="NOT outlier",
)
+ # Handle background updates for Sliding Sync tables
+ #
+ self.db_pool.updates.register_background_update_handler(
+ _BackgroundUpdates.SLIDING_SYNC_PREFILL_JOINED_ROOMS_TO_RECALCULATE_TABLE_BG_UPDATE,
+ self._sliding_sync_prefill_joined_rooms_to_recalculate_table_bg_update,
+ )
+ # Add some background updates to populate the sliding sync tables
+ self.db_pool.updates.register_background_update_handler(
+ _BackgroundUpdates.SLIDING_SYNC_JOINED_ROOMS_BG_UPDATE,
+ self._sliding_sync_joined_rooms_bg_update,
+ )
+ self.db_pool.updates.register_background_update_handler(
+ _BackgroundUpdates.SLIDING_SYNC_MEMBERSHIP_SNAPSHOTS_BG_UPDATE,
+ self._sliding_sync_membership_snapshots_bg_update,
+ )
+
+ # We want this to run on the main database at startup before we start processing
+ # events.
+ #
+ # FIXME: This can be removed once we bump `SCHEMA_COMPAT_VERSION` and run the
+ # foreground update for
+ # `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` (tracked by
+ # https://github.com/element-hq/synapse/issues/17623)
+ with db_conn.cursor(txn_name="resolve_sliding_sync") as txn:
+ _resolve_stale_data_in_sliding_sync_tables(
+ txn=txn,
+ )
+
async def _background_reindex_fields_sender(
self, progress: JsonDict, batch_size: int
) -> int:
@@ -1073,7 +1133,7 @@ class EventsBackgroundUpdatesStore(SQLBaseStore):
PersistEventsStore._add_chain_cover_index(
txn,
self.db_pool,
- self.event_chain_id_gen, # type: ignore[attr-defined]
+ self.event_chain_id_gen,
event_to_room_id,
event_to_types,
cast(Dict[str, StrCollection], event_to_auth_chain),
@@ -1516,3 +1576,961 @@ class EventsBackgroundUpdatesStore(SQLBaseStore):
)
return batch_size
+
+ async def _sliding_sync_prefill_joined_rooms_to_recalculate_table_bg_update(
+ self, progress: JsonDict, _batch_size: int
+ ) -> int:
+ """
+ Prefill `sliding_sync_joined_rooms_to_recalculate` table with all rooms we know about already.
+ """
+
+ def _txn(txn: LoggingTransaction) -> None:
+ # We do this as one big bulk insert. This has been tested on a bigger
+ # homeserver with ~10M rooms and took 60s. There is potential for this to
+ # starve disk usage while this goes on.
+ #
+ # We upsert in case we have to run this multiple times.
+ #
+ # The `WHERE TRUE` clause is to avoid "Parsing Ambiguity"
+ txn.execute(
+ """
+ INSERT INTO sliding_sync_joined_rooms_to_recalculate
+ (room_id)
+ SELECT room_id FROM rooms WHERE ?
+ ON CONFLICT (room_id)
+ DO NOTHING;
+ """,
+ (True,),
+ )
+
+ await self.db_pool.runInteraction(
+ "_sliding_sync_prefill_joined_rooms_to_recalculate_table_bg_update",
+ _txn,
+ )
+
+ # Background update is done.
+ await self.db_pool.updates._end_background_update(
+ _BackgroundUpdates.SLIDING_SYNC_PREFILL_JOINED_ROOMS_TO_RECALCULATE_TABLE_BG_UPDATE
+ )
+ return 0
+
+ async def _sliding_sync_joined_rooms_bg_update(
+ self, progress: JsonDict, batch_size: int
+ ) -> int:
+ """
+ Background update to populate the `sliding_sync_joined_rooms` table.
+ """
+ # We don't need to fetch any progress state because we just grab the next N
+ # events in `sliding_sync_joined_rooms_to_recalculate`
+
+ def _get_rooms_to_update_txn(txn: LoggingTransaction) -> List[Tuple[str]]:
+ """
+ Returns:
+ A list of room ID's to update along with the progress value
+ (event_stream_ordering) indicating the continuation point in the
+ `current_state_events` table for the next batch.
+ """
+ # Fetch the set of room IDs that we want to update
+ #
+ # We use `current_state_events` table as the barometer for whether the
+ # server is still participating in the room because if we're
+ # `no_longer_in_room`, this table would be cleared out for the given
+ # `room_id`.
+ txn.execute(
+ """
+ SELECT room_id
+ FROM sliding_sync_joined_rooms_to_recalculate
+ LIMIT ?
+ """,
+ (batch_size,),
+ )
+
+ rooms_to_update_rows = cast(List[Tuple[str]], txn.fetchall())
+
+ return rooms_to_update_rows
+
+ rooms_to_update = await self.db_pool.runInteraction(
+ "_sliding_sync_joined_rooms_bg_update._get_rooms_to_update_txn",
+ _get_rooms_to_update_txn,
+ )
+
+ if not rooms_to_update:
+ await self.db_pool.updates._end_background_update(
+ _BackgroundUpdates.SLIDING_SYNC_JOINED_ROOMS_BG_UPDATE
+ )
+ return 0
+
+ # Map from room_id to insert/update state values in the `sliding_sync_joined_rooms` table.
+ joined_room_updates: Dict[str, SlidingSyncStateInsertValues] = {}
+ # Map from room_id to stream_ordering/bump_stamp, etc values
+ joined_room_stream_ordering_updates: Dict[
+ str, _JoinedRoomStreamOrderingUpdate
+ ] = {}
+ # As long as we get this value before we fetch the current state, we can use it
+ # to check if something has changed since that point.
+ most_recent_current_state_delta_stream_id = (
+ await self.get_max_stream_id_in_current_state_deltas()
+ )
+ for (room_id,) in rooms_to_update:
+ current_state_ids_map = await self.db_pool.runInteraction(
+ "_sliding_sync_joined_rooms_bg_update._get_relevant_sliding_sync_current_state_event_ids_txn",
+ PersistEventsStore._get_relevant_sliding_sync_current_state_event_ids_txn,
+ room_id,
+ )
+
+ # If we're not joined to the room a) it doesn't belong in the
+ # `sliding_sync_joined_rooms` table so we should skip and b) we won't have
+ # any `current_state_events` for the room.
+ if not current_state_ids_map:
+ continue
+
+ fetched_events = await self.get_events(current_state_ids_map.values())
+
+ current_state_map: StateMap[EventBase] = {
+ state_key: fetched_events[event_id]
+ for state_key, event_id in current_state_ids_map.items()
+ # `get_events(...)` will filter out events for unknown room versions
+ if event_id in fetched_events
+ }
+
+ # Even if we are joined to the room, this can happen for unknown room
+ # versions (old room versions that aren't known anymore) since
+ # `get_events(...)` will filter out events for unknown room versions
+ if not current_state_map:
+ continue
+
+ state_insert_values = (
+ PersistEventsStore._get_sliding_sync_insert_values_from_state_map(
+ current_state_map
+ )
+ )
+ # We should have some insert values for each room, even if they are `None`
+ assert state_insert_values
+ joined_room_updates[room_id] = state_insert_values
+
+ # Figure out the stream_ordering of the latest event in the room
+ most_recent_event_pos_results = await self.get_last_event_pos_in_room(
+ room_id, event_types=None
+ )
+ assert most_recent_event_pos_results is not None, (
+ f"We should not be seeing `None` here because the room ({room_id}) should at-least have a create event "
+ + "given we pulled the room out of `current_state_events`"
+ )
+ most_recent_event_stream_ordering = most_recent_event_pos_results[1].stream
+ assert most_recent_event_stream_ordering > 0, (
+ "We should have at-least one event in the room (our own join membership event for example) "
+ + "that isn't backfilled (negative `stream_ordering`) if we are joined to the room."
+ )
+ # Figure out the latest `bump_stamp` in the room. This could be `None` for a
+ # federated room you just joined where all of events are still `outliers` or
+ # backfilled history. In the Sliding Sync API, we default to the user's
+ # membership event `stream_ordering` if we don't have a `bump_stamp` so
+ # having it as `None` in this table is fine.
+ bump_stamp_event_pos_results = await self.get_last_event_pos_in_room(
+ room_id, event_types=SLIDING_SYNC_DEFAULT_BUMP_EVENT_TYPES
+ )
+ most_recent_bump_stamp = None
+ if (
+ bump_stamp_event_pos_results is not None
+ and bump_stamp_event_pos_results[1].stream > 0
+ ):
+ most_recent_bump_stamp = bump_stamp_event_pos_results[1].stream
+
+ joined_room_stream_ordering_updates[room_id] = (
+ _JoinedRoomStreamOrderingUpdate(
+ most_recent_event_stream_ordering=most_recent_event_stream_ordering,
+ most_recent_bump_stamp=most_recent_bump_stamp,
+ )
+ )
+
+ def _fill_table_txn(txn: LoggingTransaction) -> None:
+ # Handle updating the `sliding_sync_joined_rooms` table
+ #
+ for (
+ room_id,
+ update_map,
+ ) in joined_room_updates.items():
+ joined_room_stream_ordering_update = (
+ joined_room_stream_ordering_updates[room_id]
+ )
+ event_stream_ordering = (
+ joined_room_stream_ordering_update.most_recent_event_stream_ordering
+ )
+ bump_stamp = joined_room_stream_ordering_update.most_recent_bump_stamp
+
+ # Check if the current state has been updated since we gathered it.
+ # We're being careful not to insert/overwrite with stale data.
+ state_deltas_since_we_gathered_current_state = (
+ self.get_current_state_deltas_for_room_txn(
+ txn,
+ room_id,
+ from_token=RoomStreamToken(
+ stream=most_recent_current_state_delta_stream_id
+ ),
+ to_token=None,
+ )
+ )
+ for state_delta in state_deltas_since_we_gathered_current_state:
+ # We only need to check for the state is relevant to the
+ # `sliding_sync_joined_rooms` table.
+ if (
+ state_delta.event_type,
+ state_delta.state_key,
+ ) in SLIDING_SYNC_RELEVANT_STATE_SET:
+ # Raising exception so we can just exit and try again. It would
+ # be hard to resolve this within the transaction because we need
+ # to get full events out that take redactions into account. We
+ # could add some retry logic here, but it's easier to just let
+ # the background update try again.
+ raise Exception(
+ "Current state was updated after we gathered it to update "
+ + "`sliding_sync_joined_rooms` in the background update. "
+ + "Raising exception so we can just try again."
+ )
+
+ # Since we fully insert rows into `sliding_sync_joined_rooms`, we can
+ # just do everything on insert and `ON CONFLICT DO NOTHING`.
+ #
+ self.db_pool.simple_upsert_txn(
+ txn,
+ table="sliding_sync_joined_rooms",
+ keyvalues={"room_id": room_id},
+ values={},
+ insertion_values={
+ **update_map,
+ # The reason we're only *inserting* (not *updating*) `event_stream_ordering`
+ # and `bump_stamp` is because if they are present, that means they are already
+ # up-to-date.
+ "event_stream_ordering": event_stream_ordering,
+ "bump_stamp": bump_stamp,
+ },
+ )
+
+ # Now that we've processed all the room, we can remove them from the
+ # queue.
+ #
+ # Note: we need to remove all the rooms from the queue we pulled out
+ # from the DB, not just the ones we've processed above. Otherwise
+ # we'll simply keep pulling out the same rooms over and over again.
+ self.db_pool.simple_delete_many_batch_txn(
+ txn,
+ table="sliding_sync_joined_rooms_to_recalculate",
+ keys=("room_id",),
+ values=rooms_to_update,
+ )
+
+ await self.db_pool.runInteraction(
+ "sliding_sync_joined_rooms_bg_update", _fill_table_txn
+ )
+
+ return len(rooms_to_update)
+
+ async def _sliding_sync_membership_snapshots_bg_update(
+ self, progress: JsonDict, batch_size: int
+ ) -> int:
+ """
+ Background update to populate the `sliding_sync_membership_snapshots` table.
+ """
+ # We do this in two phases: a) the initial phase where we go through all
+ # room memberships, and then b) a second phase where we look at new
+ # memberships (this is to handle the case where we downgrade and then
+ # upgrade again).
+ #
+ # We have to do this as two phases (rather than just the second phase
+ # where we iterate on event_stream_ordering), as the
+ # `event_stream_ordering` column may have null values for old rows.
+ # Therefore we first do the set of historic rooms and *then* look at any
+ # new rows (which will have a non-null `event_stream_ordering`).
+ initial_phase = progress.get("initial_phase")
+ if initial_phase is None:
+ # If this is the first run, store the current max stream position.
+ # We know we will go through all memberships less than the current
+ # max in the initial phase.
+ progress = {
+ "initial_phase": True,
+ "last_event_stream_ordering": self.get_room_max_stream_ordering(),
+ }
+ await self.db_pool.updates._background_update_progress(
+ _BackgroundUpdates.SLIDING_SYNC_MEMBERSHIP_SNAPSHOTS_BG_UPDATE,
+ progress,
+ )
+ initial_phase = True
+
+ last_room_id = progress.get("last_room_id", "")
+ last_event_stream_ordering = progress["last_event_stream_ordering"]
+
+ def _find_memberships_to_update_txn(
+ txn: LoggingTransaction,
+ ) -> List[
+ Tuple[str, Optional[str], str, str, str, str, int, Optional[str], bool]
+ ]:
+ # Fetch the set of event IDs that we want to update
+
+ if initial_phase:
+ # There are some old out-of-band memberships (before
+ # https://github.com/matrix-org/synapse/issues/6983) where we don't have
+ # the corresponding room stored in the `rooms` table`. We use `LEFT JOIN
+ # rooms AS r USING (room_id)` to find the rooms missing from `rooms` and
+ # insert a row for them below.
+ txn.execute(
+ """
+ SELECT
+ c.room_id,
+ r.room_id,
+ c.user_id,
+ e.sender,
+ c.event_id,
+ c.membership,
+ e.stream_ordering,
+ e.instance_name,
+ e.outlier
+ FROM local_current_membership AS c
+ INNER JOIN events AS e USING (event_id)
+ LEFT JOIN rooms AS r ON (c.room_id = r.room_id)
+ WHERE c.room_id > ?
+ ORDER BY c.room_id ASC
+ LIMIT ?
+ """,
+ (last_room_id, batch_size),
+ )
+ elif last_event_stream_ordering is not None:
+ # It's important to sort by `event_stream_ordering` *ascending* (oldest to
+ # newest) so that if we see that this background update in progress and want
+ # to start the catch-up process, we can safely assume that it will
+ # eventually get to the rooms we want to catch-up on anyway (see
+ # `_resolve_stale_data_in_sliding_sync_tables()`).
+ #
+ # `c.room_id` is duplicated to make it match what we're doing in the
+ # `initial_phase`. But we can avoid doing the extra `rooms` table join
+ # because we can assume all of these new events won't have this problem.
+ txn.execute(
+ """
+ SELECT
+ c.room_id,
+ c.room_id,
+ c.user_id,
+ e.sender,
+ c.event_id,
+ c.membership,
+ c.event_stream_ordering,
+ e.instance_name,
+ e.outlier
+ FROM local_current_membership AS c
+ INNER JOIN events AS e USING (event_id)
+ WHERE event_stream_ordering > ?
+ ORDER BY event_stream_ordering ASC
+ LIMIT ?
+ """,
+ (last_event_stream_ordering, batch_size),
+ )
+ else:
+ raise Exception("last_event_stream_ordering should not be None")
+
+ memberships_to_update_rows = cast(
+ List[
+ Tuple[
+ str, Optional[str], str, str, str, str, int, Optional[str], bool
+ ]
+ ],
+ txn.fetchall(),
+ )
+
+ return memberships_to_update_rows
+
+ memberships_to_update_rows = await self.db_pool.runInteraction(
+ "sliding_sync_membership_snapshots_bg_update._find_memberships_to_update_txn",
+ _find_memberships_to_update_txn,
+ )
+
+ if not memberships_to_update_rows:
+ if initial_phase:
+ # Move onto the next phase.
+ await self.db_pool.updates._background_update_progress(
+ _BackgroundUpdates.SLIDING_SYNC_MEMBERSHIP_SNAPSHOTS_BG_UPDATE,
+ {
+ "initial_phase": False,
+ "last_event_stream_ordering": last_event_stream_ordering,
+ },
+ )
+ return 0
+ else:
+ # We've finished both phases, we're done.
+ await self.db_pool.updates._end_background_update(
+ _BackgroundUpdates.SLIDING_SYNC_MEMBERSHIP_SNAPSHOTS_BG_UPDATE
+ )
+ return 0
+
+ def _find_previous_membership_txn(
+ txn: LoggingTransaction, event_id: str, user_id: str
+ ) -> Tuple[str, str]:
+ # Find the previous invite/knock event before the leave event. This
+ # is done by looking at the auth events of the invite/knock and
+ # finding the corresponding membership event.
+ txn.execute(
+ """
+ SELECT m.event_id, m.membership
+ FROM event_auth AS a
+ INNER JOIN room_memberships AS m ON (a.auth_id = m.event_id)
+ WHERE a.event_id = ? AND m.user_id = ?
+ """,
+ (event_id, user_id),
+ )
+ row = txn.fetchone()
+
+ # We should see a corresponding previous invite/knock event
+ assert row is not None
+ previous_event_id, membership = row
+
+ return previous_event_id, membership
+
+ # Map from (room_id, user_id) to ...
+ to_insert_membership_snapshots: Dict[
+ Tuple[str, str], SlidingSyncMembershipSnapshotSharedInsertValues
+ ] = {}
+ to_insert_membership_infos: Dict[Tuple[str, str], SlidingSyncMembershipInfo] = (
+ {}
+ )
+ for (
+ room_id,
+ room_id_from_rooms_table,
+ user_id,
+ sender,
+ membership_event_id,
+ membership,
+ membership_event_stream_ordering,
+ membership_event_instance_name,
+ is_outlier,
+ ) in memberships_to_update_rows:
+ # We don't know how to handle `membership` values other than these. The
+ # code below would need to be updated.
+ assert membership in (
+ Membership.JOIN,
+ Membership.INVITE,
+ Membership.KNOCK,
+ Membership.LEAVE,
+ Membership.BAN,
+ )
+
+ # There are some old out-of-band memberships (before
+ # https://github.com/matrix-org/synapse/issues/6983) where we don't have the
+ # corresponding room stored in the `rooms` table`. We have a `FOREIGN KEY`
+ # constraint on the `sliding_sync_membership_snapshots` table so we have to
+ # fix-up these memberships by adding the room to the `rooms` table.
+ if room_id_from_rooms_table is None:
+ await self.db_pool.simple_insert(
+ table="rooms",
+ values={
+ "room_id": room_id,
+ # Only out-of-band memberships are missing from the `rooms`
+ # table so that is the only type of membership we're dealing
+ # with here. Since we don't calculate the "chain cover" for
+ # out-of-band memberships, we can just set this to `True` as if
+ # the user ever joins the room, we will end up calculating the
+ # "chain cover" anyway.
+ "has_auth_chain_index": True,
+ },
+ )
+
+ # Map of values to insert/update in the `sliding_sync_membership_snapshots` table
+ sliding_sync_membership_snapshots_insert_map: (
+ SlidingSyncMembershipSnapshotSharedInsertValues
+ ) = {}
+ if membership == Membership.JOIN:
+ # If we're still joined, we can pull from current state.
+ current_state_ids_map: StateMap[
+ str
+ ] = await self.hs.get_storage_controllers().state.get_current_state_ids(
+ room_id,
+ state_filter=StateFilter.from_types(
+ SLIDING_SYNC_RELEVANT_STATE_SET
+ ),
+ # Partially-stated rooms should have all state events except for
+ # remote membership events so we don't need to wait at all because
+ # we only want some non-membership state
+ await_full_state=False,
+ )
+ # We're iterating over rooms that we are joined to so they should
+ # have `current_state_events` and we should have some current state
+ # for each room
+ assert current_state_ids_map
+
+ fetched_events = await self.get_events(current_state_ids_map.values())
+
+ current_state_map: StateMap[EventBase] = {
+ state_key: fetched_events[event_id]
+ for state_key, event_id in current_state_ids_map.items()
+ # `get_events(...)` will filter out events for unknown room versions
+ if event_id in fetched_events
+ }
+
+ # Can happen for unknown room versions (old room versions that aren't known
+ # anymore) since `get_events(...)` will filter out events for unknown room
+ # versions
+ if not current_state_map:
+ continue
+
+ state_insert_values = (
+ PersistEventsStore._get_sliding_sync_insert_values_from_state_map(
+ current_state_map
+ )
+ )
+ sliding_sync_membership_snapshots_insert_map.update(state_insert_values)
+ # We should have some insert values for each room, even if they are `None`
+ assert sliding_sync_membership_snapshots_insert_map
+
+ # We have current state to work from
+ sliding_sync_membership_snapshots_insert_map["has_known_state"] = True
+ elif membership in (Membership.INVITE, Membership.KNOCK) or (
+ membership == Membership.LEAVE and is_outlier
+ ):
+ invite_or_knock_event_id = membership_event_id
+ invite_or_knock_membership = membership
+
+ # If the event is an `out_of_band_membership` (special case of
+ # `outlier`), we never had historical state so we have to pull from
+ # the stripped state on the previous invite/knock event. This gives
+ # us a consistent view of the room state regardless of your
+ # membership (i.e. the room shouldn't disappear if your using the
+ # `is_encrypted` filter and you leave).
+ if membership == Membership.LEAVE and is_outlier:
+ invite_or_knock_event_id, invite_or_knock_membership = (
+ await self.db_pool.runInteraction(
+ "sliding_sync_membership_snapshots_bg_update._find_previous_membership",
+ _find_previous_membership_txn,
+ membership_event_id,
+ user_id,
+ )
+ )
+
+ # Pull from the stripped state on the invite/knock event
+ invite_or_knock_event = await self.get_event(invite_or_knock_event_id)
+
+ raw_stripped_state_events = None
+ if invite_or_knock_membership == Membership.INVITE:
+ invite_room_state = invite_or_knock_event.unsigned.get(
+ "invite_room_state"
+ )
+ raw_stripped_state_events = invite_room_state
+ elif invite_or_knock_membership == Membership.KNOCK:
+ knock_room_state = invite_or_knock_event.unsigned.get(
+ "knock_room_state"
+ )
+ raw_stripped_state_events = knock_room_state
+
+ sliding_sync_membership_snapshots_insert_map = PersistEventsStore._get_sliding_sync_insert_values_from_stripped_state(
+ raw_stripped_state_events
+ )
+
+ # We should have some insert values for each room, even if no
+ # stripped state is on the event because we still want to record
+ # that we have no known state
+ assert sliding_sync_membership_snapshots_insert_map
+ elif membership in (Membership.LEAVE, Membership.BAN):
+ # Pull from historical state
+ state_ids_map = await self.hs.get_storage_controllers().state.get_state_ids_for_event(
+ membership_event_id,
+ state_filter=StateFilter.from_types(
+ SLIDING_SYNC_RELEVANT_STATE_SET
+ ),
+ # Partially-stated rooms should have all state events except for
+ # remote membership events so we don't need to wait at all because
+ # we only want some non-membership state
+ await_full_state=False,
+ )
+
+ fetched_events = await self.get_events(state_ids_map.values())
+
+ state_map: StateMap[EventBase] = {
+ state_key: fetched_events[event_id]
+ for state_key, event_id in state_ids_map.items()
+ # `get_events(...)` will filter out events for unknown room versions
+ if event_id in fetched_events
+ }
+
+ # Can happen for unknown room versions (old room versions that aren't known
+ # anymore) since `get_events(...)` will filter out events for unknown room
+ # versions
+ if not state_map:
+ continue
+
+ state_insert_values = (
+ PersistEventsStore._get_sliding_sync_insert_values_from_state_map(
+ state_map
+ )
+ )
+ sliding_sync_membership_snapshots_insert_map.update(state_insert_values)
+ # We should have some insert values for each room, even if they are `None`
+ assert sliding_sync_membership_snapshots_insert_map
+
+ # We have historical state to work from
+ sliding_sync_membership_snapshots_insert_map["has_known_state"] = True
+ else:
+ # We don't know how to handle this type of membership yet
+ #
+ # FIXME: We should use `assert_never` here but for some reason
+ # the exhaustive matching doesn't recognize the `Never` here.
+ # assert_never(membership)
+ raise AssertionError(
+ f"Unexpected membership {membership} ({membership_event_id}) that we don't know how to handle yet"
+ )
+
+ to_insert_membership_snapshots[(room_id, user_id)] = (
+ sliding_sync_membership_snapshots_insert_map
+ )
+ to_insert_membership_infos[(room_id, user_id)] = SlidingSyncMembershipInfo(
+ user_id=user_id,
+ sender=sender,
+ membership_event_id=membership_event_id,
+ membership=membership,
+ membership_event_stream_ordering=membership_event_stream_ordering,
+ # If instance_name is null we default to "master"
+ membership_event_instance_name=membership_event_instance_name
+ or "master",
+ )
+
+ def _fill_table_txn(txn: LoggingTransaction) -> None:
+ # Handle updating the `sliding_sync_membership_snapshots` table
+ #
+ for key, insert_map in to_insert_membership_snapshots.items():
+ room_id, user_id = key
+ membership_info = to_insert_membership_infos[key]
+ sender = membership_info.sender
+ membership_event_id = membership_info.membership_event_id
+ membership = membership_info.membership
+ membership_event_stream_ordering = (
+ membership_info.membership_event_stream_ordering
+ )
+ membership_event_instance_name = (
+ membership_info.membership_event_instance_name
+ )
+
+ # We don't need to upsert the state because we never partially
+ # insert/update the snapshots and anything already there is up-to-date
+ # EXCEPT for the `forgotten` field since that is updated out-of-band
+ # from the membership changes.
+ #
+ # Even though we're only doing insertions, we're using
+ # `simple_upsert_txn()` here to avoid unique violation errors that would
+ # happen from `simple_insert_txn()`
+ self.db_pool.simple_upsert_txn(
+ txn,
+ table="sliding_sync_membership_snapshots",
+ keyvalues={"room_id": room_id, "user_id": user_id},
+ values={},
+ insertion_values={
+ **insert_map,
+ "sender": sender,
+ "membership_event_id": membership_event_id,
+ "membership": membership,
+ "event_stream_ordering": membership_event_stream_ordering,
+ "event_instance_name": membership_event_instance_name,
+ },
+ )
+ # We need to find the `forgotten` value during the transaction because
+ # we can't risk inserting stale data.
+ txn.execute(
+ """
+ UPDATE sliding_sync_membership_snapshots
+ SET
+ forgotten = (SELECT forgotten FROM room_memberships WHERE event_id = ?)
+ WHERE room_id = ? and user_id = ?
+ """,
+ (
+ membership_event_id,
+ room_id,
+ user_id,
+ ),
+ )
+
+ await self.db_pool.runInteraction(
+ "sliding_sync_membership_snapshots_bg_update", _fill_table_txn
+ )
+
+ # Update the progress
+ (
+ room_id,
+ _room_id_from_rooms_table,
+ _user_id,
+ _sender,
+ _membership_event_id,
+ _membership,
+ membership_event_stream_ordering,
+ _membership_event_instance_name,
+ _is_outlier,
+ ) = memberships_to_update_rows[-1]
+
+ progress = {
+ "initial_phase": initial_phase,
+ "last_room_id": room_id,
+ "last_event_stream_ordering": membership_event_stream_ordering,
+ }
+
+ await self.db_pool.updates._background_update_progress(
+ _BackgroundUpdates.SLIDING_SYNC_MEMBERSHIP_SNAPSHOTS_BG_UPDATE,
+ progress,
+ )
+
+ return len(memberships_to_update_rows)
+
+
+def _resolve_stale_data_in_sliding_sync_tables(
+ txn: LoggingTransaction,
+) -> None:
+ """
+ Clears stale/out-of-date entries from the
+ `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` tables.
+
+ This accounts for when someone downgrades their Synapse version and then upgrades it
+ again. This will ensure that we don't have any stale/out-of-date data in the
+ `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` tables since any new
+ events sent in rooms would have also needed to be written to the sliding sync
+ tables. For example a new event needs to bump `event_stream_ordering` in
+ `sliding_sync_joined_rooms` table or some state in the room changing (like the room
+ name). Or another example of someone's membership changing in a room affecting
+ `sliding_sync_membership_snapshots`.
+
+ This way, if a row exists in the sliding sync tables, we are able to rely on it
+ (accurate data). And if a row doesn't exist, we use a fallback to get the same info
+ until the background updates fill in the rows or a new event comes in triggering it
+ to be fully inserted.
+
+ FIXME: This can be removed once we bump `SCHEMA_COMPAT_VERSION` and run the
+ foreground update for
+ `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` (tracked by
+ https://github.com/element-hq/synapse/issues/17623)
+ """
+
+ _resolve_stale_data_in_sliding_sync_joined_rooms_table(txn)
+ _resolve_stale_data_in_sliding_sync_membership_snapshots_table(txn)
+
+
+def _resolve_stale_data_in_sliding_sync_joined_rooms_table(
+ txn: LoggingTransaction,
+) -> None:
+ """
+ Clears stale/out-of-date entries from the `sliding_sync_joined_rooms` table and
+ kicks-off the background update to catch-up with what we missed while Synapse was
+ downgraded.
+
+ See `_resolve_stale_data_in_sliding_sync_tables()` description above for more
+ context.
+ """
+
+ # Find the point when we stopped writing to the `sliding_sync_joined_rooms` table
+ txn.execute(
+ """
+ SELECT event_stream_ordering
+ FROM sliding_sync_joined_rooms
+ ORDER BY event_stream_ordering DESC
+ LIMIT 1
+ """,
+ )
+
+ # If we have nothing written to the `sliding_sync_joined_rooms` table, there is
+ # nothing to clean up
+ row = cast(Optional[Tuple[int]], txn.fetchone())
+ max_stream_ordering_sliding_sync_joined_rooms_table = None
+ depends_on = None
+ if row is not None:
+ (max_stream_ordering_sliding_sync_joined_rooms_table,) = row
+
+ txn.execute(
+ """
+ SELECT room_id
+ FROM events
+ WHERE stream_ordering > ?
+ GROUP BY room_id
+ ORDER BY MAX(stream_ordering) ASC
+ """,
+ (max_stream_ordering_sliding_sync_joined_rooms_table,),
+ )
+
+ room_rows = txn.fetchall()
+ # No new events have been written to the `events` table since the last time we wrote
+ # to the `sliding_sync_joined_rooms` table so there is nothing to clean up. This is
+ # the expected normal scenario for people who have not downgraded their Synapse
+ # version.
+ if not room_rows:
+ return
+
+ # 1000 is an arbitrary batch size with no testing
+ for chunk in batch_iter(room_rows, 1000):
+ # Handle updating the `sliding_sync_joined_rooms` table
+ #
+ # Clear out the stale data
+ DatabasePool.simple_delete_many_batch_txn(
+ txn,
+ table="sliding_sync_joined_rooms",
+ keys=("room_id",),
+ values=chunk,
+ )
+
+ # Update the `sliding_sync_joined_rooms_to_recalculate` table with the rooms
+ # that went stale and now need to be recalculated.
+ DatabasePool.simple_upsert_many_txn_native_upsert(
+ txn,
+ table="sliding_sync_joined_rooms_to_recalculate",
+ key_names=("room_id",),
+ key_values=chunk,
+ value_names=(),
+ # No value columns, therefore make a blank list so that the following
+ # zip() works correctly.
+ value_values=[() for x in range(len(chunk))],
+ )
+ else:
+ # Avoid adding the background updates when there is no data to run them on (if
+ # the homeserver has no rooms). The portdb script refuses to run with pending
+ # background updates and since we potentially add them every time the server
+ # starts, we add this check for to allow the script to breath.
+ txn.execute("SELECT 1 FROM local_current_membership LIMIT 1")
+ row = txn.fetchone()
+ if row is None:
+ # There are no rooms, so don't schedule the bg update.
+ return
+
+ # Re-run the `sliding_sync_joined_rooms_to_recalculate` prefill if there is
+ # nothing in the `sliding_sync_joined_rooms` table
+ DatabasePool.simple_upsert_txn_native_upsert(
+ txn,
+ table="background_updates",
+ keyvalues={
+ "update_name": _BackgroundUpdates.SLIDING_SYNC_PREFILL_JOINED_ROOMS_TO_RECALCULATE_TABLE_BG_UPDATE
+ },
+ values={},
+ # Only insert the row if it doesn't already exist. If it already exists,
+ # we're already working on it
+ insertion_values={
+ "progress_json": "{}",
+ },
+ )
+ depends_on = (
+ _BackgroundUpdates.SLIDING_SYNC_PREFILL_JOINED_ROOMS_TO_RECALCULATE_TABLE_BG_UPDATE
+ )
+
+ # Now kick-off the background update to catch-up with what we missed while Synapse
+ # was downgraded.
+ #
+ # We may need to catch-up on everything if we have nothing written to the
+ # `sliding_sync_joined_rooms` table yet. This could happen if someone had zero rooms
+ # on their server (so the normal background update completes), downgrade Synapse
+ # versions, join and create some new rooms, and upgrade again.
+ DatabasePool.simple_upsert_txn_native_upsert(
+ txn,
+ table="background_updates",
+ keyvalues={
+ "update_name": _BackgroundUpdates.SLIDING_SYNC_JOINED_ROOMS_BG_UPDATE
+ },
+ values={},
+ # Only insert the row if it doesn't already exist. If it already exists, we will
+ # eventually fill in the rows we're trying to populate.
+ insertion_values={
+ # Empty progress is expected since it's not used for this background update.
+ "progress_json": "{}",
+ # Wait for the prefill to finish
+ "depends_on": depends_on,
+ },
+ )
+
+
+def _resolve_stale_data_in_sliding_sync_membership_snapshots_table(
+ txn: LoggingTransaction,
+) -> None:
+ """
+ Clears stale/out-of-date entries from the `sliding_sync_membership_snapshots` table
+ and kicks-off the background update to catch-up with what we missed while Synapse
+ was downgraded.
+
+ See `_resolve_stale_data_in_sliding_sync_tables()` description above for more
+ context.
+ """
+
+ # Find the point when we stopped writing to the `sliding_sync_membership_snapshots` table
+ txn.execute(
+ """
+ SELECT event_stream_ordering
+ FROM sliding_sync_membership_snapshots
+ ORDER BY event_stream_ordering DESC
+ LIMIT 1
+ """,
+ )
+
+ # If we have nothing written to the `sliding_sync_membership_snapshots` table,
+ # there is nothing to clean up
+ row = cast(Optional[Tuple[int]], txn.fetchone())
+ max_stream_ordering_sliding_sync_membership_snapshots_table = None
+ if row is not None:
+ (max_stream_ordering_sliding_sync_membership_snapshots_table,) = row
+
+ # XXX: Since `forgotten` is simply a flag on the `room_memberships` table that is
+ # set out-of-band, there is no way to tell whether it was set while Synapse was
+ # downgraded. The only thing the user can do is `/forget` again if they run into
+ # this.
+ #
+ # This only picks up changes to memberships.
+ txn.execute(
+ """
+ SELECT user_id, room_id
+ FROM local_current_membership
+ WHERE event_stream_ordering > ?
+ ORDER BY event_stream_ordering ASC
+ """,
+ (max_stream_ordering_sliding_sync_membership_snapshots_table,),
+ )
+
+ membership_rows = txn.fetchall()
+ # No new events have been written to the `events` table since the last time we wrote
+ # to the `sliding_sync_membership_snapshots` table so there is nothing to clean up.
+ # This is the expected normal scenario for people who have not downgraded their
+ # Synapse version.
+ if not membership_rows:
+ return
+
+ # 1000 is an arbitrary batch size with no testing
+ for chunk in batch_iter(membership_rows, 1000):
+ # Handle updating the `sliding_sync_membership_snapshots` table
+ #
+ DatabasePool.simple_delete_many_batch_txn(
+ txn,
+ table="sliding_sync_membership_snapshots",
+ keys=("user_id", "room_id"),
+ values=chunk,
+ )
+ else:
+ # Avoid adding the background updates when there is no data to run them on (if
+ # the homeserver has no rooms). The portdb script refuses to run with pending
+ # background updates and since we potentially add them every time the server
+ # starts, we add this check for to allow the script to breath.
+ txn.execute("SELECT 1 FROM local_current_membership LIMIT 1")
+ row = txn.fetchone()
+ if row is None:
+ # There are no rooms, so don't schedule the bg update.
+ return
+
+ # Now kick-off the background update to catch-up with what we missed while Synapse
+ # was downgraded.
+ #
+ # We may need to catch-up on everything if we have nothing written to the
+ # `sliding_sync_membership_snapshots` table yet. This could happen if someone had
+ # zero rooms on their server (so the normal background update completes), downgrade
+ # Synapse versions, join and create some new rooms, and upgrade again.
+ #
+ progress_json: JsonDict = {}
+ if max_stream_ordering_sliding_sync_membership_snapshots_table is not None:
+ progress_json["initial_phase"] = False
+ progress_json["last_event_stream_ordering"] = (
+ max_stream_ordering_sliding_sync_membership_snapshots_table
+ )
+
+ DatabasePool.simple_upsert_txn_native_upsert(
+ txn,
+ table="background_updates",
+ keyvalues={
+ "update_name": _BackgroundUpdates.SLIDING_SYNC_MEMBERSHIP_SNAPSHOTS_BG_UPDATE
+ },
+ values={},
+ # Only insert the row if it doesn't already exist. If it already exists, we will
+ # eventually fill in the rows we're trying to populate.
+ insertion_values={
+ "progress_json": json_encoder.encode(progress_json),
+ },
+ )
diff --git a/synapse/storage/databases/main/events_worker.py b/synapse/storage/databases/main/events_worker.py
index 4d4877c4c3..6079cc4a52 100644
--- a/synapse/storage/databases/main/events_worker.py
+++ b/synapse/storage/databases/main/events_worker.py
@@ -457,6 +457,8 @@ class EventsWorkerStore(SQLBaseStore):
) -> Optional[EventBase]:
"""Get an event from the database by event_id.
+ Events for unknown room versions will also be filtered out.
+
Args:
event_id: The event_id of the event to fetch
@@ -511,6 +513,10 @@ class EventsWorkerStore(SQLBaseStore):
) -> Dict[str, EventBase]:
"""Get events from the database
+ Unknown events will be omitted from the response.
+
+ Events for unknown room versions will also be filtered out.
+
Args:
event_ids: The event_ids of the events to fetch
@@ -553,6 +559,8 @@ class EventsWorkerStore(SQLBaseStore):
Unknown events will be omitted from the response.
+ Events for unknown room versions will also be filtered out.
+
Args:
event_ids: The event_ids of the events to fetch
diff --git a/synapse/storage/databases/main/purge_events.py b/synapse/storage/databases/main/purge_events.py
index 3b81ed943c..fc4c286595 100644
--- a/synapse/storage/databases/main/purge_events.py
+++ b/synapse/storage/databases/main/purge_events.py
@@ -454,6 +454,10 @@ class PurgeEventsStore(StateGroupWorkerStore, CacheInvalidationWorkerStore):
# so must be deleted first.
"local_current_membership",
"room_memberships",
+ # Note: the sliding_sync_ tables have foreign keys to the `events` table
+ # so must be deleted first.
+ "sliding_sync_joined_rooms",
+ "sliding_sync_membership_snapshots",
"events",
"federation_inbound_events_staging",
"receipts_graph",
diff --git a/synapse/storage/databases/main/roommember.py b/synapse/storage/databases/main/roommember.py
index 71baf57663..722686d4b8 100644
--- a/synapse/storage/databases/main/roommember.py
+++ b/synapse/storage/databases/main/roommember.py
@@ -1337,6 +1337,12 @@ class RoomMemberWorkerStore(EventsWorkerStore, CacheInvalidationWorkerStore):
keyvalues={"user_id": user_id, "room_id": room_id},
updatevalues={"forgotten": 1},
)
+ self.db_pool.simple_update_txn(
+ txn,
+ table="sliding_sync_membership_snapshots",
+ keyvalues={"user_id": user_id, "room_id": room_id},
+ updatevalues={"forgotten": 1},
+ )
self._invalidate_cache_and_stream(txn, self.did_forget, (user_id, room_id))
self._invalidate_cache_and_stream(
diff --git a/synapse/storage/databases/main/state_deltas.py b/synapse/storage/databases/main/state_deltas.py
index eaa13da368..ba52fff652 100644
--- a/synapse/storage/databases/main/state_deltas.py
+++ b/synapse/storage/databases/main/state_deltas.py
@@ -161,45 +161,80 @@ class StateDeltasStore(SQLBaseStore):
self._get_max_stream_id_in_current_state_deltas_txn,
)
- @trace
- async def get_current_state_deltas_for_room(
- self, room_id: str, from_token: RoomStreamToken, to_token: RoomStreamToken
+ def get_current_state_deltas_for_room_txn(
+ self,
+ txn: LoggingTransaction,
+ room_id: str,
+ *,
+ from_token: Optional[RoomStreamToken],
+ to_token: Optional[RoomStreamToken],
) -> List[StateDelta]:
- """Get the state deltas between two tokens."""
-
- if not self._curr_state_delta_stream_cache.has_entity_changed(
- room_id, from_token.stream
- ):
- return []
+ """
+ Get the state deltas between two tokens.
- def get_current_state_deltas_for_room_txn(
- txn: LoggingTransaction,
- ) -> List[StateDelta]:
- sql = """
+ (> `from_token` and <= `to_token`)
+ """
+ from_clause = ""
+ from_args = []
+ if from_token is not None:
+ from_clause = "AND ? < stream_id"
+ from_args = [from_token.stream]
+
+ to_clause = ""
+ to_args = []
+ if to_token is not None:
+ to_clause = "AND stream_id <= ?"
+ to_args = [to_token.get_max_stream_pos()]
+
+ sql = f"""
SELECT instance_name, stream_id, type, state_key, event_id, prev_event_id
FROM current_state_delta_stream
- WHERE room_id = ? AND ? < stream_id AND stream_id <= ?
+ WHERE room_id = ? {from_clause} {to_clause}
ORDER BY stream_id ASC
"""
- txn.execute(
- sql, (room_id, from_token.stream, to_token.get_max_stream_pos())
+ txn.execute(sql, [room_id] + from_args + to_args)
+
+ return [
+ StateDelta(
+ stream_id=row[1],
+ room_id=room_id,
+ event_type=row[2],
+ state_key=row[3],
+ event_id=row[4],
+ prev_event_id=row[5],
)
+ for row in txn
+ if _filter_results_by_stream(from_token, to_token, row[0], row[1])
+ ]
- return [
- StateDelta(
- stream_id=row[1],
- room_id=room_id,
- event_type=row[2],
- state_key=row[3],
- event_id=row[4],
- prev_event_id=row[5],
- )
- for row in txn
- if _filter_results_by_stream(from_token, to_token, row[0], row[1])
- ]
+ @trace
+ async def get_current_state_deltas_for_room(
+ self,
+ room_id: str,
+ *,
+ from_token: Optional[RoomStreamToken],
+ to_token: Optional[RoomStreamToken],
+ ) -> List[StateDelta]:
+ """
+ Get the state deltas between two tokens.
+
+ (> `from_token` and <= `to_token`)
+ """
+
+ if (
+ from_token is not None
+ and not self._curr_state_delta_stream_cache.has_entity_changed(
+ room_id, from_token.stream
+ )
+ ):
+ return []
return await self.db_pool.runInteraction(
- "get_current_state_deltas_for_room", get_current_state_deltas_for_room_txn
+ "get_current_state_deltas_for_room",
+ self.get_current_state_deltas_for_room_txn,
+ room_id,
+ from_token=from_token,
+ to_token=to_token,
)
@trace
diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py
index e33a8cfe97..1a59e0b5a8 100644
--- a/synapse/storage/databases/main/stream.py
+++ b/synapse/storage/databases/main/stream.py
@@ -1264,12 +1264,76 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
return None
+ async def get_last_event_pos_in_room(
+ self,
+ room_id: str,
+ event_types: Optional[StrCollection] = None,
+ ) -> Optional[Tuple[str, PersistedEventPosition]]:
+ """
+ Returns the ID and event position of the last event in a room.
+
+ Based on `get_last_event_pos_in_room_before_stream_ordering(...)`
+
+ Args:
+ room_id
+ event_types: Optional allowlist of event types to filter by
+
+ Returns:
+ The ID of the most recent event and it's position, or None if there are no
+ events in the room that match the given event types.
+ """
+
+ def _get_last_event_pos_in_room_txn(
+ txn: LoggingTransaction,
+ ) -> Optional[Tuple[str, PersistedEventPosition]]:
+ event_type_clause = ""
+ event_type_args: List[str] = []
+ if event_types is not None and len(event_types) > 0:
+ event_type_clause, event_type_args = make_in_list_sql_clause(
+ txn.database_engine, "type", event_types
+ )
+ event_type_clause = f"AND {event_type_clause}"
+
+ sql = f"""
+ SELECT event_id, stream_ordering, instance_name
+ FROM events
+ LEFT JOIN rejections USING (event_id)
+ WHERE room_id = ?
+ {event_type_clause}
+ AND NOT outlier
+ AND rejections.event_id IS NULL
+ ORDER BY stream_ordering DESC
+ LIMIT 1
+ """
+
+ txn.execute(
+ sql,
+ [room_id] + event_type_args,
+ )
+
+ row = cast(Optional[Tuple[str, int, str]], txn.fetchone())
+ if row is not None:
+ event_id, stream_ordering, instance_name = row
+
+ return event_id, PersistedEventPosition(
+ # If instance_name is null we default to "master"
+ instance_name or "master",
+ stream_ordering,
+ )
+
+ return None
+
+ return await self.db_pool.runInteraction(
+ "get_last_event_pos_in_room",
+ _get_last_event_pos_in_room_txn,
+ )
+
@trace
async def get_last_event_pos_in_room_before_stream_ordering(
self,
room_id: str,
end_token: RoomStreamToken,
- event_types: Optional[Collection[str]] = None,
+ event_types: Optional[StrCollection] = None,
) -> Optional[Tuple[str, PersistedEventPosition]]:
"""
Returns the ID and event position of the last event in a room at or before a
diff --git a/synapse/storage/schema/__init__.py b/synapse/storage/schema/__init__.py
index 581d00346b..316541d818 100644
--- a/synapse/storage/schema/__init__.py
+++ b/synapse/storage/schema/__init__.py
@@ -19,7 +19,7 @@
#
#
-SCHEMA_VERSION = 86 # remember to update the list below when updating
+SCHEMA_VERSION = 87 # remember to update the list below when updating
"""Represents the expectations made by the codebase about the database schema
This should be incremented whenever the codebase changes its requirements on the
@@ -142,6 +142,10 @@ Changes in SCHEMA_VERSION = 85
Changes in SCHEMA_VERSION = 86
- Add a column `authenticated` to the tables `local_media_repository` and `remote_media_cache`
+
+Changes in SCHEMA_VERSION = 87
+ - Add tables to store Sliding Sync data for quick filtering/sorting
+ (`sliding_sync_joined_rooms`, `sliding_sync_membership_snapshots`)
"""
diff --git a/synapse/storage/schema/main/delta/87/01_sliding_sync_memberships.sql b/synapse/storage/schema/main/delta/87/01_sliding_sync_memberships.sql
new file mode 100644
index 0000000000..2f71e541f8
--- /dev/null
+++ b/synapse/storage/schema/main/delta/87/01_sliding_sync_memberships.sql
@@ -0,0 +1,169 @@
+--
+-- This file is licensed under the Affero General Public License (AGPL) version 3.
+--
+-- Copyright (C) 2024 New Vector, Ltd
+--
+-- This program is free software: you can redistribute it and/or modify
+-- it under the terms of the GNU Affero General Public License as
+-- published by the Free Software Foundation, either version 3 of the
+-- License, or (at your option) any later version.
+--
+-- See the GNU Affero General Public License for more details:
+-- <https://www.gnu.org/licenses/agpl-3.0.html>.
+
+-- This table is a list/queue used to keep track of which rooms need to be inserted into
+-- `sliding_sync_joined_rooms`. We do this to avoid reading from `current_state_events`
+-- during the background update to populate `sliding_sync_joined_rooms` which works but
+-- it takes a lot of work for the database to grab `DISTINCT` room_ids given how many
+-- state events there are for each room.
+--
+-- This table is prefilled with every room in the `rooms` table (see the
+-- `sliding_sync_prefill_joined_rooms_to_recalculate_table_bg_update` background
+-- update). This table is also updated whenever we come across stale data so that we can
+-- catch-up with all of the new data if Synapse was downgraded (see
+-- `_resolve_stale_data_in_sliding_sync_tables`).
+--
+-- FIXME: This can be removed once we bump `SCHEMA_COMPAT_VERSION` and run the
+-- foreground update for
+-- `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` (tracked by
+-- https://github.com/element-hq/synapse/issues/17623)
+CREATE TABLE IF NOT EXISTS sliding_sync_joined_rooms_to_recalculate(
+ room_id TEXT NOT NULL REFERENCES rooms(room_id),
+ PRIMARY KEY (room_id)
+);
+
+-- A table for storing room meta data (current state relevant to sliding sync) that the
+-- local server is still participating in (someone local is joined to the room).
+--
+-- We store the joined rooms in separate table from `sliding_sync_membership_snapshots`
+-- because we need up-to-date information for joined rooms and it can be shared across
+-- everyone who is joined.
+--
+-- This table is kept in sync with `current_state_events` which means if the server is
+-- no longer participating in a room, the row will be deleted.
+CREATE TABLE IF NOT EXISTS sliding_sync_joined_rooms(
+ room_id TEXT NOT NULL REFERENCES rooms(room_id),
+ -- The `stream_ordering` of the most-recent/latest event in the room
+ event_stream_ordering BIGINT NOT NULL REFERENCES events(stream_ordering),
+ -- The `stream_ordering` of the last event according to the `bump_event_types`
+ bump_stamp BIGINT,
+ -- `m.room.create` -> `content.type` (current state)
+ --
+ -- Useful for the `spaces`/`not_spaces` filter in the Sliding Sync API
+ room_type TEXT,
+ -- `m.room.name` -> `content.name` (current state)
+ --
+ -- Useful for the room meta data and `room_name_like` filter in the Sliding Sync API
+ room_name TEXT,
+ -- `m.room.encryption` -> `content.algorithm` (current state)
+ --
+ -- Useful for the `is_encrypted` filter in the Sliding Sync API
+ is_encrypted BOOLEAN DEFAULT FALSE NOT NULL,
+ -- `m.room.tombstone` -> `content.replacement_room` (according to the current state at the
+ -- time of the membership).
+ --
+ -- Useful for the `include_old_rooms` functionality in the Sliding Sync API
+ tombstone_successor_room_id TEXT,
+ PRIMARY KEY (room_id)
+);
+
+-- So we can purge rooms easily.
+--
+-- The primary key is already `room_id`
+
+-- So we can sort by `stream_ordering
+CREATE UNIQUE INDEX IF NOT EXISTS sliding_sync_joined_rooms_event_stream_ordering ON sliding_sync_joined_rooms(event_stream_ordering);
+
+-- A table for storing a snapshot of room meta data (historical current state relevant
+-- for sliding sync) at the time of a local user's membership. Only has rows for the
+-- latest membership event for a given local user in a room which matches
+-- `local_current_membership` .
+--
+-- We store all memberships including joins. This makes it easy to reference this table
+-- to find all membership for a given user and shares the same semantics as
+-- `local_current_membership`. And we get to avoid some table maintenance; if we only
+-- stored non-joins, we would have to delete the row for the user when the user joins
+-- the room. Stripped state doesn't include the `m.room.tombstone` event, so we just
+-- assume that the room doesn't have a tombstone.
+--
+-- For remote invite/knocks where the server is not participating in the room, we will
+-- use stripped state events to populate this table. We assume that if any stripped
+-- state is given, it will include all possible stripped state events types. For
+-- example, if stripped state is given but `m.room.encryption` isn't included, we will
+-- assume that the room is not encrypted.
+--
+-- We don't include `bump_stamp` here because we can just use the `stream_ordering` from
+-- the membership event itself as the `bump_stamp`.
+CREATE TABLE IF NOT EXISTS sliding_sync_membership_snapshots(
+ room_id TEXT NOT NULL REFERENCES rooms(room_id),
+ user_id TEXT NOT NULL,
+ -- Useful to be able to tell leaves from kicks (where the `user_id` is different from the `sender`)
+ sender TEXT NOT NULL,
+ membership_event_id TEXT NOT NULL REFERENCES events(event_id),
+ membership TEXT NOT NULL,
+ -- This is an integer just to match `room_memberships` and also means we don't need
+ -- to do any casting.
+ forgotten INTEGER DEFAULT 0 NOT NULL,
+ -- `stream_ordering` of the `membership_event_id`
+ event_stream_ordering BIGINT NOT NULL REFERENCES events(stream_ordering),
+ -- `instance_name` of the worker that persisted the `membership_event_id`.
+ -- Useful for crafting `PersistedEventPosition(...)`
+ event_instance_name TEXT NOT NULL,
+ -- For remote invites/knocks that don't include any stripped state, we want to be
+ -- able to distinguish between a room with `None` as valid value for some state and
+ -- room where the state is completely unknown. Basically, this should be True unless
+ -- no stripped state was provided for a remote invite/knock (False).
+ has_known_state BOOLEAN DEFAULT FALSE NOT NULL,
+ -- `m.room.create` -> `content.type` (according to the current state at the time of
+ -- the membership).
+ --
+ -- Useful for the `spaces`/`not_spaces` filter in the Sliding Sync API
+ room_type TEXT,
+ -- `m.room.name` -> `content.name` (according to the current state at the time of
+ -- the membership).
+ --
+ -- Useful for the room meta data and `room_name_like` filter in the Sliding Sync API
+ room_name TEXT,
+ -- `m.room.encryption` -> `content.algorithm` (according to the current state at the
+ -- time of the membership).
+ --
+ -- Useful for the `is_encrypted` filter in the Sliding Sync API
+ is_encrypted BOOLEAN DEFAULT FALSE NOT NULL,
+ -- `m.room.tombstone` -> `content.replacement_room` (according to the current state at the
+ -- time of the membership).
+ --
+ -- Useful for the `include_old_rooms` functionality in the Sliding Sync API
+ tombstone_successor_room_id TEXT,
+ PRIMARY KEY (room_id, user_id)
+);
+
+-- So we can purge rooms easily.
+--
+-- Since we're using a multi-column index as the primary key (room_id, user_id), the
+-- first index column (room_id) is always usable for searching so we don't need to
+-- create a separate index for it.
+--
+-- CREATE INDEX IF NOT EXISTS sliding_sync_membership_snapshots_room_id ON sliding_sync_membership_snapshots(room_id);
+
+-- So we can fetch all rooms for a given user
+CREATE INDEX IF NOT EXISTS sliding_sync_membership_snapshots_user_id ON sliding_sync_membership_snapshots(user_id);
+-- So we can sort by `stream_ordering
+CREATE UNIQUE INDEX IF NOT EXISTS sliding_sync_membership_snapshots_event_stream_ordering ON sliding_sync_membership_snapshots(event_stream_ordering);
+
+
+-- Add a series of background updates to populate the new `sliding_sync_joined_rooms` table:
+--
+-- 1. Add a background update to prefill `sliding_sync_joined_rooms_to_recalculate`.
+-- We do a one-shot bulk insert from the `rooms` table to prefill.
+-- 2. Add a background update to populate the new `sliding_sync_joined_rooms` table
+-- based on the rooms listed in the `sliding_sync_joined_rooms_to_recalculate`
+-- table.
+--
+INSERT INTO background_updates (ordering, update_name, progress_json) VALUES
+ (8701, 'sliding_sync_prefill_joined_rooms_to_recalculate_table_bg_update', '{}');
+INSERT INTO background_updates (ordering, update_name, progress_json, depends_on) VALUES
+ (8701, 'sliding_sync_joined_rooms_bg_update', '{}', 'sliding_sync_prefill_joined_rooms_to_recalculate_table_bg_update');
+
+-- Add a background updates to populate the new `sliding_sync_membership_snapshots` table
+INSERT INTO background_updates (ordering, update_name, progress_json) VALUES
+ (8701, 'sliding_sync_membership_snapshots_bg_update', '{}');
|
