diff options
| -rw-r--r-- | changelog.d/4409.misc | 1 | ||||
| -rw-r--r-- | synapse/http/endpoint.py | 75 | ||||
| -rw-r--r-- | synapse/http/matrixfederationclient.py | 30 | ||||
| -rw-r--r-- | tests/http/test_fedclient.py | 54 |
4 files changed, 67 insertions, 93 deletions
diff --git a/changelog.d/4409.misc b/changelog.d/4409.misc new file mode 100644 index 0000000000..9cf2adfbb1 --- /dev/null +++ b/changelog.d/4409.misc @@ -0,0 +1 @@ +Remove redundant federation connection wrapping code diff --git a/synapse/http/endpoint.py b/synapse/http/endpoint.py index f86a0b624e..1c3b7ea28a 100644 --- a/synapse/http/endpoint.py +++ b/synapse/http/endpoint.py @@ -140,82 +140,15 @@ def matrix_federation_endpoint(reactor, destination, tls_client_options_factory= default_port = 8448 if port is None: - return _WrappingEndpointFac(SRVClientEndpoint( + return SRVClientEndpoint( reactor, "matrix", domain, protocol="tcp", default_port=default_port, endpoint=transport_endpoint, endpoint_kw_args=endpoint_kw_args - ), reactor) + ) else: - return _WrappingEndpointFac(transport_endpoint( + return transport_endpoint( reactor, domain, port, **endpoint_kw_args - ), reactor) - - -class _WrappingEndpointFac(object): - def __init__(self, endpoint_fac, reactor): - self.endpoint_fac = endpoint_fac - self.reactor = reactor - - @defer.inlineCallbacks - def connect(self, protocolFactory): - conn = yield self.endpoint_fac.connect(protocolFactory) - conn = _WrappedConnection(conn, self.reactor) - defer.returnValue(conn) - - -class _WrappedConnection(object): - """Wraps a connection and calls abort on it if it hasn't seen any action - for 2.5-3 minutes. - """ - __slots__ = ["conn", "last_request"] - - def __init__(self, conn, reactor): - object.__setattr__(self, "conn", conn) - object.__setattr__(self, "last_request", time.time()) - self._reactor = reactor - - def __getattr__(self, name): - return getattr(self.conn, name) - - def __setattr__(self, name, value): - setattr(self.conn, name, value) - - def _time_things_out_maybe(self): - # We use a slightly shorter timeout here just in case the callLater is - # triggered early. Paranoia ftw. - # TODO: Cancel the previous callLater rather than comparing time.time()? - if time.time() - self.last_request >= 2.5 * 60: - self.abort() - # Abort the underlying TLS connection. The abort() method calls - # loseConnection() on the TLS connection which tries to - # shutdown the connection cleanly. We call abortConnection() - # since that will promptly close the TLS connection. - # - # In Twisted >18.4; the TLS connection will be None if it has closed - # which will make abortConnection() throw. Check that the TLS connection - # is not None before trying to close it. - if self.transport.getHandle() is not None: - self.transport.abortConnection() - - def request(self, request): - self.last_request = time.time() - - # Time this connection out if we haven't send a request in the last - # N minutes - # TODO: Cancel the previous callLater? - self._reactor.callLater(3 * 60, self._time_things_out_maybe) - - d = self.conn.request(request) - - def update_request_time(res): - self.last_request = time.time() - # TODO: Cancel the previous callLater? - self._reactor.callLater(3 * 60, self._time_things_out_maybe) - return res - - d.addCallback(update_request_time) - - return d + ) class SRVClientEndpoint(object): diff --git a/synapse/http/matrixfederationclient.py b/synapse/http/matrixfederationclient.py index ea2fc64b99..250bb1ef91 100644 --- a/synapse/http/matrixfederationclient.py +++ b/synapse/http/matrixfederationclient.py @@ -321,23 +321,23 @@ class MatrixFederationHttpClient(object): url_str, ) - # we don't want all the fancy cookie and redirect handling that - # treq.request gives: just use the raw Agent. - request_deferred = self.agent.request( - method_bytes, - url_bytes, - headers=Headers(headers_dict), - bodyProducer=producer, - ) - - request_deferred = timeout_deferred( - request_deferred, - timeout=_sec_timeout, - reactor=self.hs.get_reactor(), - ) - try: with Measure(self.clock, "outbound_request"): + # we don't want all the fancy cookie and redirect handling + # that treq.request gives: just use the raw Agent. + request_deferred = self.agent.request( + method_bytes, + url_bytes, + headers=Headers(headers_dict), + bodyProducer=producer, + ) + + request_deferred = timeout_deferred( + request_deferred, + timeout=_sec_timeout, + reactor=self.hs.get_reactor(), + ) + response = yield make_deferred_yieldable( request_deferred, ) diff --git a/tests/http/test_fedclient.py b/tests/http/test_fedclient.py index b2e38276d8..8426eee400 100644 --- a/tests/http/test_fedclient.py +++ b/tests/http/test_fedclient.py @@ -17,6 +17,7 @@ from mock import Mock from twisted.internet.defer import TimeoutError from twisted.internet.error import ConnectingCancelledError, DNSLookupError +from twisted.test.proto_helpers import StringTransport from twisted.web.client import ResponseNeverReceived from twisted.web.http import HTTPChannel @@ -44,7 +45,7 @@ class FederationClientTests(HomeserverTestCase): def test_dns_error(self): """ - If the DNS raising returns an error, it will bubble up. + If the DNS lookup returns an error, it will bubble up. """ d = self.cl.get_json("testserv2:8008", "foo/bar", timeout=10000) self.pump() @@ -63,7 +64,7 @@ class FederationClientTests(HomeserverTestCase): self.pump() # Nothing happened yet - self.assertFalse(d.called) + self.assertNoResult(d) # Make sure treq is trying to connect clients = self.reactor.tcpClients @@ -72,7 +73,7 @@ class FederationClientTests(HomeserverTestCase): self.assertEqual(clients[0][1], 8008) # Deferred is still without a result - self.assertFalse(d.called) + self.assertNoResult(d) # Push by enough to time it out self.reactor.advance(10.5) @@ -94,7 +95,7 @@ class FederationClientTests(HomeserverTestCase): self.pump() # Nothing happened yet - self.assertFalse(d.called) + self.assertNoResult(d) # Make sure treq is trying to connect clients = self.reactor.tcpClients @@ -107,7 +108,7 @@ class FederationClientTests(HomeserverTestCase): client.makeConnection(conn) # Deferred is still without a result - self.assertFalse(d.called) + self.assertNoResult(d) # Push by enough to time it out self.reactor.advance(10.5) @@ -135,7 +136,7 @@ class FederationClientTests(HomeserverTestCase): client.makeConnection(conn) # Deferred does not have a result - self.assertFalse(d.called) + self.assertNoResult(d) # Send it the HTTP response client.dataReceived(b"HTTP/1.1 200 OK\r\nServer: Fake\r\n\r\n") @@ -159,7 +160,7 @@ class FederationClientTests(HomeserverTestCase): client.makeConnection(conn) # Deferred does not have a result - self.assertFalse(d.called) + self.assertNoResult(d) # Send it the HTTP response client.dataReceived( @@ -195,3 +196,42 @@ class FederationClientTests(HomeserverTestCase): request = server.requests[0] content = request.content.read() self.assertEqual(content, b'{"a":"b"}') + + def test_closes_connection(self): + """Check that the client closes unused HTTP connections""" + d = self.cl.get_json("testserv:8008", "foo/bar") + + self.pump() + + # there should have been a call to connectTCP + clients = self.reactor.tcpClients + self.assertEqual(len(clients), 1) + (_host, _port, factory, _timeout, _bindAddress) = clients[0] + + # complete the connection and wire it up to a fake transport + client = factory.buildProtocol(None) + conn = StringTransport() + client.makeConnection(conn) + + # that should have made it send the request to the connection + self.assertRegex(conn.value(), b"^GET /foo/bar") + + # Send the HTTP response + client.dataReceived( + b"HTTP/1.1 200 OK\r\n" + b"Content-Type: application/json\r\n" + b"Content-Length: 2\r\n" + b"\r\n" + b"{}" + ) + + # We should get a successful response + r = self.successResultOf(d) + self.assertEqual(r, {}) + + self.assertFalse(conn.disconnecting) + + # wait for a while + self.pump(120) + + self.assertTrue(conn.disconnecting) |