Fix error log when writing a large response. (#20246)

We sometimes see a lot of `ERROR` logs like the following:

```
Closing scope Scope<... master.write_bytes_to_request> which is not the currently-active one None
```

This error is generated by `opentracing.Scope` checking that it is the
"active" one. Synapse tracks "active" spans via logcontexts, so this
indirectly asserts that the scope is closed in the context it was opened
in. However, the producer methods are often called from the reactor and
therefore withing the sentinel logcontext, which produces the error
above.

This specifically happens when the producer tries to write large
responses but gets paused, and then later resumes.

The fix is to simply use `Span` directly, rather than scopes. `Span`
does not perform the checks.

I noticed this when deploying #19979, though it is unrelated.

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.

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:

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:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+#
+
+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)