dandri/synapse
1
0
Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2026-07-29 16:21:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

Get full events for _sliding_sync_joined_rooms_backfill

Browse Source
This commit is contained in:
Eric Eastwood
2024-08-21 14:39:54 -05:00
parent a5e06c6a8d
commit d3f90e4bd8
4 changed files with 285 additions and 197 deletions
Download Patch File Download Diff File Expand all files Collapse all files
synapse/storage/databases/main/events.py
+24 -71
View File
@@ -1549,20 +1549,24 @@ class PersistEventsStore:
txn, {m for m in members_to_cache_bust if not self.hs.is_mine_id(m)}
)
# TODO: We can probably remove this function in favor of other stuff.
# TODO: This doesn't take into account redactions
@classmethod
def _get_relevant_sliding_sync_current_state_event_ids_txn(
cls, txn: LoggingTransaction, room_id: str
) -> MutableStateMap[str]:
) -> Tuple[MutableStateMap[str], int]:
"""
Fetch the current state event IDs for the relevant (to the
`sliding_sync_joined_rooms` table) state types for the given room.
Returns:
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`.
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
(
@@ -1587,75 +1591,25 @@ class PersistEventsStore:
(event_type, state_key): event_id for event_id, event_type, state_key in txn
}
return current_state_map
# TODO: We can probably remove this function in favor of other stuff.
# TODO: Should we put this next to the other `_get_sliding_sync_*` function?
@classmethod
def _get_sliding_sync_insert_values_from_state_ids_map_txn(
cls, txn: LoggingTransaction, state_map: StateMap[str]
) -> SlidingSyncStateInsertValues:
"""
Fetch events in the `state_map` and extract the relevant state values 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 = {}
# Fetch the raw event JSON from the database
(
event_id_in_list_clause,
event_id_args,
) = make_in_list_sql_clause(
txn.database_engine,
"event_id",
state_map.values(),
)
txn.execute(
f"""
SELECT type, state_key, json FROM event_json
INNER JOIN events USING (event_id)
WHERE {event_id_in_list_clause}
"""
SELECT stream_id
FROM current_state_delta_stream
WHERE
room_id = ?
ORDER BY stream_id DESC
LIMIT 1
""",
event_id_args,
(room_id,),
)
row = txn.fetchone()
# If we're able to fetch the `current_state_events` above, we should have rows
# in `current_state_delta_stream` as well.
assert row, "Failed to fetch the `last_current_state_delta_stream_id`"
last_current_state_delta_stream_id = row[0]
# Parse the raw event JSON
for row in txn:
event_type, state_key, json = row
event_json = db_to_json(json)
return current_state_map, last_current_state_delta_stream_id
if event_type == EventTypes.Create:
room_type = event_json.get("content", {}).get(
EventContentFields.ROOM_TYPE
)
sliding_sync_insert_map["room_type"] = room_type
elif event_type == EventTypes.RoomEncryption:
encryption_algorithm = event_json.get("content", {}).get(
EventContentFields.ENCRYPTION_ALGORITHM
)
is_encrypted = encryption_algorithm is not None
sliding_sync_insert_map["is_encrypted"] = is_encrypted
elif event_type == EventTypes.Name:
room_name = event_json.get("content", {}).get(
EventContentFields.ROOM_NAME
)
sliding_sync_insert_map["room_name"] = room_name
else:
# We only expect to see events according to the
# `SLIDING_SYNC_RELEVANT_STATE_SET`.
raise AssertionError(
f"Unexpected event (we should not be fetching extra events): ({event_type}, {state_key})"
)
return sliding_sync_insert_map
# TODO: Should we put this next to the other `_get_sliding_sync_*` functions?
@classmethod
def _get_sliding_sync_insert_values_from_state_map(
cls, state_map: StateMap[EventBase]
@@ -1699,7 +1653,6 @@ class PersistEventsStore:
return sliding_sync_insert_map
# TODO: Should we put this next to the other `_get_sliding_sync_*` function?
@classmethod
def _get_sliding_sync_insert_values_from_stripped_state_txn(
cls, txn: LoggingTransaction, unsigned_stripped_state_events: Any
synapse/storage/databases/main/events_bg_updates.py
+134 -95
View File
@@ -39,11 +39,12 @@ from synapse.storage.databases.main.events import (
PersistEventsStore,
SlidingSyncMembershipInfo,
SlidingSyncMembershipSnapshotSharedInsertValues,
SlidingSyncStateInsertValues,
)
from synapse.storage.databases.main.events_worker import EventsWorkerStore
from synapse.storage.engines import BaseDatabaseEngine
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, StateMap, 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
@@ -111,7 +112,7 @@ class _CalculateChainCover:
finished_room_map: Dict[str, Tuple[int, int]]
class EventsBackgroundUpdatesStore(EventsWorkerStore, SQLBaseStore):
class EventsBackgroundUpdatesStore(StreamWorkerStore, StateDeltasStore, SQLBaseStore):
def __init__(
self,
database: DatabasePool,
@@ -1549,46 +1550,7 @@ class EventsBackgroundUpdatesStore(EventsWorkerStore, SQLBaseStore):
"""
last_room_id = progress.get("last_room_id", "")
def make_sql_clause_for_get_last_event_pos_in_room(
database_engine: BaseDatabaseEngine,
event_types: Optional[StrCollection] = None,
) -> Tuple[str, list]:
"""
Returns the ID and event position of the last event in a room at or before a
stream ordering.
Based on `get_last_event_pos_in_room_before_stream_ordering(...)`
Args:
database_engine
event_types: Optional allowlist of event types to filter by
Returns:
A tuple of SQL query and the args
"""
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(
database_engine, "type", event_types
)
event_type_clause = f"AND {event_type_clause}"
sql = f"""
SELECT stream_ordering
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
"""
return sql, event_type_args
def _txn(txn: LoggingTransaction) -> int:
def _get_rooms_to_update_txn(txn: LoggingTransaction) -> List[str]:
# Fetch the set of room IDs that we want to update
txn.execute(
"""
@@ -1600,85 +1562,162 @@ class EventsBackgroundUpdatesStore(EventsWorkerStore, SQLBaseStore):
(last_room_id, batch_size),
)
rooms_to_update_rows = txn.fetchall()
if not rooms_to_update_rows:
return 0
rooms_to_update_rows = cast(List[Tuple[str]], txn.fetchall())
for (room_id,) in rooms_to_update_rows:
# TODO: Handle redactions
current_state_map = PersistEventsStore._get_relevant_sliding_sync_current_state_event_ids_txn(
txn, room_id
return [row[0] for row in rooms_to_update_rows]
rooms_to_update = await self.db_pool.runInteraction(
"_sliding_sync_joined_rooms_backfill._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_BACKFILL
)
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/last_current_state_delta_stream_id values
joined_room_stream_ordering_updates: Dict[str, Tuple[int, int, int]] = {}
for room_id in rooms_to_update:
current_state_ids_map, last_current_state_delta_stream_id = (
await self.db_pool.runInteraction(
"_sliding_sync_joined_rooms_backfill._get_relevant_sliding_sync_current_state_event_ids_txn",
PersistEventsStore._get_relevant_sliding_sync_current_state_event_ids_txn,
room_id,
)
# We're iterating over rooms pulled from the current_state_events table
# so we should have some current state for each room
assert current_state_map
)
# We're iterating over rooms pulled from the current_state_events table
# so we should have some current state for each room
assert current_state_ids_map
sliding_sync_joined_rooms_insert_map = PersistEventsStore._get_sliding_sync_insert_values_from_state_ids_map_txn(
txn, current_state_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()
}
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 sliding_sync_joined_rooms_insert_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
), "We should not be seeing `None` here because the room should at-least have a create event"
# Figure out the latest bump_stamp in the room
bump_stamp_event_pos_results = await self.get_last_event_pos_in_room(
room_id, event_types=SLIDING_SYNC_DEFAULT_BUMP_EVENT_TYPES
)
assert bump_stamp_event_pos_results, (
"We should not be seeing `None` here because the room should at-least have a create event "
+ "(unless `SLIDING_SYNC_DEFAULT_BUMP_EVENT_TYPES` no longer includes the room create event)"
)
joined_room_stream_ordering_updates[room_id] = (
most_recent_event_pos_results[1].stream,
bump_stamp_event_pos_results[1].stream,
last_current_state_delta_stream_id,
)
def _backfill_table_txn(txn: LoggingTransaction) -> None:
last_successful_room_id: Optional[str] = None
for room_id, insert_map in joined_room_updates.items():
(
most_recent_event_stream_ordering_clause,
most_recent_event_stream_ordering_args,
) = make_sql_clause_for_get_last_event_pos_in_room(
txn.database_engine, event_types=None
)
bump_stamp_clause, bump_stamp_args = (
make_sql_clause_for_get_last_event_pos_in_room(
txn.database_engine,
event_types=SLIDING_SYNC_DEFAULT_BUMP_EVENT_TYPES,
event_stream_ordering,
bump_stamp,
last_current_state_delta_stream_id,
) = joined_room_stream_ordering_updates[room_id]
# Check if the current state has been updated since we gathered it
state_deltas_since_we_gathered_current_state = (
self.get_current_state_deltas_for_room_txn(
txn,
room_id,
from_token=RoomStreamToken(
stream=last_current_state_delta_stream_id
),
to_token=None,
)
)
for state_delta in state_deltas_since_we_gathered_current_state:
# We only need to check if 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:
# Save our progress before we exit early
if last_successful_room_id is not None:
self.db_pool.updates._background_update_progress_txn(
txn,
_BackgroundUpdates.SLIDING_SYNC_JOINED_ROOMS_BACKFILL,
{"last_room_id": room_id},
)
# Raising exception so we can just exit and 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."
)
# Pulling keys/values separately is safe and will produce congruent
# lists
insert_keys = sliding_sync_joined_rooms_insert_map.keys()
insert_values = sliding_sync_joined_rooms_insert_map.values()
insert_keys = insert_map.keys()
insert_values = insert_map.values()
# Since we partially update the `sliding_sync_joined_rooms` as new state
# is sent, we need to update the fields `ON CONFLICT`. We just have to be careful
# we're not overwriting it with stale data.
sql = f"""
INSERT INTO sliding_sync_joined_rooms
(room_id, event_stream_ordering, bump_stamp, {", ".join(insert_keys)})
VALUES (
?,
({most_recent_event_stream_ordering_clause}),
({bump_stamp_clause}),
?, ?, ?,
{", ".join("?" for _ in insert_values)}
)
ON CONFLICT (room_id)
DO UPDATE SET
event_stream_ordering = EXCLUDED.event_stream_ordering,
bump_stamp = EXCLUDED.bump_stamp,
event_stream_ordering = CASE
WHEN event_stream_ordering IS NULL OR event_stream_ordering < EXCLUDED.event_stream_ordering
THEN EXCLUDED.event_stream_ordering
ELSE event_stream_ordering
END,
bump_stamp = CASE
WHEN bump_stamp IS NULL OR bump_stamp < EXCLUDED.bump_stamp
THEN EXCLUDED.bump_stamp
ELSE bump_stamp
END,
{", ".join(f"{key} = EXCLUDED.{key}" for key in insert_keys)}
"""
args = (
[room_id, room_id]
+ most_recent_event_stream_ordering_args
+ [room_id]
+ bump_stamp_args
+ list(insert_values)
args = [room_id, event_stream_ordering, bump_stamp] + list(
insert_values
)
txn.execute(sql, args)
self.db_pool.updates._background_update_progress_txn(
txn,
_BackgroundUpdates.SLIDING_SYNC_JOINED_ROOMS_BACKFILL,
{"last_room_id": rooms_to_update_rows[-1][0]},
)
# Keep track of the last successful room_id
last_successful_room_id = room_id
return len(rooms_to_update_rows)
count = await self.db_pool.runInteraction(
"sliding_sync_joined_rooms_backfill", _txn
await self.db_pool.runInteraction(
"sliding_sync_joined_rooms_backfill", _backfill_table_txn
)
if not count:
await self.db_pool.updates._end_background_update(
_BackgroundUpdates.SLIDING_SYNC_JOINED_ROOMS_BACKFILL
)
# Update the progress
await self.db_pool.updates._background_update_progress(
_BackgroundUpdates.SLIDING_SYNC_JOINED_ROOMS_BACKFILL,
{"last_room_id": rooms_to_update[-1]},
)
return count
return len(rooms_to_update)
async def _sliding_sync_membership_snapshots_backfill(
self, progress: JsonDict, batch_size: int
synapse/storage/databases/main/state_deltas.py
+66 -31
View File
@@ -161,45 +161,80 @@ class StateDeltasStore(SQLBaseStore):
self._get_max_stream_id_in_current_state_deltas_txn,
)
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.
(> `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 = ? {from_clause} {to_clause}
ORDER BY stream_id ASC
"""
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])
]
@trace
async def get_current_state_deltas_for_room(
self, room_id: str, from_token: RoomStreamToken, to_token: RoomStreamToken
self,
room_id: str,
*,
from_token: Optional[RoomStreamToken],
to_token: Optional[RoomStreamToken],
) -> List[StateDelta]:
"""Get the state deltas between two tokens."""
"""
Get the state deltas between two tokens.
if not self._curr_state_delta_stream_cache.has_entity_changed(
room_id, from_token.stream
(> `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 []
def get_current_state_deltas_for_room_txn(
txn: LoggingTransaction,
) -> List[StateDelta]:
sql = """
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 <= ?
ORDER BY stream_id ASC
"""
txn.execute(
sql, (room_id, from_token.stream, to_token.get_max_stream_pos())
)
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 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
synapse/storage/databases/main/stream.py
+61
View File
@@ -1263,6 +1263,67 @@ 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(Tuple[str, int, str], txn.fetchone())
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 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,