Add endpoints for backfilling history (MSC2716)

changelog.d/9247.feature

@@ -0,0 +1 @@
+Add experimental support for backfilling history into rooms ([MSC2716](https://github.com/matrix-org/matrix-doc/pull/2716)).

scripts-dev/complement.sh

@@ -11,12 +11,10 @@ cd "$(dirname $0)/.."
 docker build -t matrixdotorg/synapse:latest -f docker/Dockerfile .
 
 # Download Complement
-wget -N https://github.com/matrix-org/complement/archive/master.tar.gz
-tar -xzf master.tar.gz
-cd complement-master
+cd ../complement
 
 # Build the Synapse image from Complement, based on the above image we just built
 docker build -t complement-synapse -f dockerfiles/Synapse.Dockerfile ./dockerfiles
 
 # Run the tests on the resulting image!
-COMPLEMENT_BASE_IMAGE=complement-synapse go test -v -count=1 ./tests
+COMPLEMENT_BASE_IMAGE=complement-synapse go test -tags msc2716 -v -count=1 ./tests/main_test.go ./tests/msc2716_test.go -run TestBackfillingHistory/parallel/Backfilled_historical_events_resolve_with_proper_state

synapse/api/auth.py

@@ -88,9 +88,11 @@ async def check_from_context(
         self, room_version: str, event, context, do_sig_check=True
     ):
         prev_state_ids = await context.get_prev_state_ids()
+        logger.info("check_from_context prev_state_ids %s", prev_state_ids)
         auth_events_ids = self.compute_auth_events(
             event, prev_state_ids, for_verification=True
         )
+        logger.info("check_from_context auth_events_ids %s", auth_events_ids)
         auth_events = await self.store.get_events(auth_events_ids)
         auth_events = {(e.type, e.state_key): e for e in auth_events.values()}
 

synapse/config/experimental.py

@@ -27,6 +27,8 @@ def read_config(self, config: JsonDict, **kwargs):
 
         # MSC2858 (multiple SSO identity providers)
         self.msc2858_enabled = experimental.get("msc2858_enabled", False)  # type: bool
+        # MSC2716 (backfill existing history)
+        self.msc2716_enabled = experimental.get("msc2716_enabled", False)  # type: bool
         # Spaces (MSC1772, MSC2946, etc)
         self.spaces_enabled = experimental.get("spaces_enabled", False)  # type: bool
         # MSC3026 (busy presence state)

synapse/events/builder.py

@@ -12,6 +12,7 @@
 # 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.
+import logging
 from typing import Any, Dict, List, Optional, Tuple, Union
 
 import attr
@@ -34,6 +35,8 @@
 from synapse.util import Clock
 from synapse.util.stringutils import random_string
 
+logger = logging.getLogger(__name__)
+
 
 @attr.s(slots=True, cmp=False, frozen=True)
 class EventBuilder:
@@ -101,6 +104,7 @@ async def build(
         self,
         prev_event_ids: List[str],
         auth_event_ids: Optional[List[str]],
+        inherit_depth: bool,
     ) -> EventBase:
         """Transform into a fully signed and hashed event
 
@@ -109,6 +113,8 @@ async def build(
             auth_event_ids: The event IDs to use as the auth events.
                 Should normally be set to None, which will cause them to be calculated
                 based on the room state at the prev_events.
+            inherit_depth: True to use the oldest depth from the prev_event_ids
+                (instead of calculating a new depth).
 
         Returns:
             The signed and hashed event.
@@ -132,8 +138,61 @@ async def build(
             auth_events = auth_event_ids
             prev_events = prev_event_ids
 
-        old_depth = await self._store.get_max_depth_of(prev_event_ids)
-        depth = old_depth + 1
+        logger.info(
+            "prev_events %s",
+            prev_events,
+        )
+
+        (
+            most_recent_prev_event_id,
+            most_recent_prev_event_depth,
+        ) = await self._store.get_max_depth_of(prev_event_ids)
+        logger.info(
+            "event_id_with_max_depth %s %s",
+            most_recent_prev_event_id,
+            most_recent_prev_event_depth,
+        )
+
+        # We want to insert the historical event after the `prev_event` but before the successor event
+        #
+        # We inherit depth from the successor event instead of the `prev_event`
+        # because events returned from `/messages` are first sorted by `topological_ordering`
+        # which is just the `depth` and then tie-break with `stream_ordering`.
+        #
+        # We mark these inserted historical events as "backfilled" which gives them a
+        # negative `stream_ordering`. If we use the same depth as the `prev_event`,
+        # then our historical event will tie-break and be sorted before the `prev_event`
+        # when it should come after.
+        #
+        # We want to use the successor event depth so they appear after `prev_event` because
+        # it has a larger `depth` but before the successor event because the `stream_ordering`
+        # is negative before the successor event.
+        if inherit_depth:
+            sucessor_event_ids = await self._store.get_successor_events(
+                [most_recent_prev_event_id]
+            )
+            logger.info("sucessor_event_ids %s", sucessor_event_ids)
+
+            # If we can't find any successor events, then it's a forward extremity of
+            # historical messages nd we can just inherit from the previous historical
+            # event which we can already assume has the correct depth where we want
+            # to insert into.
+            if not sucessor_event_ids:
+                depth = most_recent_prev_event_depth
+            else:
+                (
+                    _,
+                    oldest_successor_depth,
+                ) = await self._store.get_min_depth_of(sucessor_event_ids)
+
+                logger.info("oldest_successor_depth %s", oldest_successor_depth)
+
+                depth = oldest_successor_depth
+        # Otherwise, progress the depth as normal
+        else:
+            depth = most_recent_prev_event_depth + 1
+
+        logger.info("depth %s", depth)
 
         # we cap depth of generated events, to ensure that they are not
         # rejected by other servers (and so that they can be persisted in

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/message.py

@@ -16,7 +16,7 @@
 # limitations under the License.
 import logging
 import random
-from typing import TYPE_CHECKING, Dict, List, Optional, Tuple
+from typing import TYPE_CHECKING, Dict, Iterable, List, Optional, Tuple
 
 from canonicaljson import encode_canonical_json
 
@@ -452,8 +452,11 @@ async def create_event(
         event_dict: dict,
         txn_id: Optional[str] = None,
         prev_event_ids: Optional[List[str]] = None,
+        inherit_depth: bool = False,
         auth_event_ids: Optional[List[str]] = None,
+        state_for_events: Optional[Iterable[EventBase]] = None,
         require_consent: bool = True,
+        outlier: bool = False,
     ) -> Tuple[EventBase, EventContext]:
         """
         Given a dict from a client, create a new event.
@@ -473,6 +476,9 @@ async def create_event(
 
                 If None, they will be requested from the database.
 
+            inherit_depth: True to use the oldest depth from the prev_event_ids
+                (instead of calculating a new depth).
+
             auth_event_ids:
                 The event ids to use as the auth_events for the new event.
                 Should normally be left as None, which will cause them to be calculated
@@ -535,11 +541,20 @@ async def create_event(
         if txn_id is not None:
             builder.internal_metadata.txn_id = txn_id
 
+        logger.info("setting internal_metadata.outlier %s", outlier)
+        # Setting the outlier on the builder does not work to propogate it over to the event
+        # because `outlier` is not part of the `interal_metadata_dict` that is serialized later.
+        # Not sure why it is separate.
+        # See https://github.com/matrix-org/synapse/pull/9247#r607595779
+        builder.internal_metadata.outlier = outlier
+
         event, context = await self.create_new_client_event(
             builder=builder,
             requester=requester,
             prev_event_ids=prev_event_ids,
+            inherit_depth=inherit_depth,
             auth_event_ids=auth_event_ids,
+            state_for_events=state_for_events,
         )
 
         # In an ideal world we wouldn't need the second part of this condition. However,
@@ -696,9 +711,13 @@ async def create_and_send_nonmember_event(
         self,
         requester: Requester,
         event_dict: dict,
+        inherit_depth: bool = False,
+        auth_event_ids: Optional[List[str]] = None,
+        state_for_events: Optional[Iterable[EventBase]] = None,
         ratelimit: bool = True,
         txn_id: Optional[str] = None,
         ignore_shadow_ban: bool = False,
+        outlier: bool = False,
     ) -> Tuple[EventBase, int]:
         """
         Creates an event, then sends it.
@@ -708,6 +727,8 @@ async def create_and_send_nonmember_event(
         Args:
             requester: The requester sending the event.
             event_dict: An entire event.
+            inherit_depth: True to use the oldest depth from the event_dict["prev_events"]
+                (instead of calculating a new depth).
             ratelimit: Whether to rate limit this send.
             txn_id: The transaction ID.
             ignore_shadow_ban: True if shadow-banned users should be allowed to
@@ -750,8 +771,16 @@ async def create_and_send_nonmember_event(
                     assert event.internal_metadata.stream_ordering
                     return event, event.internal_metadata.stream_ordering
 
+            prev_events = event_dict.get("prev_events")
+
             event, context = await self.create_event(
-                requester, event_dict, txn_id=txn_id
+                requester,
+                event_dict,
+                txn_id=txn_id,
+                prev_event_ids=prev_events,
+                auth_event_ids=auth_event_ids,
+                inherit_depth=inherit_depth,
+                outlier=outlier,
             )
 
             assert self.hs.is_mine_id(event.sender), "User must be our own: %s" % (
@@ -782,7 +811,9 @@ async def create_new_client_event(
         builder: EventBuilder,
         requester: Optional[Requester] = None,
         prev_event_ids: Optional[List[str]] = None,
+        inherit_depth: bool = False,
         auth_event_ids: Optional[List[str]] = None,
+        state_for_events: Optional[Iterable[EventBase]] = None,
     ) -> Tuple[EventBase, EventContext]:
         """Create a new event for a local client
 
@@ -795,6 +826,9 @@ async def create_new_client_event(
 
                 If None, they will be requested from the database.
 
+            inherit_depth: True to use the oldest depth from the prev_event_ids
+                (instead of calculating a new depth).
+
             auth_event_ids:
                 The event ids to use as the auth_events for the new event.
                 Should normally be left as None, which will cause them to be calculated
@@ -823,9 +857,16 @@ async def create_new_client_event(
         ), "Attempting to create an event with no prev_events"
 
         event = await builder.build(
-            prev_event_ids=prev_event_ids, auth_event_ids=auth_event_ids
+            prev_event_ids=prev_event_ids,
+            auth_event_ids=auth_event_ids,
+            inherit_depth=inherit_depth,
         )
-        context = await self.state.compute_event_context(event)
+
+        # Pass on the outlier property from the builder to the event
+        if builder.internal_metadata.outlier:
+            event.internal_metadata.outlier = builder.internal_metadata.outlier
+
+        context = await self.state.compute_event_context(event, state_for_events)
         if requester:
             context.app_service = requester.app_service
 
@@ -1231,13 +1272,21 @@ async def persist_and_notify_client_event(
             if prev_state_ids:
                 raise AuthError(403, "Changing the room create event is forbidden")
 
+        backfilled = False
+        if event.content.get("m.historical", None):
+            backfilled = True
+
+        logger.info("backfilled %s", backfilled)
+
         # Note that this returns the event that was persisted, which may not be
         # the same as we passed in if it was deduplicated due transaction IDs.
         (
             event,
             event_pos,
             max_stream_token,
-        ) = await self.storage.persistence.persist_event(event, context=context)
+        ) = await self.storage.persistence.persist_event(
+            event, context=context, backfilled=backfilled
+        )
 
         if self._ephemeral_events_enabled:
             # If there's an expiry timestamp on the event, schedule its expiry.
@@ -1409,6 +1458,7 @@ async def _rebuild_event_after_third_party_rules(
         event = await builder.build(
             prev_event_ids=original_event.prev_event_ids(),
             auth_event_ids=original_event.auth_event_ids(),
+            inherit_depth=False,
         )
 
         # we rebuild the event context, to be on the safe side. If nothing else,

synapse/handlers/room_member.py

@@ -180,6 +180,8 @@ async def _local_membership_update(
         ratelimit: bool = True,
         content: Optional[dict] = None,
         require_consent: bool = True,
+        outlier: bool = False,
+        state_for_events: Optional[Iterable[EventBase]] = None,
     ) -> Tuple[str, int]:
         user_id = target.to_string()
 
@@ -218,6 +220,8 @@ async def _local_membership_update(
             txn_id=txn_id,
             prev_event_ids=prev_event_ids,
             require_consent=require_consent,
+            outlier=outlier,
+            state_for_events=state_for_events,
         )
 
         prev_state_ids = await context.get_prev_state_ids()
@@ -312,6 +316,8 @@ async def update_membership(
         ratelimit: bool = True,
         content: Optional[dict] = None,
         require_consent: bool = True,
+        outlier: bool = False,
+        state_for_events: Optional[Iterable[EventBase]] = None,
     ) -> Tuple[str, int]:
         """Update a user's membership in a room.
 
@@ -352,6 +358,8 @@ async def update_membership(
                 ratelimit=ratelimit,
                 content=content,
                 require_consent=require_consent,
+                outlier=outlier,
+                state_for_events=state_for_events,
             )
 
         return result
@@ -368,6 +376,8 @@ async def update_membership_locked(
         ratelimit: bool = True,
         content: Optional[dict] = None,
         require_consent: bool = True,
+        outlier: bool = False,
+        state_for_events: Optional[Iterable[EventBase]] = None,
     ) -> Tuple[str, int]:
         """Helper for update_membership.
 
@@ -623,6 +633,8 @@ async def update_membership_locked(
             prev_event_ids=latest_event_ids,
             content=content,
             require_consent=require_consent,
+            outlier=outlier,
+            state_for_events=state_for_events,
         )
 
     async def transfer_room_state_on_room_upgrade(