Wait for response

Spawning from https://github.com/element-hq/synapse/pull/18804#discussion_r2272826978
2025-12-05 01:10:13 +00:00 · 2025-08-20 16:39:55 -05:00
10 changed files with 35 additions and 35 deletions
--- a/synapse/http/proxy.py
+++ b/synapse/http/proxy.py
@@ -173,7 +173,7 @@ class ProxyResource(_AsyncResource):

        return response.code, response

-    def _send_response(
+    async def _send_response(
        self,
        request: "SynapseRequest",
        code: int,
@@ -205,7 +205,7 @@ class ProxyResource(_AsyncResource):

        response.deliverBody(_ProxyResponseBody(request))

-    def _send_error_response(
+    async def _send_error_response(
        self,
        f: failure.Failure,
        request: "SynapseRequest",
--- a/synapse/http/server.py
+++ b/synapse/http/server.py
@@ -75,7 +75,7 @@ from synapse.api.errors import (
    UnrecognizedRequestError,
 )
 from synapse.config.homeserver import HomeServerConfig
-from synapse.logging.context import defer_to_thread, preserve_fn, run_in_background
+from synapse.logging.context import defer_to_thread, preserve_fn
 from synapse.logging.opentracing import active_span, start_active_span, trace_servlet
 from synapse.util import Clock, json_encoder
 from synapse.util.caches import intern_dict
@@ -111,7 +111,7 @@ HTML_ERROR_TEMPLATE = """<!DOCTYPE html>
 HTTP_STATUS_REQUEST_CANCELLED = 499


-def return_json_error(
+async def return_json_error(
    f: failure.Failure, request: "SynapseRequest", config: Optional[HomeServerConfig]
 ) -> None:
    """Sends a JSON error response to clients."""
@@ -163,7 +163,7 @@ def return_json_error(
                # abortConnection throws if the connection is already closed
                pass
    else:
-        respond_with_json(
+        await respond_with_json(
            request,
            error_code,
            error_dict,
@@ -342,13 +342,13 @@ class _AsyncResource(resource.Resource, metaclass=abc.ABCMeta):

                if callback_return is not None:
                    code, response = callback_return
-                    self._send_response(request, code, response)
+                    await self._send_response(request, code, response)
        except Exception:
            # failure.Failure() fishes the original Failure out
            # of our stack, and thus gives us a sensible stack
            # trace.
            f = failure.Failure()
-            self._send_error_response(f, request)
+            await self._send_error_response(f, request)

    async def _async_render(
        self, request: "SynapseRequest"
@@ -380,7 +380,7 @@ class _AsyncResource(resource.Resource, metaclass=abc.ABCMeta):
        raise UnrecognizedRequestError(code=405)

    @abc.abstractmethod
-    def _send_response(
+    async def _send_response(
        self,
        request: "SynapseRequest",
        code: int,
@@ -389,7 +389,7 @@ class _AsyncResource(resource.Resource, metaclass=abc.ABCMeta):
        raise NotImplementedError()

    @abc.abstractmethod
-    def _send_error_response(
+    async def _send_error_response(
        self,
        f: failure.Failure,
        request: "SynapseRequest",
@@ -415,7 +415,7 @@ class DirectServeJsonResource(_AsyncResource):
        super().__init__(clock, extract_context)
        self.canonical_json = canonical_json

-    def _send_response(
+    async def _send_response(
        self,
        request: "SynapseRequest",
        code: int,
@@ -423,7 +423,7 @@ class DirectServeJsonResource(_AsyncResource):
    ) -> None:
        """Implements _AsyncResource._send_response"""
        # TODO: Only enable CORS for the requests that need it.
-        respond_with_json(
+        await respond_with_json(
            request,
            code,
            response_object,
@@ -431,13 +431,13 @@ class DirectServeJsonResource(_AsyncResource):
            canonical_json=self.canonical_json,
        )

-    def _send_error_response(
+    async def _send_error_response(
        self,
        f: failure.Failure,
        request: "SynapseRequest",
    ) -> None:
        """Implements _AsyncResource._send_error_response"""
-        return_json_error(f, request, None)
+        await return_json_error(f, request, None)


@attr.s(slots=True, frozen=True, auto_attribs=True)
@@ -565,13 +565,13 @@ class JsonResource(DirectServeJsonResource):

        return callback_return

-    def _send_error_response(
+    async def _send_error_response(
        self,
        f: failure.Failure,
        request: "SynapseRequest",
    ) -> None:
        """Implements _AsyncResource._send_error_response"""
-        return_json_error(f, request, self.hs.config)
+        await return_json_error(f, request, self.hs.config)


 class DirectServeHtmlResource(_AsyncResource):
@@ -593,7 +593,7 @@ class DirectServeHtmlResource(_AsyncResource):

        super().__init__(clock, extract_context)

-    def _send_response(
+    async def _send_response(
        self,
        request: "SynapseRequest",
        code: int,
@@ -606,7 +606,7 @@ class DirectServeHtmlResource(_AsyncResource):

        respond_with_html_bytes(request, code, html_bytes)

-    def _send_error_response(
+    async def _send_error_response(
        self,
        f: failure.Failure,
        request: "SynapseRequest",
@@ -780,7 +780,7 @@ def _encode_json_bytes(json_object: object) -> bytes:
    return json_encoder.encode(json_object).encode("utf-8")


-def respond_with_json(
+async def respond_with_json(
    request: "SynapseRequest",
    code: int,
    json_object: Any,
@@ -824,9 +824,7 @@ def respond_with_json(
    if send_cors:
        set_cors_headers(request)

-    run_in_background(
-        _async_write_json_to_request_in_thread, request, encoder, json_object
-    )
+    await _async_write_json_to_request_in_thread(request, encoder, json_object)
    return NOT_DONE_YET


@@ -882,6 +880,8 @@ async def _async_write_json_to_request_in_thread(
    Note: We don't use JsonEncoder.iterencode here as that falls back to the
    Python implementation (rather than the C backend), which is *much* more
    expensive.
+
+    The actual writing of bytes is not finished when this returns.
    """

    def encode(opentracing_span: "Optional[opentracing.Span]") -> bytes:
--- a/synapse/media/_base.py
+++ b/synapse/media/_base.py
@@ -122,9 +122,9 @@ MAXIMUM_ALLOWED_MAX_TIMEOUT_MS = 60_000
 _IMMUTABLE_ETAG = "1"


-def respond_404(request: SynapseRequest) -> None:
+async def respond_404(request: SynapseRequest) -> None:
    assert request.path is not None
-    respond_with_json(
+    await respond_with_json(
        request,
        404,
        cs_error("Not found '%s'" % (request.path.decode(),), code=Codes.NOT_FOUND),
@@ -154,7 +154,7 @@ async def respond_with_file(

        finish_request(request)
    else:
-        respond_404(request)
+        await respond_404(request)


 def add_file_headers(
--- a/synapse/media/media_repository.py
+++ b/synapse/media/media_repository.py
@@ -396,8 +396,8 @@ class MediaRepository:

        return MXCUri(self.server_name, media_id)

-    def respond_not_yet_uploaded(self, request: SynapseRequest) -> None:
-        respond_with_json(
+    async def respond_not_yet_uploaded(self, request: SynapseRequest) -> None:
+        await respond_with_json(
            request,
            504,
            cs_error("Media has not been uploaded yet", code=Codes.NOT_YET_UPLOADED),
@@ -455,7 +455,7 @@ class MediaRepository:
            await self.clock.sleep(0.5)

        logger.info("Media %s has not yet been uploaded", media_id)
-        self.respond_not_yet_uploaded(request)
+        await self.respond_not_yet_uploaded(request)
        return None

    async def get_local_media(
--- a/synapse/media/thumbnailer.py
+++ b/synapse/media/thumbnailer.py
@@ -699,7 +699,7 @@ class ThumbnailProvider:
            logger.info("Failed to find any generated thumbnails")

            assert request.path is not None
-            respond_with_json(
+            await respond_with_json(
                request,
                400,
                cs_error(
--- a/synapse/rest/client/media.py
+++ b/synapse/rest/client/media.py
@@ -112,7 +112,7 @@ class MediaConfigResource(RestServlet):
            )
        )
        response = user_specific_config if user_specific_config else self.limits_dict
-        respond_with_json(request, 200, response, send_cors=True)
+        await respond_with_json(request, 200, response, send_cors=True)


 class ThumbnailResource(RestServlet):
--- a/synapse/rest/media/config_resource.py
+++ b/synapse/rest/media/config_resource.py
@@ -50,4 +50,4 @@ class MediaConfigResource(RestServlet):
            )
        )
        response = user_specific_config if user_specific_config else self.limits_dict
-        respond_with_json(request, 200, response, send_cors=True)
+        await respond_with_json(request, 200, response, send_cors=True)
--- a/synapse/rest/media/create_resource.py
+++ b/synapse/rest/media/create_resource.py
@@ -79,7 +79,7 @@ class CreateResource(RestServlet):
            content_uri,
            unused_expires_at,
        )
-        respond_with_json(
+        await respond_with_json(
            request,
            200,
            {
--- a/synapse/rest/media/upload_resource.py
+++ b/synapse/rest/media/upload_resource.py
@@ -130,7 +130,7 @@ class UploadServlet(BaseUploadServlet):

        logger.info("Uploaded content with URI '%s'", content_uri)

-        respond_with_json(
+        await respond_with_json(
            request, 200, {"content_uri": str(content_uri)}, send_cors=True
        )

@@ -184,4 +184,4 @@ class AsyncUploadServlet(BaseUploadServlet):
                raise SynapseError(400, "Bad content")

            logger.info("Uploaded content for media ID %r", media_id)
-            respond_with_json(request, 200, {}, send_cors=True)
+            await respond_with_json(request, 200, {}, send_cors=True)
--- a/tests/http/test_additional_resource.py
+++ b/tests/http/test_additional_resource.py
@@ -36,7 +36,7 @@ class _AsyncTestCustomEndpoint:

    async def handle_request(self, request: Request) -> None:
        assert isinstance(request, SynapseRequest)
-        respond_with_json(request, 200, {"some_key": "some_value_async"})
+        await respond_with_json(request, 200, {"some_key": "some_value_async"})


 class _SyncTestCustomEndpoint:
@@ -45,7 +45,7 @@ class _SyncTestCustomEndpoint:

    async def handle_request(self, request: Request) -> None:
        assert isinstance(request, SynapseRequest)
-        respond_with_json(request, 200, {"some_key": "some_value_sync"})
+        await respond_with_json(request, 200, {"some_key": "some_value_sync"})


 class AdditionalResourceTests(HomeserverTestCase):