Honor allow_half_open when a client aborts early

doc/admin-guide/files/records.yaml.en.rst

@@ -6036,6 +6036,15 @@ Sockets
    Turn on or off support for connection half open for client side. Default is on, so
    after client sends FIN, the connection is still there.
 
+   When this is enabled and the client aborts before the origin server has sent its response header, |TS| keeps the transaction
+   alive so that the response can still be fetched and cached, which is known as a **background fill**. This is done even for
+   transports such as TLS and HTTP/2 that have no way to half close a connection.
+
+   When this is disabled, a client abort ends the transaction: the connection to the origin server is closed rather than held
+   open for a client that is no longer there. See :ts:cv:`proxy.config.http.background_fill_completed_threshold` and
+   :ts:cv:`proxy.config.http.background_fill_active_timeout` for controlling background fills that have already started
+   delivering the response to the client.
+
 .. ts:cv:: CONFIG proxy.config.http.wait_for_cache INT 0
 
    Accepting inbound connections and starting the cache are independent

src/proxy/http/HttpSM.cc

@@ -910,9 +910,12 @@ HttpSM::state_watch_for_client_abort(int event, void *data)
         if (netvc) {
           netvc->do_io_shutdown(IO_SHUTDOWN_READ);
         }
-      } else if (t_state.txn_conf->cache_http &&
+      } else if (t_state.txn_conf->allow_half_open > 0 && t_state.txn_conf->cache_http &&
                  (server_entry != nullptr && server_entry->vc_read_handler == &HttpSM::state_read_server_response_header)) {
-        // if HttpSM is waiting response header from origin server, keep it for a while to run background fetch
+        // Half open connections are configured, but the transport does not support them (e.g. TLS or HTTP/2). If HttpSM is
+        // waiting response header from origin server, keep it for a while to run background fetch. Note that the operator
+        // disabling half open connections is handled below: the transaction is aborted rather than kept alive for a client
+        // that is no longer there.
         _ua.get_txn()->do_io_shutdown(IO_SHUTDOWN_READWRITE);
       } else {
         _ua.get_txn()->do_io_close();

tests/gold_tests/cache/abort_detecting_origin.py

@@ -0,0 +1,116 @@
+#!/usr/bin/env python3
+'''An origin server that reports whether the proxy closed the connection.
+
+The server accepts a single connection, reads the request, then waits for the
+configured delay before responding. While waiting, it watches the connection for
+the proxy closing it, which is what a proxy is expected to do when its client
+aborts the request before the origin responds.
+'''
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you under the Apache License, Version 2.0 (the
+#  "License"); you may not use this file except in compliance
+#  with the License.  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+import argparse
+import select
+import socket
+import sys
+import time
+
+ABORT_DETECTED = 'proxy_closed_connection'
+ABORT_NOT_DETECTED = 'proxy_kept_connection_open'
+
+RESPONSE_BODY = b'0123456789'
+RESPONSE = (
+    b'HTTP/1.1 200 OK\r\n'
+    b'Content-Type: text/plain\r\n'
+    b'Cache-Control: max-age=300\r\n'
+    b'Content-Length: ' + str(len(RESPONSE_BODY)).encode() + b'\r\n'
+    b'Connection: close\r\n'
+    b'\r\n' + RESPONSE_BODY)
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('port', type=int, help='The port to listen on.')
+    parser.add_argument('--delay', type=float, default=10.0, help='Seconds to wait before sending the response.')
+    return parser.parse_args()
+
+
+def read_request(connection: socket.socket) -> bool:
+    '''Read the request headers off of the connection.
+
+    :param connection: The accepted connection to read from.
+    :returns: Whether a complete set of request headers was read.
+    '''
+    request = b''
+    while b'\r\n\r\n' not in request:
+        chunk = connection.recv(4096)
+        if not chunk:
+            return False
+        request += chunk
+    request_line = request.split(b'\r\n')[0].decode(errors='replace')
+    print(f'Received request: {request_line}', flush=True)
+    return True
+
+
+def wait_for_abort(connection: socket.socket, delay: float) -> bool:
+    '''Wait for the delay, watching for the peer closing the connection.
+
+    :param connection: The accepted connection to watch.
+    :param delay: The number of seconds to wait before giving up.
+    :returns: Whether the peer closed the connection during the delay.
+    '''
+    deadline = time.monotonic() + delay
+    while True:
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            return False
+        readable, _, _ = select.select([connection], [], [], remaining)
+        if not readable:
+            return False
+        try:
+            if not connection.recv(4096):
+                return True
+        except ConnectionResetError:
+            return True
+
+
+def main() -> int:
+    args = parse_args()
+
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listener:
+        listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        listener.bind(('127.0.0.1', args.port))
+        listener.listen(5)
+        print(f'Listening on port {args.port}', flush=True)
+
+        # Readiness probes connect and close without sending a request, so keep
+        # accepting connections until a request arrives.
+        while True:
+            connection, _ = listener.accept()
+            with connection:
+                if not read_request(connection):
+                    print('Connection closed before the request was complete.', flush=True)
+                    continue
+                if wait_for_abort(connection, args.delay):
+                    print(ABORT_DETECTED, flush=True)
+                    return 0
+                print(ABORT_NOT_DETECTED, flush=True)
+                connection.sendall(RESPONSE)
+                return 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())

tests/gold_tests/cache/client_abort_before_response.test.py

@@ -0,0 +1,135 @@
+'''
+Verify origin connection handling when a client aborts before the origin responds.
+'''
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you under the Apache License, Version 2.0 (the
+#  "License"); you may not use this file except in compliance
+#  with the License.  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+import os
+
+Test.Summary = __doc__
+Test.ContinueOnFail = True
+
+ORIGIN_SCRIPT = os.path.join(Test.TestDirectory, 'abort_detecting_origin.py')
+
+# The origin waits this long before responding. The client gives up well before
+# that, so the origin is still waiting when the client aborts.
+ORIGIN_DELAY_SECONDS = 6
+CLIENT_TIMEOUT_SECONDS = 2
+
+# How long the test run waits for the origin to reach a conclusion about the
+# connection after the client aborts.
+ORIGIN_WAIT_SECONDS = ORIGIN_DELAY_SECONDS + 3
+
+# These are printed by abort_detecting_origin.py.
+ABORT_DETECTED = 'proxy_closed_connection'
+ABORT_NOT_DETECTED = 'proxy_kept_connection_open'
+
+
+class ClientAbortBeforeResponseTest:
+    '''Verify how ATS treats the origin connection when the client goes away.
+
+    A client abort before the origin sends its response header should close the
+    origin connection when the operator disables half open connections. TLS and
+    HTTP/2 clients cannot half close their connections, so with half open
+    connections configured ATS keeps such transactions alive to fill the cache
+    with the response the client will not receive. See issue #13549.
+    '''
+
+    def __init__(self, name: str, enable_tls: bool, allow_half_open: int, expect_abort: bool):
+        '''
+        :param name: The name to use for the processes of this test case.
+        :param enable_tls: Whether the client talks to ATS over TLS.
+        :param allow_half_open: The proxy.config.http.allow_half_open value to configure.
+        :param expect_abort: Whether ATS is expected to close the origin connection.
+        '''
+        self._name = name
+        self._enable_tls = enable_tls
+        self._allow_half_open = allow_half_open
+        self._expect_abort = expect_abort
+        port_variable = f'{name}_origin_port'
+        Test.GetTcpPort(port_variable)
+        self._origin_port = getattr(Test.Variables, port_variable)
+        self._setup_ts()
+
+    def _setup_ts(self) -> None:
+        self._ts = Test.MakeATSProcess(f'ts_{self._name}', enable_tls=self._enable_tls, enable_cache=True)
+        self._ts.Disk.records_config.update(
+            {
+                'proxy.config.diags.debug.enabled': 1,
+                'proxy.config.diags.debug.tags': 'http',
+                'proxy.config.http.allow_half_open': self._allow_half_open,
+                'proxy.config.http.cache.required_headers': 0,
+            })
+        if self._enable_tls:
+            self._ts.addDefaultSSLFiles()
+            self._ts.Disk.records_config.update(
+                {
+                    'proxy.config.ssl.server.cert.path': self._ts.Variables.SSLDir,
+                    'proxy.config.ssl.server.private_key.path': self._ts.Variables.SSLDir,
+                })
+            self._ts.Disk.ssl_multicert_yaml.AddLines(
+                [
+                    'ssl_multicert:',
+                    '  - dest_ip: "*"',
+                    '    ssl_cert_name: server.pem',
+                    '    ssl_key_name: server.key',
+                ])
+        self._ts.Disk.remap_config.AddLine(f'map / http://127.0.0.1:{self._origin_port}/')
+
+    def _client_url(self) -> str:
+        if self._enable_tls:
+            return f'https://127.0.0.1:{self._ts.Variables.ssl_port}/slow'
+        return f'http://127.0.0.1:{self._ts.Variables.port}/slow'
+
+    def run(self) -> None:
+        scheme = 'https' if self._enable_tls else 'http'
+        tr = Test.AddTestRun(f'Client abort over {scheme} with allow_half_open {self._allow_half_open}')
+
+        origin = tr.Processes.Process(
+            f'origin_{self._name}', f'python3 {ORIGIN_SCRIPT} {self._origin_port} --delay {ORIGIN_DELAY_SECONDS}')
+        origin.Ready = When.PortOpen(self._origin_port)
+        origin.ReturnCode = 0
+
+        if self._expect_abort:
+            expected, unexpected = ABORT_DETECTED, ABORT_NOT_DETECTED
+        else:
+            expected, unexpected = ABORT_NOT_DETECTED, ABORT_DETECTED
+        origin.Streams.All += Testers.ContainsExpression(expected, f'The origin should report {expected}.')
+        origin.Streams.All += Testers.ExcludesExpression(unexpected, f'The origin should not report {unexpected}.')
+
+        # curl gives up before the origin responds, aborting the request. The
+        # sleep afterwards gives the origin time to reach its own conclusion
+        # about the connection.
+        tr.MakeCurlCommandMulti(
+            f'{{curl}} -s -k -o /dev/null --max-time {CLIENT_TIMEOUT_SECONDS} {self._client_url()}; '
+            f'sleep {ORIGIN_WAIT_SECONDS}',
+            ts=self._ts)
+        tr.Processes.Default.ReturnCode = 0
+        tr.Processes.Default.StartBefore(self._ts)
+        tr.Processes.Default.StartBefore(origin)
+        tr.StillRunningAfter = self._ts
+
+
+# The operator disabled half open connections, so ATS should not keep the origin
+# connection open for a client that hung up.
+ClientAbortBeforeResponseTest('http_half_open_disabled', enable_tls=False, allow_half_open=0, expect_abort=True).run()
+
+if not Condition.CurlUsingUnixDomainSocket():
+    ClientAbortBeforeResponseTest('https_half_open_disabled', enable_tls=True, allow_half_open=0, expect_abort=True).run()
+
+    # TLS connections cannot be half closed, but half open connections are
+    # configured, so ATS finishes the fetch to fill the cache.
+    ClientAbortBeforeResponseTest('https_half_open_enabled', enable_tls=True, allow_half_open=1, expect_abort=False).run()