diff --git a/changelog.d/20246.bugfix b/changelog.d/20246.bugfix new file mode 100644 index 0000000000..e2095ba861 --- /dev/null +++ b/changelog.d/20246.bugfix @@ -0,0 +1 @@ +Fix spurious `Closing scope ... which is not the currently-active one` errors being logged when writing large responses with tracing enabled. diff --git a/synapse/http/server.py b/synapse/http/server.py index a0ae20be16..37dd77b008 100644 --- a/synapse/http/server.py +++ b/synapse/http/server.py @@ -758,10 +758,15 @@ class _ByteProducer: self._request: Request | None = request self._iterator = iterator self._paused = False - self.tracing_scope = start_active_span( - "write_bytes_to_request", - ) - self.tracing_scope.__enter__() + + # Start a span for writing bytes to the request. We manually manage its + # lifecycle, so set `finish_on_close=False`. + # + # Note that we cannot use `active_span()` in the functions below, as + # they are called from the reactor and therefore outside the request's + # context. + with start_active_span("write_bytes_to_request", finish_on_close=False): + self._span = active_span() try: self._request.registerProducer(self, True) @@ -772,8 +777,7 @@ class _ByteProducer: logger.info("Connection disconnected before response was written: %r", e) # We drop our references to data we'll not use. - self._iterator = iter(()) - self.tracing_scope.__exit__(type(e), None, e.__traceback__) + self.stopProducing() else: # Start producing if `registerProducer` was successful self.resumeProducing() @@ -787,9 +791,8 @@ class _ByteProducer: self._request.write(b"".join(data)) def pauseProducing(self) -> None: - opentracing_span = active_span() - if opentracing_span is not None: - opentracing_span.log_kv({"event": "producer_paused"}) + if self._span is not None: + self._span.log_kv({"event": "producer_paused"}) self._paused = True def resumeProducing(self) -> None: @@ -800,9 +803,8 @@ class _ByteProducer: self._paused = False - opentracing_span = active_span() - if opentracing_span is not None: - opentracing_span.log_kv({"event": "producer_resumed"}) + if self._span is not None: + self._span.log_kv({"event": "producer_resumed"}) # Write until there's backpressure telling us to stop. while not self._paused: @@ -836,9 +838,13 @@ class _ByteProducer: self._send_data(buffer) def stopProducing(self) -> None: - # Clear a circular reference. + # Clear a circular reference and drop references to the data. + self._iterator = iter(()) self._request = None - self.tracing_scope.__exit__(None, None, None) + + if self._span is not None: + self._span.finish() + self._span = None def _encode_json_bytes(json_object: object) -> bytes: diff --git a/tests/http/test_server.py b/tests/http/test_server.py new file mode 100644 index 0000000000..24ce91bd1e --- /dev/null +++ b/tests/http/test_server.py @@ -0,0 +1,118 @@ +# +# This file is licensed under the Affero General Public License (AGPL) version 3. +# +# Copyright (C) 2026 Element Creations, Ltd +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# See the GNU Affero General Public License for more details: +# . +# + +from typing import cast +from unittest.mock import Mock + +from twisted.web.server import Request + +from synapse.http.server import _ByteProducer +from synapse.logging.context import ( + LoggingContext, +) +from synapse.logging.opentracing import start_active_span + +try: + import jaeger_client + import opentracing + + from synapse.logging.scopecontextmanager import LogContextScopeManager +except ImportError: + jaeger_client = None # type: ignore + +from synapse.util.iterutils import chunk_seq + +from tests.unittest import TestCase + + +class ByteProducerTestCase(TestCase): + if jaeger_client is None: + skip = "Requires jaeger_client" # type: ignore[unreachable] + + def test_paused_response(self) -> None: + """Test that a paused response is correctly resumed and finished without + logging errors. + + This is a regression test where writing a large response that gets + paused by the transport produces the error "Closing scope ... which is + not the currently-active one" + """ + + ## 1. First, set up the Jaeger tracer. + config = jaeger_client.config.Config( + config={}, service_name="test", scope_manager=LogContextScopeManager() + ) + tracer = config.create_tracer( + sampler=jaeger_client.ConstSampler(True), + reporter=jaeger_client.reporter.NullReporter(), + ) + previous_tracer = opentracing.tracer + opentracing.set_global_tracer(tracer) + self.addCleanup(opentracing.set_global_tracer, previous_tracer) + + ## 2. Now create a mock Request. + request = Mock(spec=Request) + + written_buffer = b"" # The data written to the request + + def write(data: bytes) -> None: + nonlocal written_buffer + written_buffer += data + + # Pause after the first write, as the transport does when its send + # buffer fills up. + if request.write.call_count == 1: + request.registerProducer.call_args.args[0].pauseProducing() + + request.write.side_effect = write + + ## 3. Now we start writing the bytes to the request via the + ## _ByteProducer. This should not log any errors or warnings. + with ( + self.assertNoLogs("synapse.logging.context", "WARNING"), + self.assertNoLogs("synapse.logging.scopecontextmanager", "ERROR"), + ): + # The data that we write to the request via the _ByteProducer. This + # is arbitrary and large to ensure multiple chunks are written. It + # needs to be big enough that _ByteProducer will not try and batch + # up the chunks internally (if it does then the assertion that we + # have multiple writes below will fail). + buffer_to_write = b"x" * 6000 + + # The initial write happens within the request log context and span + with ( + LoggingContext(name="request", server_name="test_server"), + start_active_span("servlet"), + ): + # Break the buffer into chunks for writing. We use an arbitrary + # chunk size that ensures we have a few distinct writes. + iterable = chunk_seq(buffer_to_write, 2000) + + # Start writing the data. The _ByteProducer will start writing + # immediately on construction. + producer = _ByteProducer(cast(Request, request), iterable) + + # The request paused after the first write, and so it should not + # have finished yet. + request.finish.assert_not_called() + self.assertLess(len(written_buffer), len(buffer_to_write)) + self.assertTrue(producer._paused) + + # Mimic the request resuming the producer. This happens from the + # reactor and so outside the request log context. + producer.resumeProducing() + + # All data should now be written and the request should be finished. + request.finish.assert_called_once() + self.assertEqual(written_buffer, buffer_to_write)