Fix bad deferred logcontext handling

changelog.d/19180.misc

@@ -0,0 +1 @@
+Fix bad deferred logcontext handling across the codebase.

synapse/handlers/worker_lock.py

@@ -184,9 +184,7 @@ def _wake_all_locks(
             locks: Collection[WaitingLock | WaitingMultiLock],
         ) -> None:
             for lock in locks:
-                deferred = lock.deferred
-                if not deferred.called:
-                    deferred.callback(None)
+                lock.release_lock()
 
         self._clock.call_later(
             0,
@@ -215,6 +213,12 @@ class WaitingLock:
         lambda: start_active_span("WaitingLock.lock")
     )
 
+    def release_lock(self) -> None:
+        """Release the lock (by resolving the deferred)"""
+        if not self.deferred.called:
+            with PreserveLoggingContext():
+                self.deferred.callback(None)
+
     async def __aenter__(self) -> None:
         self._lock_span.__enter__()
 
@@ -298,6 +302,12 @@ class WaitingMultiLock:
         lambda: start_active_span("WaitingLock.lock")
     )
 
+    def release_lock(self) -> None:
+        """Release the lock (by resolving the deferred)"""
+        if not self.deferred.called:
+            with PreserveLoggingContext():
+                self.deferred.callback(None)
+
     async def __aenter__(self) -> None:
         self._lock_span.__enter__()
 

synapse/http/client.py

@@ -77,7 +77,11 @@
 from synapse.http.proxyagent import ProxyAgent
 from synapse.http.replicationagent import ReplicationAgent
 from synapse.http.types import QueryParams
-from synapse.logging.context import make_deferred_yieldable, run_in_background
+from synapse.logging.context import (
+    PreserveLoggingContext,
+    make_deferred_yieldable,
+    run_in_background,
+)
 from synapse.logging.opentracing import set_tag, start_active_span, tags
 from synapse.metrics import SERVER_NAME_LABEL
 from synapse.types import ISynapseReactor, StrSequence
@@ -1036,7 +1040,8 @@ def _maybe_fail(self) -> None:
         Report a max size exceed error and disconnect the first time this is called.
         """
         if not self.deferred.called:
-            self.deferred.errback(BodyExceededMaxSize())
+            with PreserveLoggingContext():
+                self.deferred.errback(BodyExceededMaxSize())
             # Close the connection (forcefully) since all the data will get
             # discarded anyway.
             assert self.transport is not None
@@ -1135,7 +1140,8 @@ def on_part_data(data: bytes, start: int, end: int) -> None:
                         logger.warning(
                             "Exception encountered writing file data to stream: %s", e
                         )
-                        self.deferred.errback()
+                        with PreserveLoggingContext():
+                            self.deferred.errback()
                     self.file_length += end - start
 
             callbacks: "multipart.MultipartCallbacks" = {
@@ -1147,7 +1153,8 @@ def on_part_data(data: bytes, start: int, end: int) -> None:
 
         self.total_length += len(incoming_data)
         if self.max_length is not None and self.total_length >= self.max_length:
-            self.deferred.errback(BodyExceededMaxSize())
+            with PreserveLoggingContext():
+                self.deferred.errback(BodyExceededMaxSize())
             # Close the connection (forcefully) since all the data will get
             # discarded anyway.
             assert self.transport is not None
@@ -1157,7 +1164,8 @@ def on_part_data(data: bytes, start: int, end: int) -> None:
             self.parser.write(incoming_data)
         except Exception as e:
             logger.warning("Exception writing to multipart parser: %s", e)
-            self.deferred.errback()
+            with PreserveLoggingContext():
+                self.deferred.errback()
             return
 
     def connectionLost(self, reason: Failure = connectionDone) -> None:
@@ -1167,9 +1175,11 @@ def connectionLost(self, reason: Failure = connectionDone) -> None:
 
         if reason.check(ResponseDone):
             self.multipart_response.length = self.file_length
-            self.deferred.callback(self.multipart_response)
+            with PreserveLoggingContext():
+                self.deferred.callback(self.multipart_response)
         else:
-            self.deferred.errback(reason)
+            with PreserveLoggingContext():
+                self.deferred.errback(reason)
 
 
 class _ReadBodyWithMaxSizeProtocol(protocol.Protocol):
@@ -1193,15 +1203,17 @@ def dataReceived(self, data: bytes) -> None:
         try:
             self.stream.write(data)
         except Exception:
-            self.deferred.errback()
+            with PreserveLoggingContext():
+                self.deferred.errback()
             return
 
         self.length += len(data)
         # The first time the maximum size is exceeded, error and cancel the
         # connection. dataReceived might be called again if data was received
         # in the meantime.
         if self.max_size is not None and self.length >= self.max_size:
-            self.deferred.errback(BodyExceededMaxSize())
+            with PreserveLoggingContext():
+                self.deferred.errback(BodyExceededMaxSize())
             # Close the connection (forcefully) since all the data will get
             # discarded anyway.
             assert self.transport is not None
@@ -1213,7 +1225,8 @@ def connectionLost(self, reason: Failure = connectionDone) -> None:
             return
 
         if reason.check(ResponseDone):
-            self.deferred.callback(self.length)
+            with PreserveLoggingContext():
+                self.deferred.callback(self.length)
         elif reason.check(PotentialDataLoss):
             # This applies to requests which don't set `Content-Length` or a
             # `Transfer-Encoding` in the response because in this case the end of the
@@ -1222,9 +1235,11 @@ def connectionLost(self, reason: Failure = connectionDone) -> None:
             # behavior is expected of some servers (like YouTube), let's ignore it.
             # Stolen from https://github.com/twisted/treq/pull/49/files
             # http://twistedmatrix.com/trac/ticket/4840
-            self.deferred.callback(self.length)
+            with PreserveLoggingContext():
+                self.deferred.callback(self.length)
         else:
-            self.deferred.errback(reason)
+            with PreserveLoggingContext():
+                self.deferred.errback(reason)
 
 
 def read_body_with_max_size(

synapse/http/connectproxyclient.py

@@ -41,6 +41,8 @@
 from twisted.python.failure import Failure
 from twisted.web import http
 
+from synapse.logging.context import PreserveLoggingContext
+
 logger = logging.getLogger(__name__)
 
 
@@ -176,14 +178,16 @@ def buildProtocol(self, addr: IAddress) -> "HTTPConnectProtocol":
     def clientConnectionFailed(self, connector: IConnector, reason: Failure) -> None:
         logger.debug("Connection to proxy failed: %s", reason)
         if not self.on_connection.called:
-            self.on_connection.errback(reason)
+            with PreserveLoggingContext():
+                self.on_connection.errback(reason)
         if isinstance(self.wrapped_factory, ClientFactory):
             return self.wrapped_factory.clientConnectionFailed(connector, reason)
 
     def clientConnectionLost(self, connector: IConnector, reason: Failure) -> None:
         logger.debug("Connection to proxy lost: %s", reason)
         if not self.on_connection.called:
-            self.on_connection.errback(reason)
+            with PreserveLoggingContext():
+                self.on_connection.errback(reason)
         if isinstance(self.wrapped_factory, ClientFactory):
             return self.wrapped_factory.clientConnectionLost(connector, reason)
 
@@ -238,14 +242,16 @@ def connectionLost(self, reason: Failure = connectionDone) -> None:
         self.http_setup_client.connectionLost(reason)
 
         if not self.connected_deferred.called:
-            self.connected_deferred.errback(reason)
+            with PreserveLoggingContext():
+                self.connected_deferred.errback(reason)
 
     def proxyConnected(self, _: Union[None, "defer.Deferred[None]"]) -> None:
         self.wrapped_connection_started = True
         assert self.transport is not None
         self.wrapped_protocol.makeConnection(self.transport)
 
-        self.connected_deferred.callback(self.wrapped_protocol)
+        with PreserveLoggingContext():
+            self.connected_deferred.callback(self.wrapped_protocol)
 
         # Get any pending data from the http buf and forward it to the original protocol
         buf = self.http_setup_client.clearLineBuffer()
@@ -303,7 +309,8 @@ def handleStatus(self, version: bytes, status: bytes, message: bytes) -> None:
 
     def handleEndHeaders(self) -> None:
         logger.debug("End Headers")
-        self.on_connected.callback(None)
+        with PreserveLoggingContext():
+            self.on_connected.callback(None)
 
     def handleResponse(self, body: bytes) -> None:
         pass

synapse/media/_base.py

@@ -45,6 +45,7 @@
 from synapse.http.server import finish_request, respond_with_json
 from synapse.http.site import SynapseRequest
 from synapse.logging.context import (
+    PreserveLoggingContext,
     defer_to_threadpool,
     make_deferred_yieldable,
     run_in_background,
@@ -753,9 +754,10 @@ def stopProducing(self) -> None:
         self.wakeup_event.set()
 
         if not self.deferred.called:
-            self.deferred.errback(
-                ConsumerRequestedStopError("Consumer asked us to stop producing")
-            )
+            with PreserveLoggingContext():
+                self.deferred.errback(
+                    ConsumerRequestedStopError("Consumer asked us to stop producing")
+                )
 
     async def start_read_loop(self) -> None:
         """This is the loop that drives reading/writing"""
@@ -809,7 +811,8 @@ def _error(self, failure: Failure) -> None:
             self.consumer = None
 
         if not self.deferred.called:
-            self.deferred.errback(failure)
+            with PreserveLoggingContext():
+                self.deferred.errback(failure)
 
     def _finish(self) -> None:
         """Called when we have finished writing (either on success or
@@ -823,4 +826,5 @@ def _finish(self) -> None:
             self.consumer = None
 
         if not self.deferred.called:
-            self.deferred.callback(None)
+            with PreserveLoggingContext():
+                self.deferred.callback(None)

synapse/util/async_helpers.py

@@ -813,7 +813,8 @@ def time_it_out() -> None:
         # will have errbacked new_d, but in case it hasn't, errback it now.
 
         if not new_d.called:
-            new_d.errback(defer.TimeoutError("Timed out after %gs" % (timeout,)))
+            with PreserveLoggingContext():
+                new_d.errback(defer.TimeoutError("Timed out after %gs" % (timeout,)))
 
     # We don't track these calls since they are short.
     delayed_call = clock.call_later(
@@ -840,11 +841,13 @@ def cancel_timeout(result: _T) -> _T:
 
     def success_cb(val: _T) -> None:
         if not new_d.called:
-            new_d.callback(val)
+            with PreserveLoggingContext():
+                new_d.callback(val)
 
     def failure_cb(val: Failure) -> None:
         if not new_d.called:
-            new_d.errback(val)
+            with PreserveLoggingContext():
+                new_d.errback(val)
 
     deferred.addCallbacks(success_cb, failure_cb)
 
@@ -946,7 +949,8 @@ def handle_cancel(new_deferred: "defer.Deferred[T]") -> None:
         # propagating. we then `unpause` it once the wrapped deferred completes, to
         # propagate the exception.
         new_deferred.pause()
-        new_deferred.errback(Failure(CancelledError()))
+        with PreserveLoggingContext():
+            new_deferred.errback(Failure(CancelledError()))
 
         deferred.addBoth(lambda _: new_deferred.unpause())
 
@@ -978,15 +982,6 @@ async def sleep(self, name: str, delay_ms: int) -> None:
         """Sleep for the given number of milliseconds, or return if the given
         `name` is explicitly woken up.
         """
-
-        # Create a deferred that gets called in N seconds
-        sleep_deferred: "defer.Deferred[None]" = defer.Deferred()
-        call = self._clock.call_later(
-            delay_ms / 1000,
-            sleep_deferred.callback,
-            None,
-        )
-
         # Create a deferred that will get called if `wake` is called with
         # the same `name`.
         stream_set = self._streams.setdefault(name, set())
@@ -996,13 +991,14 @@ async def sleep(self, name: str, delay_ms: int) -> None:
         try:
             # Wait for either the delay or for `wake` to be called.
             await make_deferred_yieldable(
-                defer.DeferredList(
-                    [sleep_deferred, notify_deferred],
-                    fireOnOneCallback=True,
-                    fireOnOneErrback=True,
-                    consumeErrors=True,
+                timeout_deferred(
+                    deferred=stop_cancellation(notify_deferred),
+                    timeout=delay_ms / 1000,
+                    clock=self._clock,
                 )
             )
+        except defer.TimeoutError:
+            pass
         finally:
             # Clean up the state
             curr_stream_set = self._streams.get(name)
@@ -1011,10 +1007,6 @@ async def sleep(self, name: str, delay_ms: int) -> None:
                 if len(curr_stream_set) == 0:
                     self._streams.pop(name)
 
-            # Cancel the sleep if we were woken up
-            if call.active():
-                call.cancel()
-
 
 class DeferredEvent:
     """Like threading.Event but for async code"""

synapse/util/caches/deferred_cache.py

@@ -39,6 +39,7 @@
 from twisted.internet import defer
 from twisted.python.failure import Failure
 
+from synapse.logging.context import PreserveLoggingContext
 from synapse.metrics import SERVER_NAME_LABEL
 from synapse.util.async_helpers import ObservableDeferred
 from synapse.util.caches.lrucache import LruCache
@@ -514,7 +515,8 @@ def complete_bulk(
             cache._completed_callback(value, self, key)
 
         if self._deferred:
-            self._deferred.callback(result)
+            with PreserveLoggingContext():
+                self._deferred.callback(result)
 
     def error_bulk(
         self, cache: DeferredCache[KT, VT], keys: Collection[KT], failure: Failure
@@ -524,4 +526,5 @@ def error_bulk(
             cache._error_callback(failure, self, key)
 
         if self._deferred:
-            self._deferred.errback(failure)
+            with PreserveLoggingContext():
+                self._deferred.errback(failure)