diff --git a/synapse/events/snapshot.py b/synapse/events/snapshot.py index 63551143d8..762d6319d5 100644 --- a/synapse/events/snapshot.py +++ b/synapse/events/snapshot.py @@ -130,6 +130,9 @@ class EventContext(UnpersistedEventContextBase): partial_state: if True, we may be storing this event with a temporary, incomplete state. + + stitched_ordering: the assigned stitched ordering for this event, or None if we + have not yet assigned one. """ _storage: "StorageControllers" @@ -142,6 +145,8 @@ class EventContext(UnpersistedEventContextBase): partial_state: bool = False + stitched_ordering: Optional[int] = None + @staticmethod def with_state( storage: "StorageControllers", diff --git a/synapse/storage/controllers/persist_events.py b/synapse/storage/controllers/persist_events.py index 120934af57..8f09903a27 100644 --- a/synapse/storage/controllers/persist_events.py +++ b/synapse/storage/controllers/persist_events.py @@ -616,6 +616,8 @@ class EventsPersistenceStorageController: if not events_and_contexts: return replaced_events + await self._assign_stitched_orders(room_id, events_and_contexts) + chunks = [ events_and_contexts[x : x + 100] for x in range(0, len(events_and_contexts), 100) @@ -676,6 +678,19 @@ class EventsPersistenceStorageController: return replaced_events + async def _assign_stitched_orders( + self, + room_id: str, + events_and_contexts: List[EventPersistencePair], + ) -> None: + current_max_stream_ordering = ( + await self.persist_events_store.get_room_max_stitched_ordering(room_id) or 0 + ) + + for _event, context in events_and_contexts: + current_max_stream_ordering += 1 + context.stitched_ordering = current_max_stream_ordering + async def _calculate_new_forward_extremities_and_state_delta( self, room_id: str, ev_ctx_rm: List[EventPersistencePair] ) -> Tuple[Optional[Set[str]], Optional[DeltaState]]: diff --git a/synapse/storage/databases/main/events.py b/synapse/storage/databases/main/events.py index abebf1e1b1..0c9555d59a 100644 --- a/synapse/storage/databases/main/events.py +++ b/synapse/storage/databases/main/events.py @@ -2687,6 +2687,7 @@ class PersistEventsStore: "contains_url", "state_key", "rejection_reason", + "stitched_ordering", ), values=[ ( @@ -2705,6 +2706,7 @@ class PersistEventsStore: "url" in event.content and isinstance(event.content["url"], str), event.get_state_key(), context.rejected, + context.stitched_ordering, ) for event, context in events_and_contexts ], @@ -3639,6 +3641,24 @@ class PersistEventsStore: backward_extremity_tuples_to_remove, ) + async def get_room_max_stitched_ordering(self, room_id: str) -> Optional[int]: + """Get the maximum stitched order for any event currently in the room. + + If no events in this room have an assigned stitched order, returns None. + """ + + def get_room_max_stitched_ordering_txn( + txn: LoggingTransaction, + ) -> Optional[int]: + sql = "SELECT MAX(stitched_ordering) FROM events WHERE room_id=?" + txn.execute(sql, [room_id]) + ret = [r[0] for r in txn] + return ret[0] + + return await self.db_pool.runInteraction( + "get_room_max_stitched_ordering", get_room_max_stitched_ordering_txn + ) + @attr.s(slots=True, auto_attribs=True) class _LinkMap: diff --git a/synapse/storage/schema/main/delta/92/10_stitched_order.sql b/synapse/storage/schema/main/delta/92/10_stitched_order.sql new file mode 100644 index 0000000000..c5a6e4337b --- /dev/null +++ b/synapse/storage/schema/main/delta/92/10_stitched_order.sql @@ -0,0 +1,16 @@ +-- +-- This file is licensed under the Affero General Public License (AGPL) version 3. +-- +-- Copyright (C) 2025 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: +-- . + +ALTER TABLE events ADD COLUMN stitched_ordering BIGINT; +CREATE UNIQUE INDEX events_stitched_order ON events(room_id, stitched_ordering); -- TODO make this concurrent +