Wait for response to be sent

synapse/http/proxy.py

@@ -173,7 +173,7 @@ async def _async_render(self, request: "SynapseRequest") -> Tuple[int, Any]:
 
         return response.code, response
 
-    def _send_response(
+    async def _send_response(
         self,
         request: "SynapseRequest",
         code: int,
@@ -205,7 +205,7 @@ def _send_response(
 
         response.deliverBody(_ProxyResponseBody(request))
 
-    def _send_error_response(
+    async def _send_error_response(
         self,
         f: failure.Failure,
         request: "SynapseRequest",

synapse/http/server.py

@@ -75,7 +75,7 @@
     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 @@
 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 @@ async def _async_render_wrapper(self, request: "SynapseRequest") -> None:
 
                 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 @@ async def _async_render(
         raise UnrecognizedRequestError(code=405)
 
     @abc.abstractmethod
-    def _send_response(
+    async def _send_response(
         self,
         request: "SynapseRequest",
         code: int,
@@ -389,7 +389,7 @@ def _send_response(
         raise NotImplementedError()
 
     @abc.abstractmethod
-    def _send_error_response(
+    async def _send_error_response(
         self,
         f: failure.Failure,
         request: "SynapseRequest",
@@ -415,29 +415,29 @@ def __init__(
         super().__init__(clock, extract_context)
         self.canonical_json = canonical_json
 
-    def _send_response(
+    async def _send_response(
         self,
         request: "SynapseRequest",
         code: int,
         response_object: Any,
     ) -> 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,
             send_cors=True,
             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 @@ async def _async_render(self, request: "SynapseRequest") -> Tuple[int, Any]:
 
         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 @@ def __init__(
 
         super().__init__(clock, extract_context)
 
-    def _send_response(
+    async def _send_response(
         self,
         request: "SynapseRequest",
         code: int,
@@ -606,7 +606,7 @@ def _send_response(
 
         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:

synapse/media/_base.py

@@ -122,9 +122,9 @@
 _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(

synapse/media/media_repository.py

@@ -396,8 +396,8 @@ async def create_or_update_content(
 
         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 @@ async def get_local_media_info(
             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(

synapse/media/thumbnailer.py

@@ -699,7 +699,7 @@ async def _select_and_respond_with_thumbnail(
             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(

synapse/rest/client/media.py

@@ -112,7 +112,7 @@ async def on_GET(self, request: SynapseRequest) -> None:
             )
         )
         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):

synapse/rest/media/config_resource.py

@@ -50,4 +50,4 @@ async def on_GET(self, request: SynapseRequest) -> None:
             )
         )
         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)

synapse/rest/media/create_resource.py

@@ -79,7 +79,7 @@ async def on_POST(self, request: SynapseRequest) -> None:
             content_uri,
             unused_expires_at,
         )
-        respond_with_json(
+        await respond_with_json(
             request,
             200,
             {

synapse/rest/media/upload_resource.py

@@ -130,7 +130,7 @@ async def on_POST(self, request: SynapseRequest) -> None:
 
         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 @@ async def on_PUT(
                 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)

tests/http/test_additional_resource.py

@@ -36,7 +36,7 @@ def __init__(self, config: JsonDict, module_api: Any) -> None:
 
     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 @@ def __init__(self, config: JsonDict, module_api: Any) -> None:
 
     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):