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

scripts-dev/complement.sh

@@ -65,4 +65,4 @@ if [[ -n "$1" ]]; then
 fi
 
 # Run the tests!
-go test -v -tags synapse_blacklist,msc2946,msc3083,msc2716,msc2403 -count=1 $EXTRA_COMPLEMENT_ARGS ./tests
+go test -v -tags synapse_blacklist,msc2946,msc3083,msc2716,msc2403 -count=1 $EXTRA_COMPLEMENT_ARGS ./tests -run TestBackfillingHistory

synapse/events/utils.py

@@ -253,13 +253,13 @@ def format_event_for_client_v1(d):
 
 def format_event_for_client_v2(d):
     drop_keys = (
-        "auth_events",
-        "prev_events",
-        "hashes",
-        "signatures",
-        "depth",
-        "origin",
-        "prev_state",
+        # "auth_events",
+        # "prev_events",
+        # "hashes",
+        # "signatures",
+        # "depth",
+        # "origin",
+        # "prev_state",
     )
     for key in drop_keys:
         d.pop(key, None)

synapse/handlers/federation.py

@@ -1054,10 +1054,14 @@ async def maybe_backfill(
         with (await self._room_backfill.queue(room_id)):
             return await self._maybe_backfill_inner(room_id, current_depth, limit)
 
+    # Todo
     async def _maybe_backfill_inner(
         self, room_id: str, current_depth: int, limit: int
     ) -> bool:
         extremities = await self.store.get_oldest_events_with_depth_in_room(room_id)
+        logger.info(
+            "_maybe_backfill_inner extremities(%d)=%s", len(extremities), extremities
+        )
 
         if not extremities:
             logger.debug("Not backfilling as no extremeties found.")
@@ -2123,8 +2127,18 @@ async def on_backfill_request(
         limit = min(limit, 100)
 
         events = await self.store.get_backfill_events(room_id, pdu_list, limit)
+        logger.info(
+            "on_backfill_request get_backfill_events events(%d)=%s",
+            len(events),
+            [f'{ev.content.get("body")}: {ev.type} ({ev.event_id})' for ev in events],
+        )
 
         events = await filter_events_for_server(self.storage, origin, events)
+        logger.info(
+            "on_backfill_request filter_events_for_server events(%d)=%s",
+            len(events),
+            events,
+        )
 
         return events
 

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,36 @@ 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.info(
+                "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.info(
+                        "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.info("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,10 @@ def _update_metadata_tables_txn(
 
             self._handle_event_relations(txn, event)
 
+            self._handle_insertion_event(txn, event)
+            self._handle_marker_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 +1760,112 @@ 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, event):
+        """Handles inserting insertion extremeties during peristence of marker events
+
+        Args:
+            txn
+            event (EventBase)
+        """
+
+        if event.type != EventTypes.MSC2716_INSERTION:
+            # Not a insertion event
+            return
+
+        logger.info("_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_marker_event(self, txn, event):
+        """Handles inserting insertion extremeties during peristence of marker events
+
+        Args:
+            txn
+            event (EventBase)
+        """
+
+        if event.type != EventTypes.MSC2716_MARKER:
+            # Not a marker event
+            return
+
+        logger.info("_handle_marker_event %s", event)
+
+        # TODO: We should attempt to backfill the insertion event instead
+        # of trying to  pack all of the info in the marker event. Otherwise,
+        # we need to pack in the insertion_prev_events and insertion_next_chunk_id.
+
+        # insertion_event_id = event.content.get(
+        #     EventContentFields.MSC2716_MARKER_INSERTION
+        # )
+        # insertion_prev_event_ids = event.content.get(
+        #     EventContentFields.MSC2716_MARKER_INSERTION_PREV_EVENTS
+        # )
+        # if not insertion_event_id or not insertion_prev_event_ids:
+        #     # Invalid marker event
+        #     return
+
+        # for prev_event_id in insertion_prev_event_ids:
+        #     self.db_pool.simple_insert_txn(
+        #         txn,
+        #         table="insertion_event_edges",
+        #         values={
+        #             "insertion_event_id": insertion_event_id,
+        #             "room_id": event.room_id,
+        #             "insertion_prev_event_id": prev_event_id,
+        #         },
+        #     )
+
+    def _handle_chunk_id(self, txn, event):
+        """Handles inserting the chunk connections between the event at the
+        start of a chunk and an insertion event
+
+        Args: txn event (EventBase)
+        """
+
+        chunk_id = event.content.get(EventContentFields.MSC2716_CHUNK_ID)
+        if chunk_id is None:
+            # No chunk connection to persist
+            return
+
+        logger.info("_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);