Fix Twisted looking things

Author: MadLittleMods

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 (
+    make_deferred_yieldable,
+    run_in_background,
+    PreserveLoggingContext,
+)
 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

@@ -48,6 +48,7 @@
     defer_to_threadpool,
     make_deferred_yieldable,
     run_in_background,
+    PreserveLoggingContext,
 )
 from synapse.util.async_helpers import DeferredEvent
 from synapse.util.clock import Clock
@@ -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)