Make historical events discoverable from backfill for servers without any scrollback history (MSC2716) (federation)

synapse/api/constants.py

@@ -196,9 +196,6 @@ class EventContentFields:
     MSC2716_CHUNK_ID = "org.matrix.msc2716.chunk_id"
     # For "marker" events
     MSC2716_MARKER_INSERTION = "org.matrix.msc2716.marker.insertion"
-    MSC2716_MARKER_INSERTION_PREV_EVENTS = (
-        "org.matrix.msc2716.marker.insertion_prev_events"
-    )
 
 
 class RoomTypes:

synapse/storage/databases/main/event_federation.py

@@ -936,15 +936,50 @@ def _get_backfill_events(self, txn, room_id, event_list, limit):
         # We want to make sure that we do a breadth-first, "depth" ordered
         # search.
 
+        # Look for the prev_event_id connected to the given event_id
         query = (
             "SELECT depth, prev_event_id FROM event_edges"
+            # Get the depth of the prev_event_id from the events table
             " INNER JOIN events"
             " ON prev_event_id = events.event_id"
+            # Find an event which matches the given event_id
             " WHERE event_edges.event_id = ?"
             " AND event_edges.is_state = ?"
             " LIMIT ?"
         )
 
+        # Look for the "insertion" events connected to the given event_id
+        # TODO: Do we need to worry about selecting only from the given room_id? The other query above doesn't
+        connected_insertion_event_query = (
+            "SELECT e.depth, i.insertion_event_id FROM insertion_event_edges AS i"
+            # Get the depth of the insertion event from the events table
+            " INNER JOIN events AS e"
+            " ON e.event_id = i.insertion_event_id"
+            # Find an insertion event which points via prev_events to the given event_id
+            " WHERE i.insertion_prev_event_id = ?"
+            " LIMIT ?"
+        )
+
+        # Find any chunk connections of a given insertion event
+        # TODO: Do we need to worry about selecting only from the given room_id? The other query above doesn't
+        chunk_connection_query = (
+            "SELECT e.depth, c.event_id FROM insertion_events AS i"
+            # Find the chunk that connects to the given insertion event
+            " INNER JOIN chunk_edges AS c"
+            " ON i.next_chunk_id = c.chunk_id"
+            # Get the depth of the chunk start event from the events table
+            " INNER JOIN events AS e"
+            " ON e.event_id = c.event_id"
+            # Find an insertion event which matches the given event_id
+            " WHERE i.insertion_event_id = ?"
+            " LIMIT ?"
+        )
+
+        # In a PriorityQueue, the lowest valued entries are retrieved first.
+        # We're using depth as the priority in the queue.
+        # Depth is lowest at the oldest-in-time message and highest and
+        # newest-in-time message. We add events to the queue with a negative depth so that
+        # we process the newest-in-time messages first going backwards in time.
         queue = PriorityQueue()
 
         for event_id in event_list:
@@ -970,9 +1005,38 @@ def _get_backfill_events(self, txn, room_id, event_list, limit):
 
             event_results.add(event_id)
 
+            txn.execute(
+                connected_insertion_event_query, (event_id, limit - len(event_results))
+            )
+            connected_insertion_event_id_results = list(txn)
+            logger.debug(
+                "_get_backfill_events: connected_insertion_event_query %s",
+                connected_insertion_event_id_results,
+            )
+            for row in connected_insertion_event_id_results:
+                if row[1] not in event_results:
+                    queue.put((-row[0], row[1]))
+
+                    # Find any chunk connections for the given insertion event
+                    txn.execute(
+                        chunk_connection_query, (row[1], limit - len(event_results))
+                    )
+                    chunk_start_event_id_results = list(txn)
+                    logger.debug(
+                        "_get_backfill_events: chunk_start_event_id_results %s",
+                        chunk_start_event_id_results,
+                    )
+                    for row in chunk_start_event_id_results:
+                        if row[1] not in event_results:
+                            queue.put((-row[0], row[1]))
+
             txn.execute(query, (event_id, False, limit - len(event_results)))
+            prev_event_id_results = list(txn)
+            logger.debug(
+                "_get_backfill_events: prev_event_ids %s", prev_event_id_results
+            )
 
-            for row in txn:
+            for row in prev_event_id_results:
                 if row[1] not in event_results:
                     queue.put((-row[0], row[1]))
 

synapse/storage/databases/main/events.py

@@ -1504,6 +1504,9 @@ def _update_metadata_tables_txn(
 
             self._handle_event_relations(txn, event)
 
+            self._handle_insertion_event(txn, event)
+            self._handle_chunk_id(txn, event)
+
             # Store the labels for this event.
             labels = event.content.get(EventContentFields.LABELS)
             if labels:
@@ -1756,6 +1759,75 @@ def _handle_event_relations(self, txn, event):
         if rel_type == RelationTypes.REPLACE:
             txn.call_after(self.store.get_applicable_edit.invalidate, (parent_id,))
 
+    def _handle_insertion_event(self, txn: LoggingTransaction, event: EventBase):
+        """Handles inserting insertion extremeties during peristence of marker events
+
+        Args:
+            txn: The database transaction object
+            event: The event to process
+        """
+
+        if event.type != EventTypes.MSC2716_INSERTION:
+            # Not a insertion event
+            return
+
+        logger.debug("_handle_insertion_event %s", event)
+
+        next_chunk_id = event.content.get(EventContentFields.MSC2716_NEXT_CHUNK_ID)
+        if next_chunk_id is None:
+            # Invalid insertion event without next chunk ID
+            return
+
+        # Keep track of the insertion event and the chunk ID
+        self.db_pool.simple_insert_txn(
+            txn,
+            table="insertion_events",
+            values={
+                "insertion_event_id": event.event_id,
+                "room_id": event.room_id,
+                "next_chunk_id": next_chunk_id,
+            },
+        )
+
+        # Insert an edge for every prev_event connection
+        for prev_event_id in event.prev_events:
+            self.db_pool.simple_insert_txn(
+                txn,
+                table="insertion_event_edges",
+                values={
+                    "insertion_event_id": event.event_id,
+                    "room_id": event.room_id,
+                    "insertion_prev_event_id": prev_event_id,
+                },
+            )
+
+    def _handle_chunk_id(self, txn: LoggingTransaction, event: EventBase):
+        """Handles inserting the chunk connections between the event at the
+        start of a chunk and an insertion event
+
+        Args:
+            txn: The database transaction object
+            event: The event to process
+        """
+
+        chunk_id = event.content.get(EventContentFields.MSC2716_CHUNK_ID)
+        if chunk_id is None:
+            # No chunk connection to persist
+            return
+
+        logger.debug("_handle_chunk_id %s %s", chunk_id, event)
+
+        # Keep track of the insertion event and the chunk ID
+        self.db_pool.simple_insert_txn(
+            txn,
+            table="chunk_edges",
+            values={
+                "event_id": event.event_id,
+                "room_id": event.room_id,
+                "chunk_id": chunk_id,
+            },
+        )
+
     def _handle_redaction(self, txn, redacted_event_id):
         """Handles receiving a redaction and checking whether we need to remove
         any redacted relations from the database.

synapse/storage/schema/main/delta/61/01insertion_event_lookups.sql

@@ -0,0 +1,50 @@
+/* Copyright 2021 The Matrix.org Foundation C.I.C
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+-- Add a table that keeps track of "insertion" events back in the history
+-- when we get a "marker" event over the "live" timeline. When navigating the DAG
+-- and we hit an event which matches `insertion_prev_event_id`, it should backfill 
+-- the "insertion" event and start navigating from there.
+
+CREATE TABLE IF NOT EXISTS insertion_events(
+    insertion_event_id TEXT NOT NULL,
+    room_id TEXT NOT NULL,
+    next_chunk_id TEXT NOT NULL,
+    UNIQUE (insertion_event_id, room_id, next_chunk_id)
+);
+
+CREATE INDEX IF NOT EXISTS insertion_events_insertion_room_id ON insertion_events(room_id);
+CREATE INDEX IF NOT EXISTS insertion_events_insertion_event_id ON insertion_events(insertion_event_id);
+CREATE INDEX IF NOT EXISTS insertion_events_next_chunk_id ON insertion_events(next_chunk_id);
+
+CREATE TABLE IF NOT EXISTS insertion_event_edges(
+    insertion_event_id TEXT NOT NULL,
+    room_id TEXT NOT NULL,
+    insertion_prev_event_id TEXT NOT NULL,
+    UNIQUE (insertion_event_id, room_id, insertion_prev_event_id)
+);
+
+CREATE INDEX IF NOT EXISTS insertion_event_edges_insertion_room_id ON insertion_event_edges(room_id);
+CREATE INDEX IF NOT EXISTS insertion_event_edges_insertion_event_id ON insertion_event_edges(insertion_event_id);
+CREATE INDEX IF NOT EXISTS insertion_event_edges_insertion_prev_event_id ON insertion_event_edges(insertion_prev_event_id);
+
+CREATE TABLE IF NOT EXISTS chunk_edges(
+    event_id TEXT NOT NULL,
+    room_id TEXT NOT NULL,
+    chunk_id TEXT NOT NULL,
+    UNIQUE (event_id, room_id)
+);
+
+CREATE INDEX IF NOT EXISTS chunk_edges_chunk_id ON chunk_edges(chunk_id);