HTTP/2: Wait END_STREAM flag on half-closed(local) stream state (#12257)

masaori335 · web-flow · commit c3a40a791118 · 2025-06-25T10:00:35.000+09:00
* HTTP/2: Wait END_STREAM flag on half-closed(local) stream state

* Actively close stream in half-closed(local) state

* Add AuTest

* Fix format of AuTests

* Address comments
diff --git a/src/proxy/http2/Http2ConnectionState.cc b/src/proxy/http2/Http2ConnectionState.cc
@@ -175,9 +175,15 @@ Http2ConnectionState::rcv_data_frame(const Http2Frame &frame)
 
     // Pure END_STREAM
     if (payload_length == 0) {
+      if (stream->get_state() == Http2StreamState::HTTP2_STREAM_STATE_CLOSED) {
+        stream->initiating_close();
+        return Http2Error(Http2ErrorClass::HTTP2_ERROR_CLASS_NONE);
+      }
+
       if (stream->is_read_enabled()) {
         stream->signal_read_event(VC_EVENT_READ_COMPLETE);
       }
+
       return Http2Error(Http2ErrorClass::HTTP2_ERROR_CLASS_NONE);
     }
   } else {
@@ -2380,13 +2386,16 @@ Http2ConnectionState::send_data_frames(Http2Stream *stream)
 
     if (result == Http2SendDataFrameResult::DONE) {
       if (!stream->is_outbound_connection()) {
-        // Delete a stream immediately
-        // TODO its should not be deleted for a several time to handling
-        // RST_STREAM and WINDOW_UPDATE.
-        // See 'closed' state written at [RFC 7540] 5.1.
-        Http2StreamDebug(this->session, stream->get_id(), "Shutdown stream");
-        stream->signal_write_event(VC_EVENT_WRITE_COMPLETE);
-        stream->do_io_close();
+        if (stream->get_state() == Http2StreamState::HTTP2_STREAM_STATE_CLOSED) {
+          // Delete a stream immediately
+          Http2StreamDebug(this->session, stream->get_id(), "Shutdown stream");
+          stream->signal_write_event(VC_EVENT_WRITE_COMPLETE);
+          stream->do_io_close();
+        } else {
+          // This stream waits for the END_STREAM in half-closed (local) state until `http.transaction_no_activity_timeout_in`.
+          // If no frame with END_STREAM is found, ATS actively closes the stream.
+          Http2StreamDebug(this->session, stream->get_id(), "waiting END_STREAM");
+        }
       } else if (stream->is_outbound_connection() && stream->is_write_vio_done()) {
         stream->signal_write_event(VC_EVENT_WRITE_COMPLETE);
       } else {
diff --git a/src/proxy/http2/Http2Stream.cc b/src/proxy/http2/Http2Stream.cc
@@ -31,6 +31,7 @@
 #include "tscore/Diags.h"
 #include "tscore/HTTPVersion.h"
 #include "tscore/ink_assert.h"
+#include "tsutil/DbgCtl.h"
 
 #include <numeric>
 
@@ -562,7 +563,10 @@ Http2Stream::do_io_close(int /* flags */)
     // We only need to do this for the client side since we only need to pass through RST_STREAM
     // from the server. If a client sends a RST_STREAM, we need to keep the server side alive so
     // the background fill can function as intended.
-    if (!this->is_outbound_connection() && this->is_state_writeable()) {
+    //
+    // In the half-closed(local) state, server can close stream actively by sending RST_STREAM frame
+    if (!this->is_outbound_connection() &&
+        (this->is_state_writeable() || _state == Http2StreamState::HTTP2_STREAM_STATE_HALF_CLOSED_LOCAL)) {
       this->get_connection_state().send_rst_stream_frame(_id, Http2ErrorCode::HTTP2_ERROR_NO_ERROR);
     }
 
@@ -590,7 +594,7 @@ Http2Stream::transaction_done()
   SCOPED_MUTEX_LOCK(lock, this->mutex, this_ethread());
   super::transaction_done();
 
-  if (!closed) {
+  if (!closed && _state == Http2StreamState::HTTP2_STREAM_STATE_CLOSED) {
     do_io_close(); // Make sure we've been closed.  If we didn't close the _proxy_ssn session better still be open
   }
   Http2ConnectionState &state = this->get_connection_state();
@@ -1081,11 +1085,20 @@ Http2Stream::clear_io_events()
   }
 }
 
-//  release and do_io_close are the same for the HTTP/2 protocol
+/**
+  Callback from HttpSM
+
+  release and do_io_close are the same for the HTTP/2 protocol
+ */
 void
 Http2Stream::release()
 {
-  this->do_io_close();
+  if (_state == Http2StreamState::HTTP2_STREAM_STATE_CLOSED) {
+    this->do_io_close();
+    return;
+  }
+
+  Http2StreamDebug("Delaying do_io_close() until stream is in the closed state");
 }
 
 void
diff --git a/tests/gold_tests/h2/clients/h2empty_data_frame.py b/tests/gold_tests/h2/clients/h2empty_data_frame.py
@@ -0,0 +1,128 @@
+#!/usr/bin/env python3
+'''
+HTTP/2 client that sends empty DATA frame with END_STREAM flag
+'''
+#  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 socket
+import ssl
+
+import h2.connection
+import h2.events
+
+import argparse
+
+
+# TODO: cleanup with other HTTP/2 clients (h2active_timeout.py and h2client.py)
+def get_socket(port: int) -> socket.socket:
+    """Create a TLS-wrapped socket.
+
+    :param port: The port to connect to.
+
+    :returns: A TLS-wrapped socket.
+    """
+
+    SERVER_NAME = 'localhost'
+    SERVER_PORT = port
+
+    # generic socket and ssl configuration
+    socket.setdefaulttimeout(15)
+
+    # Configure an ssl client side context which will not check the server's certificate.
+    ctx = ssl.create_default_context()
+    ctx.check_hostname = False
+    ctx.verify_mode = ssl.CERT_NONE
+    ctx.set_alpn_protocols(['h2'])
+
+    # open a socket to the server and initiate TLS/SSL
+    tls_socket = socket.create_connection((SERVER_NAME, SERVER_PORT))
+    tls_socket = ctx.wrap_socket(tls_socket, server_hostname=SERVER_NAME)
+    return tls_socket
+
+
+def make_request(port: int, path: str, n: int) -> None:
+    """Establish an HTTP/2 connection and send a request.
+
+    :param port: The port to connect to.
+    :param path: The path to request.
+    :param n: Number of streams to open.
+    """
+
+    tls_socket = get_socket(port)
+
+    h2_connection = h2.connection.H2Connection()
+    h2_connection.initiate_connection()
+    tls_socket.sendall(h2_connection.data_to_send())
+
+    headers = [
+        (':method', 'GET'),
+        (':path', path),
+        (':authority', 'localhost'),
+        (':scheme', 'https'),
+    ]
+
+    for stream_id in range(1, n * 2, 2):
+        h2_connection.send_headers(stream_id, headers, end_stream=False)
+        h2_connection.send_data(stream_id, b'', end_stream=True)
+
+    tls_socket.sendall(h2_connection.data_to_send())
+
+    # keep reading from server
+    terminated = False
+    error_code = 0
+    while not terminated:
+        # read raw data from the socket
+        data = tls_socket.recv(65536 * 1024)
+        if not data:
+            break
+
+        # feed raw data into h2, and process resulting events
+        events = h2_connection.receive_data(data)
+        for event in events:
+            print(event)
+            if isinstance(event, h2.events.ConnectionTerminated):
+                if not event.error_code == 0:
+                    error_code = event.error_code
+                terminated = True
+
+        # send any pending data to the server
+        tls_socket.sendall(h2_connection.data_to_send())
+
+    # tell the server we are closing the h2 connection
+    h2_connection.close_connection()
+    tls_socket.sendall(h2_connection.data_to_send())
+
+    # close the socket
+    tls_socket.close()
+
+    if error_code != 0:
+        exit(1)
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("port", type=int, help="Port to use")
+    parser.add_argument("path", help="The path to request")
+    parser.add_argument("-n", type=int, default=1, help="Number of streams to open")
+
+    args = parser.parse_args()
+
+    make_request(args.port, args.path, args.n)
+
+
+if __name__ == '__main__':
+    main()
diff --git a/tests/gold_tests/h2/http2_empty_data_frame.test.py b/tests/gold_tests/h2/http2_empty_data_frame.test.py
@@ -0,0 +1,71 @@
+#  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 sys
+
+Test.Sumary = '''
+Verify Empty DATA Frame Handling
+'''
+
+
+class Http2EmptyDataFrameTest:
+
+    def __init__(self):
+        self.__setupOriginServer()
+        self.__setupTS()
+        self.__setupClient()
+
+    def __setupOriginServer(self):
+        self._server = Test.MakeHttpBinServer("httpbin")
+
+    def __setupTS(self):
+        self._ts = Test.MakeATSProcess(f"ts", enable_tls=True, enable_cache=True)
+        self._ts.addDefaultSSLFiles()
+        self._ts.Disk.records_config.update(
+            {
+                'proxy.config.diags.debug.enabled': 1,
+                'proxy.config.diags.debug.tags': 'http2',
+                'proxy.config.ssl.server.cert.path': f"{self._ts.Variables.SSLDir}",
+                'proxy.config.ssl.server.private_key.path': f"{self._ts.Variables.SSLDir}",
+                'proxy.config.http.insert_response_via_str': 2,
+                'proxy.config.http2.active_timeout_in': 3,
+                'proxy.config.http2.stream_error_rate_threshold': 0.1  # default
+            })
+        self._ts.Disk.remap_config.AddLine(f"map / http://127.0.0.1:{self._server.Variables.Port}")
+        self._ts.Disk.ssl_multicert_config.AddLine('dest_ip=* ssl_cert_name=server.pem ssl_key_name=server.key')
+
+    def __setupClient(self):
+        self._ts.Setup.CopyAs("clients/h2empty_data_frame.py", Test.RunDirectory)
+
+    def run(self):
+        tr = Test.AddTestRun("warm-up cache")
+
+        tr.Processes.Default.StartBefore(self._ts)
+        tr.Processes.Default.StartBefore(self._server)
+
+        # warm up the cache
+        tr.Processes.Default.Command = f"{sys.executable} h2empty_data_frame.py {self._ts.Variables.ssl_port} /cache/10 -n 1"
+        tr.Processes.Default.ReturnCode = 0
+
+        # verify 20 streams doesn't hit `proxy.config.http2.stream_error_rate_threshold`
+        tr = Test.AddTestRun("open 20 streams")
+        tr.Processes.Default.Command = f"{sys.executable} h2empty_data_frame.py {self._ts.Variables.ssl_port} /cache/10 -n 20"
+        tr.Processes.Default.ReturnCode = 0
+
+        tr.StillRunningAfter = self._ts
+
+
+Http2EmptyDataFrameTest().run()
