Send to-device messages to application services

synapse/appservice/scheduler.py

@@ -52,8 +52,8 @@
     TYPE_CHECKING,
     Awaitable,
     Callable,
+    Collection,
     Dict,
-    Iterable,
     List,
     Optional,
     Set,
@@ -112,12 +112,13 @@ async def start(self) -> None:
     def enqueue_for_appservice(
         self,
         appservice: ApplicationService,
-        events: Optional[Iterable[EventBase]] = None,
-        ephemeral: Optional[Iterable[JsonDict]] = None,
-        to_device_messages: Optional[Iterable[JsonDict]] = None,
+        events: Optional[Collection[EventBase]] = None,
+        ephemeral: Optional[Collection[JsonDict]] = None,
+        to_device_messages: Optional[Collection[JsonDict]] = None,
     ) -> None:
         """
         Enqueue some data to be sent off to an application service.
+
         Args:
             appservice: The application service to create and send a transaction to.
             events: The persistent room events to send.
@@ -127,7 +128,7 @@ def enqueue_for_appservice(
                 'to_user_id' fields.
         """
         # We purposefully allow this method to run with empty events/ephemeral
-        # iterables, so that callers do not need to check iterable size themselves.
+        # collections, so that callers do not need to check iterable size themselves.
         if not events and not ephemeral and not to_device_messages:
             return
 

synapse/config/experimental.py

@@ -57,8 +57,7 @@ def read_config(self, config: JsonDict, **kwargs):
 
         # MSC2409 (this setting only relates to optionally sending to-device messages).
         # Presence, typing and read receipt EDUs are already sent to application services that
-        # have opted in to receive them. This setting, if enabled, adds to-device messages
-        # to that list.
+        # have opted in to receive them. If enabled, this adds to-device messages to that list.
         self.msc2409_to_device_messages_enabled: bool = experimental.get(
             "msc2409_to_device_messages_enabled", False
         )

synapse/handlers/appservice.py

@@ -477,16 +477,17 @@ async def _get_to_device_messages(
     ) -> List[JsonDict]:
         """
         Given an application service, determine which events it should receive
-        from those between the last-recorded typing event stream token for this
+        from those between the last-recorded to-device message stream token for this
         appservice and the given stream token.
 
         Args:
             service: The application service to check for which events it should receive.
             new_token: The latest to-device event stream token.
-            users: The users that should receive new to-device messages.
+            users: The users to be notified for the new to-device messages
+                (ie, the recipients of the messages).
 
         Returns:
-            A list of JSON dictionaries containing data derived from the typing events
+            A list of JSON dictionaries containing data derived from the to-device events
                 that should be sent to the given application service.
         """
         # Get the stream token that this application service has processed up until

synapse/notifier.py

@@ -462,7 +462,7 @@ def on_new_event(
                 )
             except Exception:
                 logger.exception(
-                    "Error notifying application services of ephemeral event"
+                    "Error notifying application services of ephemeral events"
                 )
 
     def on_new_replication_data(self) -> None:

synapse/storage/databases/main/deviceinbox.py

@@ -157,14 +157,21 @@ async def get_messages_for_user_devices(
         Returns:
             A dictionary of (user id, device id) -> list of to-device messages.
         """
-        # We expect the stream ID returned by _get_new_device_messages to always
-        # return to_stream_id. So, no need to return it from this function.
-        user_id_device_id_to_messages, _ = await self._get_device_messages(
+        # We expect the stream ID returned by _get_device_messages to always
+        # be to_stream_id. So, no need to return it from this function.
+        (
+            user_id_device_id_to_messages,
+            last_processed_stream_id,
+        ) = await self._get_device_messages(
             user_ids=user_ids,
             from_stream_id=from_stream_id,
             to_stream_id=to_stream_id,
         )
 
+        assert (
+            last_processed_stream_id == to_stream_id
+        ), "Expected _get_device_messages to process all to-device messages up to `to_stream_id`"
+
         return user_id_device_id_to_messages
 
     async def get_messages_for_device(
@@ -190,7 +197,8 @@ async def get_messages_for_device(
 
         Returns:
             A tuple containing:
-                * A dictionary of (user id, device id) -> list of to-device messages.
+                * A list of to-device messages within the given stream id range intended for
+                  the given user / device combo.
                 * The last-processed stream ID. Subsequent calls of this function with the
                   same device should pass this value as 'from_stream_id'.
         """
@@ -199,7 +207,7 @@ async def get_messages_for_device(
             last_processed_stream_id,
         ) = await self._get_device_messages(
             user_ids=[user_id],
-            device_ids=[device_id],
+            device_id=device_id,
             from_stream_id=from_stream_id,
             to_stream_id=to_stream_id,
             limit=limit,
@@ -210,7 +218,7 @@ async def get_messages_for_device(
             return [], to_stream_id
 
         # Extract the messages, no need to return the user and device ID again
-        to_device_messages = list(user_id_device_id_to_messages.values())[0]
+        to_device_messages = user_id_device_id_to_messages.get((user_id, device_id), [])
 
         return to_device_messages, last_processed_stream_id
 
@@ -219,7 +227,7 @@ async def _get_device_messages(
         user_ids: Collection[str],
         from_stream_id: int,
         to_stream_id: int,
-        device_ids: Optional[Collection[str]] = None,
+        device_id: Optional[str] = None,
         limit: Optional[int] = None,
     ) -> Tuple[Dict[Tuple[str, str], List[JsonDict]], int]:
         """
@@ -230,7 +238,7 @@ async def _get_device_messages(
 
         Note that a stream ID can be shared by multiple copies of the same message with
         different recipient devices. Stream IDs are only unique in the context of a single
-        user ID / device ID pair Thus, applying a limit (of messages to return) when working
+        user ID / device ID pair. Thus, applying a limit (of messages to return) when working
         with a sliding window of stream IDs is only possible when querying messages of a
         single user device.
 
@@ -240,8 +248,9 @@ async def _get_device_messages(
             user_ids: The user IDs to filter device messages by.
             from_stream_id: The lower boundary of stream id to filter with (exclusive).
             to_stream_id: The upper boundary of stream id to filter with (inclusive).
-            device_ids: If provided, only messages destined for these device IDs will be returned.
-                If not provided, all device IDs for the given user IDs will be used.
+            device_id: A device ID to query to-device messages for. If not provided, to-device
+                messages from all device IDs for the given user IDs will be queried. May not be
+                provided if `user_ids` contains more than one entry.
             limit: The maximum number of to-device messages to return. Can only be used when
                 passing a single user ID / device ID tuple.
 
@@ -252,28 +261,32 @@ async def _get_device_messages(
                     there may be more messages to retrieve. If `limit` is not set, then this
                     is always equal to 'to_stream_id'.
         """
-        # A limit can only be applied when querying for a single user ID / device ID tuple.
-        if limit:
-            if not device_ids:
+        if not user_ids:
+            logger.warning("No users provided upon querying for device IDs")
+            return {}, to_stream_id
+
+        if len(user_ids) > 1:
+            if device_id is not None:
                 raise AssertionError(
-                    "Programming error: _get_new_device_messages was passed 'limit' "
-                    "but not device_ids. This could lead to querying multiple user ID "
-                    "/ device ID pairs, which is not compatible with 'limit'"
+                    "Programming error: 'device_id' cannot be supplied to "
+                    "_get_device_messages when >1 user_id has been provided"
                 )
 
-            if len(user_ids) > 1 or len(device_ids) > 1:
+            # A limit can only be applied when querying for a single user ID / device ID tuple.
+            if limit:
                 raise AssertionError(
-                    "Programming error: _get_new_device_messages was passed 'limit' "
-                    "with >1 user id/device id pair"
+                    "Programming error: _get_device_messages was passed 'limit' "
+                    "with >1 user_id"
                 )
 
         user_ids_to_query: Set[str] = set()
         device_ids_to_query: Set[str] = set()
 
-        if device_ids is not None:
-            # If a collection of device IDs were passed, use them to filter results.
+        # Note that a device ID could be an empty str
+        if device_id is not None:
+            # If a device ID was passed, use it to filter results.
             # Otherwise, device IDs will be derived from the given collection of user IDs.
-            device_ids_to_query.update(device_ids)
+            device_ids_to_query.add(device_id)
 
         # Determine which users have devices with pending messages
         for user_id in user_ids:
@@ -290,7 +303,7 @@ def get_device_messages_txn(txn: LoggingTransaction):
             # If a list of device IDs was not provided, retrieve all devices IDs
             # for the given users. We explicitly do not query hidden devices, as
             # hidden devices should not receive to-device messages.
-            if not device_ids:
+            if not device_ids_to_query:
                 user_device_dicts = self.db_pool.simple_select_many_txn(
                     txn,
                     table="devices",
@@ -304,7 +317,7 @@ def get_device_messages_txn(txn: LoggingTransaction):
                     {row["device_id"] for row in user_device_dicts}
                 )
 
-            if not user_ids_to_query or not device_ids_to_query:
+            if not device_ids_to_query:
                 # We've ended up with no devices to query.
                 return {}, to_stream_id
 
@@ -335,16 +348,17 @@ def get_device_messages_txn(txn: LoggingTransaction):
             )
 
             # If a limit was provided, limit the data retrieved from the database
-            if limit:
+            if limit is not None:
                 sql += "LIMIT ?"
                 sql_args += (limit,)
 
             txn.execute(sql, sql_args)
 
             # Create and fill a dictionary of (user ID, device ID) -> list of messages
             # intended for each device.
+            last_processed_stream_pos = to_stream_id
             recipient_device_to_messages: Dict[Tuple[str, str], List[JsonDict]] = {}
-            for message_count, row in enumerate(txn, start=1):
+            for row in txn:
                 last_processed_stream_pos = row[0]
                 recipient_user_id = row[1]
                 recipient_device_id = row[2]
@@ -355,13 +369,14 @@ def get_device_messages_txn(txn: LoggingTransaction):
                     (recipient_user_id, recipient_device_id), []
                 ).append(message_dict)
 
-                if limit and message_count == limit:
-                    # We ended up hitting the message limit. There may be more messages to retrieve.
-                    # Return what we have, as well as the last stream position that was processed.
-                    #
-                    # The caller is expected to set this as the lower (exclusive) bound
-                    # for the next query of this device.
-                    return recipient_device_to_messages, last_processed_stream_pos
+            if limit is not None and txn.rowcount == limit:
+                # We ended up bumping up against the message limit. There may be more messages
+                # to retrieve. Return what we have, as well as the last stream position that
+                # was processed.
+                #
+                # The caller is expected to set this as the lower (exclusive) bound
+                # for the next query of this device.
+                return recipient_device_to_messages, last_processed_stream_pos
 
             # The limit was not reached, thus we know that recipient_device_to_messages
             # contains all to-device messages for the given device and stream id range.

synapse/storage/schema/main/delta/65/06_msc2409_add_device_id_appservice_stream_type.sql → synapse/storage/schema/main/delta/68/02_msc2409_add_device_id_appservice_stream_type.sql

@@ -1,4 +1,4 @@
-/* Copyright 2021 The Matrix.org Foundation C.I.C
+/* Copyright 2022 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.