Add tests and fixup

synapse/storage/persist_events.py

@@ -463,6 +463,14 @@ async def _persist_events(
                             )
                             current_state, delta_ids, new_latest_event_ids = res
 
+                            # there should always be at least one forward extremity.
+                            # (except during the initial persistence of the send_join
+                            # results, in which case there will be no existing
+                            # extremities, so we'll `continue` above and skip this bit.)
+                            assert new_latest_event_ids, "No forward extremities left!"
+
+                            new_forward_extremeties[room_id] = new_latest_event_ids
+
                         # If either are not None then there has been a change,
                         # and we need to work out the delta (or use that
                         # given)
@@ -760,22 +768,28 @@ async def _get_new_state_after_events(
 
             dropped = set(new_latest_event_ids) - new_new_extrems
 
+            logger.debug("Might drop events: %s", dropped)
+
             # We only drop events if:
             #   1. we're not currently persisting them;
             #   2. they're not our own events (or are dummy events); and
-            #   3. they're either over N minutes old or we have newer events
-            #      from the same domain.
+            #   3. they're either:
+            #       1. over N hours old and more than N events ago (we use depth
+            #          to calculate); or
+            #       2. we are persisting an event from the same domain.
             #
             # The idea is that we don't want to drop events that are "legitmate"
             # extremities (that we would want to include as prev events), only
             # "stuck" extremities that are e.g. due to a gap in the graph.
             #
             # Note that we either drop all of them or none of them. If we only
-            # drop some of the events we don't know if state res would come to the same conclusion.
+            # drop some of the events we don't know if state res would come to
+            # the same conclusion.
 
             drop = True
             for ev, _ in events_context:
                 if ev.event_id in dropped:
+                    logger.debug("Not dropping as in to persist list")
                     drop = False
                     break
 
@@ -786,26 +800,37 @@ async def _get_new_state_after_events(
                     redact_behaviour=EventRedactBehaviour.AS_IS,
                 )
 
-                new_senders = {e.sender for e, _ in events_context}
+                new_senders = {get_domain_from_id(e.sender) for e, _ in events_context}
 
-                twenty_minutes_ago = self._clock.time_msec() - 20 * 60 * 1000
+                one_day_ago = self._clock.time_msec() - 20 * 60 * 1000
+                current_depth = max(e.depth for e, _ in events_context)
                 for event in dropped_events.values():
                     if (
                         self.is_mine_id(event.sender)
                         and event.type != "org.matrix.dummy_event"
                     ):
+                        logger.debug("Not dropping own event")
                         drop = False
                         break
 
-                    if event.origin_server_ts < twenty_minutes_ago:
+                    if (
+                        event.origin_server_ts < one_day_ago
+                        and event.depth < current_depth - 100
+                    ):
                         continue
                     if get_domain_from_id(event.sender) in new_senders:
                         continue
 
+                    logger.debug(
+                        "Not dropping as too new and not in new_senders: %s",
+                        new_senders,
+                    )
+
                     drop = False
                     break
 
             if drop:
+                logger.debug("Dropping events %s", dropped)
                 new_latest_event_ids = new_new_extrems
 
         return res.state, None, new_latest_event_ids

tests/storage/test_events.py

@@ -0,0 +1,182 @@
+# -*- coding: utf-8 -*-
+# Copyright 2020 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.
+
+
+from synapse.api.constants import EventTypes, Membership
+from synapse.api.room_versions import RoomVersions
+from synapse.federation.federation_base import event_from_pdu_json
+from synapse.rest import admin
+from synapse.rest.client.v1 import login, room
+
+from tests.unittest import HomeserverTestCase
+
+
+class ExtremPruneTestCase(HomeserverTestCase):
+    servlets = [
+        admin.register_servlets,
+        room.register_servlets,
+        login.register_servlets,
+    ]
+
+    def test_prune_gap(self):
+        """Test that we drop extremities after a gap when we see an event from
+        the same domain.
+        """
+
+        state = self.hs.get_state_handler()
+        persistence = self.hs.get_storage().persistence
+        store = self.hs.get_datastore()
+
+        self.register_user("user", "pass")
+        token = self.login("user", "pass")
+        room_id = self.helper.create_room_as(
+            "user", room_version=RoomVersions.V6.identifier, tok=token
+        )
+
+        body = self.helper.send(room_id, body="Test", tok=token)
+        local_message_event_id = body["event_id"]
+
+        # Fudge a remote event an persist it. This will be the extremity before
+        # the gap.
+        remote_event_1 = event_from_pdu_json(
+            {
+                "type": EventTypes.Message,
+                "content": {},
+                "room_id": room_id,
+                "sender": "@user:other",
+                "depth": 5,
+                "prev_events": [local_message_event_id],
+                "auth_events": [],
+                "origin_server_ts": self.clock.time_msec(),
+            },
+            RoomVersions.V6,
+        )
+
+        context = self.get_success(state.compute_event_context(remote_event_1))
+        self.get_success(persistence.persist_event(remote_event_1, context))
+
+        # Check that the current extremities is the remote event.
+        extremities = self.get_success(store.get_prev_events_for_room(room_id))
+        self.assertCountEqual(extremities, [remote_event_1.event_id])
+
+        # Fudge a second event which points to an event we don't have. This is a
+        # state event so that the state changes (otherwise we won't prune the
+        # extremity as they'll have the same state group).
+        remote_event_2 = event_from_pdu_json(
+            {
+                "type": EventTypes.Member,
+                "state_key": "@user:other",
+                "content": {"membership": Membership.JOIN},
+                "room_id": room_id,
+                "sender": "@user:other",
+                "depth": 10,
+                "prev_events": ["$some_unknown_message"],
+                "auth_events": [],
+                "origin_server_ts": self.clock.time_msec(),
+            },
+            RoomVersions.V6,
+        )
+
+        state_before_gap = self.get_success(state.get_current_state(room_id))
+
+        context = self.get_success(
+            state.compute_event_context(
+                remote_event_2, old_state=state_before_gap.values()
+            )
+        )
+
+        self.get_success(persistence.persist_event(remote_event_2, context))
+
+        # Check the new extremity is just the new remote event.
+        extremities = self.get_success(store.get_prev_events_for_room(room_id))
+        self.assertCountEqual(extremities, [remote_event_2.event_id])
+
+    def test_dont_prune_gap_if_state_different(self):
+        """Test that we don't prune extremities after a gap if the resolved
+        state is different.
+        """
+
+        state = self.hs.get_state_handler()
+        persistence = self.hs.get_storage().persistence
+        store = self.hs.get_datastore()
+
+        self.register_user("user", "pass")
+        token = self.login("user", "pass")
+        room_id = self.helper.create_room_as(
+            "user", room_version=RoomVersions.V6.identifier, tok=token
+        )
+
+        body = self.helper.send(room_id, body="Test", tok=token)
+        local_message_event_id = body["event_id"]
+
+        # Fudge a remote event an persist it. This will be the extremity before
+        # the gap.
+        remote_event_1 = event_from_pdu_json(
+            {
+                "type": EventTypes.Message,
+                "state_key": "@user:other",
+                "content": {},
+                "room_id": room_id,
+                "sender": "@user:other",
+                "depth": 5,
+                "prev_events": [local_message_event_id],
+                "auth_events": [],
+                "origin_server_ts": self.clock.time_msec(),
+            },
+            RoomVersions.V6,
+        )
+
+        context = self.get_success(state.compute_event_context(remote_event_1))
+        self.get_success(persistence.persist_event(remote_event_1, context))
+
+        # Check that the current extremities is the remote event.
+        extremities = self.get_success(store.get_prev_events_for_room(room_id))
+        self.assertCountEqual(extremities, [remote_event_1.event_id])
+
+        # Fudge a second event which points to an event we don't have.
+        remote_event_2 = event_from_pdu_json(
+            {
+                "type": EventTypes.Message,
+                "state_key": "@user:other",
+                "content": {},
+                "room_id": room_id,
+                "sender": "@user:other",
+                "depth": 10,
+                "prev_events": ["$some_unknown_message"],
+                "auth_events": [],
+                "origin_server_ts": self.clock.time_msec(),
+            },
+            RoomVersions.V6,
+        )
+
+        # Now we persist it with state with a dropped history visibility
+        # setting. The state resolution across the old and new event will then
+        # include it, and so the resolved state won't match the new state.
+        state_before_gap = dict(self.get_success(state.get_current_state(room_id)))
+        state_before_gap.pop(("m.room.history_visibility", ""))
+
+        context = self.get_success(
+            state.compute_event_context(
+                remote_event_2, old_state=state_before_gap.values()
+            )
+        )
+
+        self.get_success(persistence.persist_event(remote_event_2, context))
+
+        # Check that we haven't dropped the old extremity.
+        extremities = self.get_success(store.get_prev_events_for_room(room_id))
+        self.assertCountEqual(
+            extremities, [remote_event_1.event_id, remote_event_2.event_id]
+        )