Move reading of multipart response into try body (#19062)

Author: anoadragon453

changelog.d/19062.bugfix

@@ -0,0 +1 @@
+Fix a bug introduced in 1.111.0 where failed attempts to download authenticated remote media would not be handled correctly.

synapse/http/matrixfederationclient.py

@@ -1755,6 +1755,7 @@ async def federation_get_file(
                     response, output_stream, boundary, expected_size + 1
                 )
                 deferred.addTimeout(self.default_timeout_seconds, self.reactor)
+                multipart_response = await make_deferred_yieldable(deferred)
         except BodyExceededMaxSize:
             msg = "Requested file is too large > %r bytes" % (expected_size,)
             logger.warning(
@@ -1791,7 +1792,6 @@ async def federation_get_file(
             )
             raise
 
-        multipart_response = await make_deferred_yieldable(deferred)
         if not multipart_response.url:
             assert multipart_response.length is not None
             length = multipart_response.length

tests/http/test_matrixfederationclient.py

@@ -414,6 +414,65 @@ def test_authed_media_redirect_response(self) -> None:
         self.assertEqual(length, len(data))
         self.assertEqual(output_stream.getvalue(), data)
 
+    @override_config(
+        {
+            "federation": {
+                # Set the timeout to a deterministic value, in case the defaults
+                # change.
+                "client_timeout": "10s",
+            }
+        }
+    )
+    def test_authed_media_timeout_reading_body(self) -> None:
+        """
+        If the HTTP request is connected, but gets no response before being
+        timed out, it'll give a RequestSendFailed with can_retry.
+
+        Regression test for https://github.com/element-hq/synapse/issues/19061
+        """
+        limiter = Ratelimiter(
+            store=self.hs.get_datastores().main,
+            clock=self.clock,
+            cfg=RatelimitSettings(key="", per_second=0.17, burst_count=1048576),
+        )
+
+        output_stream = io.BytesIO()
+
+        d = defer.ensureDeferred(
+            # timeout is set by `client_timeout`, which we override above.
+            self.cl.federation_get_file(
+                "testserv:8008", "path", output_stream, limiter, "127.0.0.1", 10000
+            )
+        )
+
+        self.pump()
+
+        conn = Mock()
+        clients = self.reactor.tcpClients
+        client = clients[0][2].buildProtocol(None)
+        client.makeConnection(conn)
+
+        # Deferred does not have a result
+        self.assertNoResult(d)
+
+        # Send it the HTTP response
+        client.dataReceived(
+            b"HTTP/1.1 200 OK\r\n"
+            b"Server: Fake\r\n"
+            # Set a large content length, prompting the federation client to
+            # wait to receive the rest of the body.
+            b"Content-Length: 1000\r\n"
+            b"Content-Type: multipart/mixed; boundary=6067d4698f8d40a0a794ea7d7379d53a\r\n\r\n"
+        )
+
+        # Push by enough to time it out
+        self.reactor.advance(10.5)
+        f = self.failureResultOf(d)
+
+        self.assertIsInstance(f.value, RequestSendFailed)
+        self.assertTrue(f.value.can_retry)
+        self.assertIsInstance(f.value.inner_exception, defer.TimeoutError)
+
     @parameterized.expand(["get_json", "post_json", "delete_json", "put_json"])
     def test_timeout_reading_body(self, method_name: str) -> None:
         """