Switch to chunk events for federation

synapse/storage/databases/main/event_federation.py

@@ -962,7 +962,7 @@ class EventFederationWorkerStore(EventsWorkerStore, SignatureWorkerStore, SQLBas
         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
+            INNER JOIN chunk_events 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 USING (event_id)

synapse/storage/databases/main/events.py

@@ -1505,7 +1505,7 @@ class PersistEventsStore:
             self._handle_event_relations(txn, event)
 
             self._handle_insertion_event(txn, event)
-            self._handle_chunk_id(txn, event)
+            self._handle_chunk_event(txn, event)
 
             # Store the labels for this event.
             labels = event.content.get(EventContentFields.LABELS)
@@ -1804,26 +1804,30 @@ class PersistEventsStore:
                 },
             )
 
-    def _handle_chunk_id(self, txn: LoggingTransaction, event: EventBase):
-        """Handles inserting the chunk edges/connections between the event at the
-        start of a chunk and an insertion event. Part of MSC2716.
+    def _handle_chunk_event(self, txn: LoggingTransaction, event: EventBase):
+        """Handles inserting the chunk edges/connections between the chunk event
+        and an insertion event. Part of MSC2716.
 
         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
+        if event.type != EventTypes.MSC2716_CHUNK:
+            # Not a chunk event
             return
 
-        logger.debug("_handle_chunk_id %s %s", chunk_id, event)
+        chunk_id = event.content.get(EventContentFields.MSC2716_CHUNK_ID)
+        if chunk_id is None:
+            # Invalid chunk event without a chunk ID
+            return
+
+        logger.debug("_handle_chunk_event 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",
+            table="chunk_events",
             values={
                 "event_id": event.event_id,
                 "room_id": event.room_id,

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

@@ -41,12 +41,12 @@ CREATE INDEX IF NOT EXISTS insertion_event_edges_event_id ON insertion_event_edg
 CREATE INDEX IF NOT EXISTS insertion_event_edges_insertion_prev_event_id ON insertion_event_edges(insertion_prev_event_id);
 
 -- Add a table that keeps track of how each chunk is labeled. The chunks are
--- connected together based insertion points `next_chunk_id`.
-CREATE TABLE IF NOT EXISTS chunk_edges(
+-- connected together based on an insertion events `next_chunk_id`.
+CREATE TABLE IF NOT EXISTS chunk_events(
     event_id TEXT NOT NULL,
     room_id TEXT NOT NULL,
     chunk_id TEXT NOT NULL,
     UNIQUE (event_id)
 );
 
-CREATE INDEX IF NOT EXISTS chunk_edges_chunk_id ON chunk_edges(chunk_id);
+CREATE INDEX IF NOT EXISTS chunk_events_chunk_id ON chunk_events(chunk_id);