Draft: Add support for MSC2716 marker events

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/parallel/Historical_messages_are_visible_when_already_joined_on_federated_server

synapse/handlers/federation.py

@@ -42,6 +42,7 @@
 
 from synapse import event_auth
 from synapse.api.constants import (
+    EventContentFields,
     EventTypes,
     Membership,
     RejectedReason,
@@ -263,6 +264,7 @@ async def on_receive_pdu(
         state = None
 
         # Get missing pdus if necessary.
+        # We don't need to worry about outliers because TODO!
         if not pdu.internal_metadata.is_outlier():
             # We only backfill backwards to the min depth.
             min_depth = await self.get_min_depth_for_context(pdu.room_id)
@@ -889,6 +891,63 @@ async def _process_received_pdu(
                     "resync_device_due_to_pdu", self._resync_device, event.sender
                 )
 
+        await self._handle_marker_event(origin, event)
+
+    async def _handle_marker_event(self, origin: str, marker_event: EventBase):
+        """Handles backfilling the insertion event when we receive a marker
+        event that points to one
+
+        Args:
+            origin: Origin of the event. Will be called to get the insertion event
+            event: The event to process
+        """
+
+        if marker_event.type != EventTypes.MSC2716_MARKER:
+            # Not a marker event
+            return
+
+        logger.info("_handle_marker_event: received %s", marker_event)
+
+        insertion_event_id = marker_event.content.get(
+            EventContentFields.MSC2716_MARKER_INSERTION
+        )
+
+        if insertion_event_id is None:
+            # Nothing to retrieve then (invalid marker)
+            return
+
+        logger.info(
+            "_handle_marker_event: backfilling insertion event %s", insertion_event_id
+        )
+
+        await self._get_events_and_persist(
+            origin,
+            marker_event.room_id,
+            [insertion_event_id],
+        )
+
+        insertion_event = await self.store.get_event(
+            insertion_event_id, allow_none=True
+        )
+        if insertion_event is None:
+            logger.warning(
+                "_handle_marker_event: server %s didn't return insertion event %s for marker %s",
+                origin,
+                insertion_event_id,
+                marker_event.event_id,
+            )
+            return
+
+        logger.info(
+            "_handle_marker_event: Succesfully backfilled insertion event %s from marker event %s",
+            insertion_event,
+            marker_event,
+        )
+
+        await self.store.insert_backward_extremity(
+            insertion_event_id, marker_event.room_id
+        )
+
     async def _resync_device(self, sender: str) -> None:
         """We have detected that the device list for the given user may be out
         of sync, so we try and resync them.
@@ -1057,7 +1116,12 @@ async def maybe_backfill(
     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)
+        oldest_events = await self.store.get_oldest_events_with_depth_in_room(room_id)
+        insertion_events_to_be_backfilled = (
+            await self.store.get_insertion_event_backwards_extremities_in_room(room_id)
+        )
+        extremities = {**oldest_events, **insertion_events_to_be_backfilled}
+        logger.info("_maybe_backfill_inner: extremities %s", extremities)
 
         if not extremities:
             logger.debug("Not backfilling as no extremeties found.")
@@ -1090,12 +1154,14 @@ async def _maybe_backfill_inner(
         #   types have.
 
         forward_events = await self.store.get_successor_events(list(extremities))
+        logger.info("_maybe_backfill_inner: forward_events %s", forward_events)
 
         extremities_events = await self.store.get_events(
             forward_events,
             redact_behaviour=EventRedactBehaviour.AS_IS,
             get_prev_content=False,
         )
+        logger.info("_maybe_backfill_inner: extremities_events %s", extremities_events)
 
         # We set `check_history_visibility_only` as we might otherwise get false
         # positives from users having been erased.
@@ -1106,6 +1172,9 @@ async def _maybe_backfill_inner(
             redact=False,
             check_history_visibility_only=True,
         )
+        logger.info(
+            "_maybe_backfill_inner: filtered_extremities %s", filtered_extremities
+        )
 
         if not filtered_extremities:
             return False
@@ -1124,15 +1193,15 @@ async def _maybe_backfill_inner(
         # much larger factor will result in triggering a backfill request much
         # earlier than necessary.
         if current_depth - 2 * limit > max_depth:
-            logger.debug(
+            logger.info(
                 "Not backfilling as we don't need to. %d < %d - 2 * %d",
                 max_depth,
                 current_depth,
                 limit,
             )
             return False
 
-        logger.debug(
+        logger.info(
             "room_id: %s, backfill: current_depth: %s, max_depth: %s, extrems: %s",
             room_id,
             current_depth,

synapse/storage/databases/main/event_federation.py

@@ -666,26 +666,48 @@ def _get_auth_chain_difference_txn(
         return {eid for eid, n in event_to_missing_sets.items() if n}
 
     async def get_oldest_events_with_depth_in_room(self, room_id):
+        def get_oldest_events_with_depth_in_room_txn(txn, room_id):
+            sql = (
+                "SELECT b.event_id, MAX(e.depth) FROM events as e"
+                " INNER JOIN event_edges as g"
+                " ON g.event_id = e.event_id"
+                " INNER JOIN event_backward_extremities as b"
+                " ON g.prev_event_id = b.event_id"
+                " WHERE b.room_id = ? AND g.is_state is ?"
+                " GROUP BY b.event_id"
+            )
+
+            txn.execute(sql, (room_id, False))
+
+            return dict(txn)
+
         return await self.db_pool.runInteraction(
             "get_oldest_events_with_depth_in_room",
-            self.get_oldest_events_with_depth_in_room_txn,
+            get_oldest_events_with_depth_in_room_txn,
             room_id,
         )
 
-    def get_oldest_events_with_depth_in_room_txn(self, txn, room_id):
-        sql = (
-            "SELECT b.event_id, MAX(e.depth) FROM events as e"
-            " INNER JOIN event_edges as g"
-            " ON g.event_id = e.event_id"
-            " INNER JOIN event_backward_extremities as b"
-            " ON g.prev_event_id = b.event_id"
-            " WHERE b.room_id = ? AND g.is_state is ?"
-            " GROUP BY b.event_id"
-        )
+    async def get_insertion_event_backwards_extremities_in_room(self, room_id):
+        def get_insertion_event_backwards_extremities_in_room_txn(txn, room_id):
+            sql = """
+                SELECT b.event_id, MAX(e.depth) FROM insertion_events as i
+                /* We only want insertion events that are also marked as backwards extremities */
+                INNER JOIN event_backward_extremities as b USING (event_id)
+                /* Get the depth of the insertion event from the events table */
+                INNER JOIN events AS e USING (event_id)
+                WHERE b.room_id = ?
+                GROUP BY b.event_id
+            """
 
-        txn.execute(sql, (room_id, False))
+            txn.execute(sql, (room_id,))
 
-        return dict(txn)
+            return dict(txn)
+
+        return await self.db_pool.runInteraction(
+            "get_insertion_event_backwards_extremities_in_room",
+            get_insertion_event_backwards_extremities_in_room_txn,
+            room_id,
+        )
 
     async def get_max_depth_of(self, event_ids: List[str]) -> Tuple[str, int]:
         """Returns the event ID and depth for the event that has the max depth from a set of event IDs
@@ -929,7 +951,7 @@ async def get_backfill_events(self, room_id: str, event_list: list, limit: int):
         return sorted(events, key=lambda e: -e.depth)
 
     def _get_backfill_events(self, txn, room_id, event_list, limit):
-        logger.debug("_get_backfill_events: %s, %r, %s", room_id, event_list, limit)
+        logger.info("_get_backfill_events: %s, %r, %s", room_id, event_list, limit)
 
         event_results = set()
 
@@ -1005,7 +1027,7 @@ def _get_backfill_events(self, txn, room_id, event_list, limit):
                 connected_insertion_event_query, (event_id, limit - len(event_results))
             )
             connected_insertion_event_id_results = txn.fetchall()
-            logger.debug(
+            logger.info(
                 "_get_backfill_events: connected_insertion_event_query %s",
                 connected_insertion_event_id_results,
             )
@@ -1020,7 +1042,7 @@ def _get_backfill_events(self, txn, room_id, event_list, limit):
                     (connected_insertion_event, limit - len(event_results)),
                 )
                 chunk_start_event_id_results = txn.fetchall()
-                logger.debug(
+                logger.info(
                     "_get_backfill_events: chunk_start_event_id_results %s",
                     chunk_start_event_id_results,
                 )
@@ -1030,7 +1052,7 @@ def _get_backfill_events(self, txn, room_id, event_list, limit):
 
             txn.execute(query, (event_id, False, limit - len(event_results)))
             prev_event_id_results = txn.fetchall()
-            logger.debug(
+            logger.info(
                 "_get_backfill_events: prev_event_ids %s", prev_event_id_results
             )
 
@@ -1122,6 +1144,21 @@ def _delete_old_forward_extrem_cache_txn(txn):
             _delete_old_forward_extrem_cache_txn,
         )
 
+    async def insert_backward_extremity(self, event_id: str, room_id: str) -> None:
+        def _insert_backward_extremity_txn(txn):
+            self.db_pool.simple_insert_txn(
+                txn,
+                table="event_backward_extremities",
+                values={
+                    "event_id": event_id,
+                    "room_id": room_id,
+                },
+            )
+
+        await self.db_pool.runInteraction(
+            "_insert_backward_extremity_txn", _insert_backward_extremity_txn
+        )
+
     async def insert_received_event_to_staging(
         self, origin: str, event: EventBase
     ) -> None:

synapse/storage/databases/main/events.py

@@ -1777,7 +1777,7 @@ def _handle_insertion_event(self, txn: LoggingTransaction, event: EventBase):
             # Invalid insertion event without next chunk ID
             return
 
-        logger.debug(
+        logger.info(
             "_handle_insertion_event (next_chunk_id=%s) %s", next_chunk_id, event
         )
 
@@ -1818,7 +1818,7 @@ def _handle_chunk_id(self, txn: LoggingTransaction, event: EventBase):
             # No chunk connection to persist
             return
 
-        logger.debug("_handle_chunk_id %s %s", chunk_id, event)
+        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(
@@ -2091,18 +2091,21 @@ def _update_backward_extremeties(self, txn, events):
         for ev in events:
             events_by_room.setdefault(ev.room_id, []).append(ev)
 
+        # From the events passed in, add all of the prev events as backwards extremities.
+        # Ignore any events that are already backwards extrems or outliers.
         query = (
             "INSERT INTO event_backward_extremities (event_id, room_id)"
             " SELECT ?, ? WHERE NOT EXISTS ("
-            " SELECT 1 FROM event_backward_extremities"
-            " WHERE event_id = ? AND room_id = ?"
+            "   SELECT 1 FROM event_backward_extremities"
+            "   WHERE event_id = ? AND room_id = ?"
             " )"
             " AND NOT EXISTS ("
-            " SELECT 1 FROM events WHERE event_id = ? AND room_id = ? "
-            " AND outlier = ?"
+            "   SELECT 1 FROM events WHERE event_id = ? AND room_id = ? "
+            "   AND outlier = ?"
             " )"
         )
 
+        logger.info("_update_backward_extremeties %s", events)
         txn.execute_batch(
             query,
             [
@@ -2113,6 +2116,8 @@ def _update_backward_extremeties(self, txn, events):
             ],
         )
 
+        # Delete all these events that we've already fetched and now know that their
+        # prev events are the new outliers.
         query = (
             "DELETE FROM event_backward_extremities"
             " WHERE event_id = ? AND room_id = ?"

synapse/visibility.py

@@ -366,13 +366,21 @@ def check_event_is_visible(event: EventBase, state: StateMap[EventBase]) -> bool
         if erased_senders:
             to_return = []
             for e in events:
-                if not is_sender_erased(e, erased_senders):
+                erased = is_sender_erased(e, erased_senders)
+                logger.info(
+                    "filter_events_for_server: (all_open) %s erased=%s", e, erased
+                )
+                if not erased:
                     to_return.append(e)
                 elif redact:
                     to_return.append(prune_event(e))
 
+            logger.info("filter_events_for_server: (all_open) to_return=%s", to_return)
             return to_return
 
+        logger.info(
+            "filter_events_for_server: all_open and no erased senders %s", events
+        )
         # If there are no erased users then we can just return the given list
         # of events without having to copy it.
         return events
@@ -429,6 +437,9 @@ def include(typ, state_key):
     for e in events:
         erased = is_sender_erased(e, erased_senders)
         visible = check_event_is_visible(e, event_to_state[e.event_id])
+        logger.info(
+            "filter_events_for_server: %s erased=%s visible=%s", e, erased, visible
+        )
         if visible and not erased:
             to_return.append(e)
         elif redact: