Merge branch 'develop' into madlittlemods/msc3575-sliding-sync-0.0.1

changelog.d/17164.bugfix

@@ -0,0 +1 @@
+Fix deduplicating of membership events to not create unused state groups.

changelog.d/17215.bugfix

@@ -0,0 +1 @@
+Fix bug where duplicate events could be sent down sync when using workers that are overloaded.

changelog.d/17229.misc

@@ -0,0 +1 @@
+Replaces all usages of `StreamIdGenerator` with `MultiWriterIdGenerator`.

changelog.d/17242.misc

@@ -0,0 +1 @@
+Clean out invalid destinations from `device_federation_outbox` table.

changelog.d/17246.misc

@@ -0,0 +1 @@
+Fix errors in logs about closing incorrect logging contexts when media gets rejected by a module.

pyproject.toml

@@ -200,10 +200,8 @@ netaddr = ">=0.7.18"
 # add a lower bound to the Jinja2 dependency.
 Jinja2 = ">=3.0"
 bleach = ">=1.4.3"
-# We use `ParamSpec` and `Concatenate`, which were added in `typing-extensions` 3.10.0.0.
-# Additionally we need https://github.com/python/typing/pull/817 to allow types to be
-# generic over ParamSpecs.
-typing-extensions = ">=3.10.0.1"
+# We use `Self`, which were added in `typing-extensions` 4.0.
+typing-extensions = ">=4.0"
 # We enforce that we have a `cryptography` version that bundles an `openssl`
 # with the latest security patches.
 cryptography = ">=3.4.7"

synapse/_scripts/synapse_port_db.py

@@ -777,22 +777,74 @@ def alter_table(txn: LoggingTransaction) -> None:
             await self._setup_events_stream_seqs()
             await self._setup_sequence(
                 "un_partial_stated_event_stream_sequence",
-                ("un_partial_stated_event_stream",),
+                [("un_partial_stated_event_stream", "stream_id")],
             )
             await self._setup_sequence(
-                "device_inbox_sequence", ("device_inbox", "device_federation_outbox")
+                "device_inbox_sequence",
+                [
+                    ("device_inbox", "stream_id"),
+                    ("device_federation_outbox", "stream_id"),
+                ],
             )
             await self._setup_sequence(
                 "account_data_sequence",
-                ("room_account_data", "room_tags_revisions", "account_data"),
+                [
+                    ("room_account_data", "stream_id"),
+                    ("room_tags_revisions", "stream_id"),
+                    ("account_data", "stream_id"),
+                ],
+            )
+            await self._setup_sequence(
+                "receipts_sequence",
+                [
+                    ("receipts_linearized", "stream_id"),
+                ],
+            )
+            await self._setup_sequence(
+                "presence_stream_sequence",
+                [
+                    ("presence_stream", "stream_id"),
+                ],
             )
-            await self._setup_sequence("receipts_sequence", ("receipts_linearized",))
-            await self._setup_sequence("presence_stream_sequence", ("presence_stream",))
             await self._setup_auth_chain_sequence()
             await self._setup_sequence(
                 "application_services_txn_id_seq",
-                ("application_services_txns",),
-                "txn_id",
+                [
+                    (
+                        "application_services_txns",
+                        "txn_id",
+                    )
+                ],
+            )
+            await self._setup_sequence(
+                "device_lists_sequence",
+                [
+                    ("device_lists_stream", "stream_id"),
+                    ("user_signature_stream", "stream_id"),
+                    ("device_lists_outbound_pokes", "stream_id"),
+                    ("device_lists_changes_in_room", "stream_id"),
+                    ("device_lists_remote_pending", "stream_id"),
+                    ("device_lists_changes_converted_stream_position", "stream_id"),
+                ],
+            )
+            await self._setup_sequence(
+                "e2e_cross_signing_keys_sequence",
+                [
+                    ("e2e_cross_signing_keys", "stream_id"),
+                ],
+            )
+            await self._setup_sequence(
+                "push_rules_stream_sequence",
+                [
+                    ("push_rules_stream", "stream_id"),
+                ],
+            )
+            await self._setup_sequence(
+                "pushers_sequence",
+                [
+                    ("pushers", "id"),
+                    ("deleted_pushers", "stream_id"),
+                ],
             )
 
             # Step 3. Get tables.
@@ -1101,12 +1153,11 @@ def _setup_events_stream_seqs_set_pos(txn: LoggingTransaction) -> None:
     async def _setup_sequence(
         self,
         sequence_name: str,
-        stream_id_tables: Iterable[str],
-        column_name: str = "stream_id",
+        stream_id_tables: Iterable[Tuple[str, str]],
     ) -> None:
         """Set a sequence to the correct value."""
         current_stream_ids = []
-        for stream_id_table in stream_id_tables:
+        for stream_id_table, column_name in stream_id_tables:
             max_stream_id = cast(
                 int,
                 await self.sqlite_store.db_pool.simple_select_one_onecol(

synapse/handlers/message.py

@@ -496,13 +496,6 @@ def __init__(self, hs: "HomeServer"):
 
         self.room_prejoin_state_types = self.hs.config.api.room_prejoin_state
 
-        self.membership_types_to_include_profile_data_in = {
-            Membership.JOIN,
-            Membership.KNOCK,
-        }
-        if self.hs.config.server.include_profile_data_on_invite:
-            self.membership_types_to_include_profile_data_in.add(Membership.INVITE)
-
         self.send_event = ReplicationSendEventRestServlet.make_client(hs)
         self.send_events = ReplicationSendEventsRestServlet.make_client(hs)
 
@@ -594,8 +587,6 @@ async def create_event(
         Creates an FrozenEvent object, filling out auth_events, prev_events,
         etc.
 
-        Adds display names to Join membership events.
-
         Args:
             requester
             event_dict: An entire event
@@ -672,29 +663,6 @@ async def create_event(
 
         self.validator.validate_builder(builder)
 
-        if builder.type == EventTypes.Member:
-            membership = builder.content.get("membership", None)
-            target = UserID.from_string(builder.state_key)
-
-            if membership in self.membership_types_to_include_profile_data_in:
-                # If event doesn't include a display name, add one.
-                profile = self.profile_handler
-                content = builder.content
-
-                try:
-                    if "displayname" not in content:
-                        displayname = await profile.get_displayname(target)
-                        if displayname is not None:
-                            content["displayname"] = displayname
-                    if "avatar_url" not in content:
-                        avatar_url = await profile.get_avatar_url(target)
-                        if avatar_url is not None:
-                            content["avatar_url"] = avatar_url
-                except Exception as e:
-                    logger.info(
-                        "Failed to get profile information for %r: %s", target, e
-                    )
-
         is_exempt = await self._is_exempt_from_privacy_policy(builder, requester)
         if require_consent and not is_exempt:
             await self.assert_accepted_privacy_policy(requester)

synapse/handlers/room_member.py

@@ -106,6 +106,13 @@ def __init__(self, hs: "HomeServer"):
         self.event_auth_handler = hs.get_event_auth_handler()
         self._worker_lock_handler = hs.get_worker_locks_handler()
 
+        self._membership_types_to_include_profile_data_in = {
+            Membership.JOIN,
+            Membership.KNOCK,
+        }
+        if self.hs.config.server.include_profile_data_on_invite:
+            self._membership_types_to_include_profile_data_in.add(Membership.INVITE)
+
         self.member_linearizer: Linearizer = Linearizer(name="member")
         self.member_as_limiter = Linearizer(max_count=10, name="member_as_limiter")
 
@@ -785,9 +792,8 @@ async def update_membership_locked(
         if (
             not self.allow_per_room_profiles and not is_requester_server_notices_user
         ) or requester.shadow_banned:
-            # Strip profile data, knowing that new profile data will be added to the
-            # event's content in event_creation_handler.create_event() using the target's
-            # global profile.
+            # Strip profile data, knowing that new profile data will be added to
+            # the event's content below using the target's global profile.
             content.pop("displayname", None)
             content.pop("avatar_url", None)
 
@@ -823,6 +829,29 @@ async def update_membership_locked(
         if action in ["kick", "unban"]:
             effective_membership_state = "leave"
 
+        if effective_membership_state not in Membership.LIST:
+            raise SynapseError(400, "Invalid membership key")
+
+        # Add profile data for joins etc, if no per-room profile.
+        if (
+            effective_membership_state
+            in self._membership_types_to_include_profile_data_in
+        ):
+            # If event doesn't include a display name, add one.
+            profile = self.profile_handler
+
+            try:
+                if "displayname" not in content:
+                    displayname = await profile.get_displayname(target)
+                    if displayname is not None:
+                        content["displayname"] = displayname
+                if "avatar_url" not in content:
+                    avatar_url = await profile.get_avatar_url(target)
+                    if avatar_url is not None:
+                        content["avatar_url"] = avatar_url
+            except Exception as e:
+                logger.info("Failed to get profile information for %r: %s", target, e)
+
         # if this is a join with a 3pid signature, we may need to turn a 3pid
         # invite into a normal invite before we can handle the join.
         if third_party_signed is not None:

synapse/handlers/sync.py

@@ -284,6 +284,23 @@ def __bool__(self) -> bool:
             or self.device_lists
         )
 
+    @staticmethod
+    def empty(next_batch: StreamToken) -> "SyncResult":
+        "Return a new empty result"
+        return SyncResult(
+            next_batch=next_batch,
+            presence=[],
+            account_data=[],
+            joined=[],
+            invited=[],
+            knocked=[],
+            archived=[],
+            to_device=[],
+            device_lists=DeviceListUpdates(),
+            device_one_time_keys_count={},
+            device_unused_fallback_key_types=[],
+        )
+
 
 @attr.s(slots=True, frozen=True, auto_attribs=True)
 class E2eeSyncResult:
@@ -497,6 +514,24 @@ async def _wait_for_sync_for_user(
         if context:
             context.tag = sync_label
 
+        if since_token is not None:
+            # We need to make sure this worker has caught up with the token. If
+            # this returns false it means we timed out waiting, and we should
+            # just return an empty response.
+            start = self.clock.time_msec()
+            if not await self.notifier.wait_for_stream_token(since_token):
+                logger.warning(
+                    "Timed out waiting for worker to catch up. Returning empty response"
+                )
+                return SyncResult.empty(since_token)
+
+            # If we've spent significant time waiting to catch up, take it off
+            # the timeout.
+            now = self.clock.time_msec()
+            if now - start > 1_000:
+                timeout -= now - start
+                timeout = max(timeout, 0)
+
         # if we have a since token, delete any to-device messages before that token
         # (since we now know that the device has received them)
         if since_token is not None:

synapse/media/media_repository.py

@@ -1049,6 +1049,11 @@ async def _generate_thumbnails(
                     finally:
                         t_byte_source.close()
 
+                    # We flush and close the file to ensure that the bytes have
+                    # been written before getting the size.
+                    f.flush()
+                    f.close()
+
                     t_len = os.path.getsize(fname)
 
                     # Write to database

synapse/media/media_storage.py

@@ -137,42 +137,37 @@ async def store_into_file(
         dirname = os.path.dirname(fname)
         os.makedirs(dirname, exist_ok=True)
 
-        main_media_repo_write_trace_scope = start_active_span(
-            "writing to main media repo"
-        )
-        main_media_repo_write_trace_scope.__enter__()
-
-        with main_media_repo_write_trace_scope:
-            try:
+        try:
+            with start_active_span("writing to main media repo"):
                 with open(fname, "wb") as f:
                     yield f, fname
 
-            except Exception as e:
-                try:
-                    os.remove(fname)
-                except Exception:
-                    pass
-
-                raise e from None
-
-        with start_active_span("writing to other storage providers"):
-            spam_check = (
-                await self._spam_checker_module_callbacks.check_media_file_for_spam(
-                    ReadableFileWrapper(self.clock, fname), file_info
+            with start_active_span("writing to other storage providers"):
+                spam_check = (
+                    await self._spam_checker_module_callbacks.check_media_file_for_spam(
+                        ReadableFileWrapper(self.clock, fname), file_info
+                    )
                 )
-            )
-            if spam_check != self._spam_checker_module_callbacks.NOT_SPAM:
-                logger.info("Blocking media due to spam checker")
-                # Note that we'll delete the stored media, due to the
-                # try/except below. The media also won't be stored in
-                # the DB.
-                # We currently ignore any additional field returned by
-                # the spam-check API.
-                raise SpamMediaException(errcode=spam_check[0])
-
-            for provider in self.storage_providers:
-                with start_active_span(str(provider)):
-                    await provider.store_file(path, file_info)
+                if spam_check != self._spam_checker_module_callbacks.NOT_SPAM:
+                    logger.info("Blocking media due to spam checker")
+                    # Note that we'll delete the stored media, due to the
+                    # try/except below. The media also won't be stored in
+                    # the DB.
+                    # We currently ignore any additional field returned by
+                    # the spam-check API.
+                    raise SpamMediaException(errcode=spam_check[0])
+
+                for provider in self.storage_providers:
+                    with start_active_span(str(provider)):
+                        await provider.store_file(path, file_info)
+
+        except Exception as e:
+            try:
+                os.remove(fname)
+            except Exception:
+                pass
+
+            raise e from None
 
     async def fetch_media(self, file_info: FileInfo) -> Optional[Responder]:
         """Attempts to fetch media described by file_info from the local cache

synapse/notifier.py

@@ -763,6 +763,29 @@ async def check_for_updates(
 
         return result
 
+    async def wait_for_stream_token(self, stream_token: StreamToken) -> bool:
+        """Wait for this worker to catch up with the given stream token."""
+
+        start = self.clock.time_msec()
+        while True:
+            current_token = self.event_sources.get_current_token()
+            if stream_token.is_before_or_eq(current_token):
+                return True
+
+            now = self.clock.time_msec()
+
+            if now - start > 10_000:
+                return False
+
+            logger.info(
+                "Waiting for current token to reach %s; currently at %s",
+                stream_token,
+                current_token,
+            )
+
+            # TODO: be better
+            await self.clock.sleep(0.5)
+
     async def _get_room_ids(
         self, user: UserID, explicit_room_id: Optional[str]
     ) -> Tuple[StrCollection, bool]: