Assign a dumb stitched ordering to incoming events

For now, we just give each incoming event the next stitched ordering.

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",

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]]:

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:

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:
+-- <https://www.gnu.org/licenses/agpl-3.0.html>.
+
+ALTER TABLE events ADD COLUMN stitched_ordering BIGINT;
+CREATE UNIQUE INDEX events_stitched_order ON events(room_id, stitched_ordering); -- TODO make this concurrent
+