Ensure that incoming to-device messages are not dropped (#17127)

... when workers are unreachable, etc.

Fixes #17117.

The general principle is just to make sure that we propagate any
exceptions to the JsonResource, so that we return an error code to the
sending server. That means that the sending server no longer considers
the message safely sent, so it will retry later.

In the issue, Erik mentions that an alternative solution would be to
persist the to-device messages into a table so that they can be retried.
This might be an improvement for performance, but even if we did that,
we still need this mechanism, since we might be unable to reach the
database. So, if we want to do that, it can be a later follow-up.

---------

Co-authored-by: Erik Johnston <erik@matrix.org>

changelog.d/17127.bugfix

@@ -0,0 +1 @@
+Fix a bug which meant that to-device messages received over federation could be dropped when the server was under load or networking problems caused problems between Synapse processes or the database.

synapse/federation/federation_server.py

@@ -546,7 +546,25 @@ async def _process_edu(edu_dict: JsonDict) -> None:
                 edu_type=edu_dict["edu_type"],
                 content=edu_dict["content"],
             )
-            await self.registry.on_edu(edu.edu_type, origin, edu.content)
+            try:
+                await self.registry.on_edu(edu.edu_type, origin, edu.content)
+            except Exception:
+                # If there was an error handling the EDU, we must reject the
+                # transaction.
+                #
+                # Some EDU types (notably, to-device messages) are, despite their name,
+                # expected to be reliable; if we weren't able to do something with it,
+                # we have to tell the sender that, and the only way the protocol gives
+                # us to do so is by sending an HTTP error back on the transaction.
+                #
+                # We log the exception now, and then raise a new SynapseError to cause
+                # the transaction to be failed.
+                logger.exception("Error handling EDU of type %s", edu.edu_type)
+                raise SynapseError(500, f"Error handing EDU of type {edu.edu_type}")
+
+                # TODO: if the first EDU fails, we should probably abort the whole
+                #   thing rather than carrying on with the rest of them. That would
+                #   probably be best done inside `concurrently_execute`.
 
         await concurrently_execute(
             _process_edu,
@@ -1414,12 +1432,7 @@ async def on_edu(self, edu_type: str, origin: str, content: dict) -> None:
         handler = self.edu_handlers.get(edu_type)
         if handler:
             with start_active_span_from_edu(content, "handle_edu"):
-                try:
-                    await handler(origin, content)
-                except SynapseError as e:
-                    logger.info("Failed to handle edu %r: %r", edu_type, e)
-                except Exception:
-                    logger.exception("Failed to handle edu %r", edu_type)
+                await handler(origin, content)
             return
 
         # Check if we can route it somewhere else that isn't us
@@ -1428,17 +1441,12 @@ async def on_edu(self, edu_type: str, origin: str, content: dict) -> None:
             # Pick an instance randomly so that we don't overload one.
             route_to = random.choice(instances)
 
-            try:
-                await self._send_edu(
-                    instance_name=route_to,
-                    edu_type=edu_type,
-                    origin=origin,
-                    content=content,
-                )
-            except SynapseError as e:
-                logger.info("Failed to handle edu %r: %r", edu_type, e)
-            except Exception:
-                logger.exception("Failed to handle edu %r", edu_type)
+            await self._send_edu(
+                instance_name=route_to,
+                edu_type=edu_type,
+                origin=origin,
+                content=content,
+            )
             return
 
         # Oh well, let's just log and move on.

synapse/handlers/devicemessage.py

@@ -104,6 +104,9 @@ async def on_direct_to_device_edu(self, origin: str, content: JsonDict) -> None:
         """
         Handle receiving to-device messages from remote homeservers.
 
+        Note that any errors thrown from this method will cause the federation /send
+        request to receive an error response.
+
         Args:
             origin: The remote homeserver.
             content: The JSON dictionary containing the to-device messages.

tests/federation/test_federation_server.py

@@ -67,6 +67,23 @@ def test_bad_request(self, query_content: bytes) -> None:
         self.assertEqual(HTTPStatus.BAD_REQUEST, channel.code, channel.result)
         self.assertEqual(channel.json_body["errcode"], "M_NOT_JSON")
 
+    def test_failed_edu_causes_500(self) -> None:
+        """If the EDU handler fails, /send should return a 500."""
+
+        async def failing_handler(_origin: str, _content: JsonDict) -> None:
+            raise Exception("bleh")
+
+        self.hs.get_federation_registry().register_edu_handler(
+            "FAIL_EDU_TYPE", failing_handler
+        )
+
+        channel = self.make_signed_federation_request(
+            "PUT",
+            "/_matrix/federation/v1/send/txn",
+            {"edus": [{"edu_type": "FAIL_EDU_TYPE", "content": {}}]},
+        )
+        self.assertEqual(500, channel.code, channel.result)
+
 
 class ServerACLsTestCase(unittest.TestCase):
     def test_blocked_server(self) -> None:

tests/federation/transport/test_server.py

@@ -59,7 +59,14 @@ def test_edu_debugging_doesnt_explode(self) -> None:
             "/_matrix/federation/v1/send/txn_id_1234/",
             content={
                 "edus": [
-                    {"edu_type": EduTypes.DEVICE_LIST_UPDATE, "content": {"foo": "bar"}}
+                    {
+                        "edu_type": EduTypes.DEVICE_LIST_UPDATE,
+                        "content": {
+                            "device_id": "QBUAZIFURK",
+                            "stream_id": 0,
+                            "user_id": "@user:id",
+                        },
+                    },
                 ],
                 "pdus": [],
             },