Gracefully handle a pending connection during shutdown.

changelog.d/8607.feature

@@ -0,0 +1 @@
+Support generating structured logs via the standard logging configuration.

changelog.d/8685.feature

@@ -0,0 +1 @@
+Support generating structured logs via the standard logging configuration.

synapse/logging/_remote.py

@@ -26,14 +26,15 @@
 from zope.interface import implementer
 
 from twisted.application.internet import ClientService
-from twisted.internet.defer import Deferred
+from twisted.internet.defer import CancelledError, Deferred
 from twisted.internet.endpoints import (
  HostnameEndpoint,
  TCP4ClientEndpoint,
  TCP6ClientEndpoint,
 )
 from twisted.internet.interfaces import IPushProducer, ITransport
 from twisted.internet.protocol import Factory, Protocol
+from twisted.python.failure import Failure
 
 logger = logging.getLogger(__name__)
 
@@ -146,17 +147,21 @@ def _connect(self) -> None:
 
  self._connection_waiter = self._service.whenConnected(failAfterFailures=1)
 
- @self._connection_waiter.addErrback
- def fail(r):
- r.printTraceback(file=sys.__stderr__)
+ def fail(failure: Failure) -> None:
+ # If the Deferred was cancelled (e.g. during shutdown) do not try to
+ # reconnect (this will cause an infinite loop of errors).
+ if failure.check(CancelledError):
+ return
+
+ # For a different error, print the traceback and re-connect.
+ failure.printTraceback(file=sys.__stderr__)
  self._connection_waiter = None
  self._connect()
 
- @self._connection_waiter.addCallback
- def writer(r):
+ def writer(result: Protocol) -> None:
  # We have a connection. If we already have a producer, and its
  # transport is the same, just trigger a resumeProducing.
- if self._producer and r.transport is self._producer.transport:
+ if self._producer and result.transport is self._producer.transport:
  self._producer.resumeProducing()
  self._connection_waiter = None
  return
@@ -167,12 +172,14 @@ def writer(r):
 
  # Make a new producer and start it.
  self._producer = LogProducer(
- buffer=self._buffer, transport=r.transport, format=self.format,
+ buffer=self._buffer, transport=result.transport, format=self.format,
  )
- r.transport.registerProducer(self._producer, True)
+ result.transport.registerProducer(self._producer, True)
  self._producer.resumeProducing()
  self._connection_waiter = None
 
+ self._connection_waiter.addCallbacks(writer, fail)
+
  def _handle_pressure(self) -> None:
  """
  Handle backpressure by shedding records.

tests/logging/test_remote_handler.py

@@ -151,3 +151,19 @@ def test_log_backpressure_cut_middle(self):
  + ["warn %s" % (i,) for i in range(15, 20)],
  logs,
  )
+
+ def test_cancel_connection(self):
+ """
+ Gracefully handle the connection being cancelled.
+ """
+ handler = RemoteHandler(
+ "127.0.0.1", 9000, maximum_buffer=10, _reactor=self.reactor
+ )
+ logger = self.get_logger(handler)
+
+ # Send a message.
+ logger.info("Hello there, %s!", "wally")
+
+ # Do not accept the connection and shutdown. This causes the pending
+ # connection to be cancelled (and should not raise any exceptions).
+ handler.close()