Hard-enforce max_active_streams_in at HTTP/2 stream creation

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

@@ -4180,14 +4180,43 @@ HTTP/2 Configuration
    This is used when :ts:cv:`proxy.config.http2.max_active_streams_in` is set
    larger than ``0``.
 
-.. ts:cv:: CONFIG proxy.config.http2.max_active_streams_in INT 0
+.. ts:cv:: CONFIG proxy.config.http2.max_active_streams_in INT 200000
    :reloadable:
 
-   Limits the maximum number of connection wide active streams.
-   When connection wide active streams are larger than this value,
+   Limits the maximum number of process-wide active inbound streams.
+   When the process-wide active stream count reaches this value,
    SETTINGS_MAX_CONCURRENT_STREAMS will be reduced to
    :ts:cv:`proxy.config.http2.min_concurrent_streams_in`.
-   To disable, set to zero (``0``).
+   To disable, set to zero (``0``). The default bounds worst-case
+   buffered response memory while staying well above any realistic
+   legitimate stream count; size it to your available memory budget.
+
+   See :ts:cv:`proxy.config.http2.max_active_streams_policy_in` to switch
+   from this advisory behavior to hard refusal of new streams once the
+   limit is reached.
+
+.. ts:cv:: CONFIG proxy.config.http2.max_active_streams_policy_in INT 0
+   :reloadable:
+
+   Selects how :ts:cv:`proxy.config.http2.max_active_streams_in` is
+   enforced for inbound HTTP/2 streams.
+
+   ===== ===================================================================
+   Value Description
+   ===== ===================================================================
+   ``0`` When the limit is reached, the advertised
+         ``SETTINGS_MAX_CONCURRENT_STREAMS`` is reduced to
+         :ts:cv:`proxy.config.http2.min_concurrent_streams_in` for new
+         connections. Already-admitted streams are not refused.
+   ``1`` New inbound streams are refused with ``REFUSED_STREAM`` once the
+         global active-stream count reaches the limit. The advertised
+         ``SETTINGS_MAX_CONCURRENT_STREAMS`` is left unchanged and
+         :ts:cv:`proxy.config.http2.min_concurrent_streams_in` is not
+         applied.
+   ===== ===================================================================
+
+   Has no effect when :ts:cv:`proxy.config.http2.max_active_streams_in`
+   is ``0``.
 
 .. ts:cv:: CONFIG proxy.config.http2.initial_window_size_in INT 65535
    :reloadable:

mgmt/RecordsConfig.cc

@@ -1371,7 +1371,9 @@ static const RecordElement RecordsConfig[] =
   ,
   {RECT_CONFIG, "proxy.config.http2.min_concurrent_streams_in", RECD_INT, "10", RECU_DYNAMIC, RR_NULL, RECC_STR, "^[0-9]+$", RECA_NULL}
   ,
-  {RECT_CONFIG, "proxy.config.http2.max_active_streams_in", RECD_INT, "0", RECU_DYNAMIC, RR_NULL, RECC_STR, "^[0-9]+$", RECA_NULL}
+  {RECT_CONFIG, "proxy.config.http2.max_active_streams_in", RECD_INT, "200000", RECU_DYNAMIC, RR_NULL, RECC_STR, "^[0-9]+$", RECA_NULL}
+  ,
+  {RECT_CONFIG, "proxy.config.http2.max_active_streams_policy_in", RECD_INT, "0", RECU_DYNAMIC, RR_NULL, RECC_STR, "^[0-1]$", RECA_NULL}
   ,
   {RECT_CONFIG, "proxy.config.http2.initial_window_size_in", RECD_INT, "65535", RECU_DYNAMIC, RR_NULL, RECC_STR, "^[0-9]+$", RECA_NULL}
   ,

proxy/http2/HTTP2.cc

@@ -92,6 +92,7 @@ static const char *const HTTP2_STAT_MAX_CONCURRENT_STREAMS_EXCEEDED_IN_NAME =
   "proxy.process.http2.max_concurrent_streams_exceeded_in";
 static const char *const HTTP2_STAT_MAX_CONCURRENT_STREAMS_EXCEEDED_OUT_NAME =
   "proxy.process.http2.max_concurrent_streams_exceeded_out";
+static const char *const HTTP2_STAT_MAX_ACTIVE_STREAMS_EXCEEDED_IN_NAME = "proxy.process.http2.max_active_streams_exceeded_in";
 
 union byte_pointer {
   byte_pointer(void *p) : ptr(p) {}
@@ -802,7 +803,8 @@ http2_decode_header_blocks(HTTPHdr *hdr, const uint8_t *buf_start, const uint32_
 // Initialize this subsystem with librecords configs (for now)
 uint32_t Http2::max_concurrent_streams_in          = 100;
 uint32_t Http2::min_concurrent_streams_in          = 10;
-uint32_t Http2::max_active_streams_in              = 0;
+uint32_t Http2::max_active_streams_in              = 200000;
+uint32_t Http2::max_active_streams_policy_in       = 0;
 bool Http2::throttling                             = false;
 uint32_t Http2::stream_priority_enabled            = 0;
 uint32_t Http2::initial_window_size                = 65535;
@@ -839,6 +841,7 @@ Http2::init()
   REC_EstablishStaticConfigInt32U(max_concurrent_streams_in, "proxy.config.http2.max_concurrent_streams_in");
   REC_EstablishStaticConfigInt32U(min_concurrent_streams_in, "proxy.config.http2.min_concurrent_streams_in");
   REC_EstablishStaticConfigInt32U(max_active_streams_in, "proxy.config.http2.max_active_streams_in");
+  REC_EstablishStaticConfigInt32U(max_active_streams_policy_in, "proxy.config.http2.max_active_streams_policy_in");
   REC_EstablishStaticConfigInt32U(stream_priority_enabled, "proxy.config.http2.stream_priority_enabled");
   REC_EstablishStaticConfigInt32U(initial_window_size, "proxy.config.http2.initial_window_size_in");
   REC_EstablishStaticConfigInt32U(max_frame_size, "proxy.config.http2.max_frame_size");
@@ -937,6 +940,8 @@ Http2::init()
                      static_cast<int>(HTTP2_STAT_MAX_CONCURRENT_STREAMS_EXCEEDED_IN), RecRawStatSyncSum);
   RecRegisterRawStat(http2_rsb, RECT_PROCESS, HTTP2_STAT_MAX_CONCURRENT_STREAMS_EXCEEDED_OUT_NAME, RECD_INT, RECP_PERSISTENT,
                      static_cast<int>(HTTP2_STAT_MAX_CONCURRENT_STREAMS_EXCEEDED_OUT), RecRawStatSyncSum);
+  RecRegisterRawStat(http2_rsb, RECT_PROCESS, HTTP2_STAT_MAX_ACTIVE_STREAMS_EXCEEDED_IN_NAME, RECD_INT, RECP_PERSISTENT,
+                     static_cast<int>(HTTP2_STAT_MAX_ACTIVE_STREAMS_EXCEEDED_IN), RecRawStatSyncSum);
 
   http2_init();
 }

proxy/http2/HTTP2.h

@@ -109,6 +109,7 @@ enum {
   HTTP2_STAT_INSUFFICIENT_AVG_WINDOW_UPDATE,
   HTTP2_STAT_MAX_CONCURRENT_STREAMS_EXCEEDED_IN,
   HTTP2_STAT_MAX_CONCURRENT_STREAMS_EXCEEDED_OUT,
+  HTTP2_STAT_MAX_ACTIVE_STREAMS_EXCEEDED_IN,
 
   HTTP2_N_STATS // Terminal counter, NOT A STAT INDEX.
 };
@@ -386,6 +387,7 @@ class Http2
   static uint32_t max_concurrent_streams_in;
   static uint32_t min_concurrent_streams_in;
   static uint32_t max_active_streams_in;
+  static uint32_t max_active_streams_policy_in;
   static bool throttling;
   static uint32_t stream_priority_enabled;
   static uint32_t initial_window_size;

proxy/http2/Http2ConnectionState.cc

@@ -407,6 +407,17 @@ rcv_headers_frame(Http2ConnectionState &cstate, const Http2Frame &frame)
 
     // Set up the State Machine
     if (!empty_request) {
+      // Hard-enforce the global active-streams cap on inbound client streams.
+      if (Http2::max_active_streams_policy_in == 1 && Http2::max_active_streams_in > 0) {
+        int64_t current_streams = 0;
+        RecGetRawStatSum(http2_rsb, HTTP2_STAT_CURRENT_CLIENT_STREAM_COUNT, &current_streams);
+        if (current_streams >= Http2::max_active_streams_in) {
+          HTTP2_INCREMENT_THREAD_DYN_STAT(HTTP2_STAT_MAX_ACTIVE_STREAMS_EXCEEDED_IN, this_ethread());
+          return Http2Error(Http2ErrorClass::HTTP2_ERROR_CLASS_STREAM, Http2ErrorCode::HTTP2_ERROR_REFUSED_STREAM,
+                            "active streams cap reached");
+        }
+      }
+
       SCOPED_MUTEX_LOCK(stream_lock, stream->mutex, this_ethread());
       stream->mark_milestone(Http2StreamMilestone::START_TXN);
       stream->cancel_active_timeout();
@@ -997,6 +1008,17 @@ rcv_continuation_frame(Http2ConnectionState &cstate, const Http2Frame &frame)
     }
 
     // Set up the State Machine
+    // Hard-enforce the global active-streams cap on inbound client streams.
+    if (!stream->has_trailing_header() && Http2::max_active_streams_policy_in == 1 && Http2::max_active_streams_in > 0) {
+      int64_t current_streams = 0;
+      RecGetRawStatSum(http2_rsb, HTTP2_STAT_CURRENT_CLIENT_STREAM_COUNT, &current_streams);
+      if (current_streams >= Http2::max_active_streams_in) {
+        HTTP2_INCREMENT_THREAD_DYN_STAT(HTTP2_STAT_MAX_ACTIVE_STREAMS_EXCEEDED_IN, this_ethread());
+        return Http2Error(Http2ErrorClass::HTTP2_ERROR_CLASS_STREAM, Http2ErrorCode::HTTP2_ERROR_REFUSED_STREAM,
+                          "active streams cap reached");
+      }
+    }
+
     SCOPED_MUTEX_LOCK(stream_lock, stream->mutex, this_ethread());
     stream->mark_milestone(Http2StreamMilestone::START_TXN);
     // This should be fine, need to verify whether we need to replace this with the
@@ -2219,6 +2241,12 @@ Http2ConnectionState::_adjust_concurrent_stream()
     return Http2::max_concurrent_streams_in;
   }
 
+  if (Http2::max_active_streams_policy_in == 1) {
+    // Under hard-enforcement the advertised value is left untouched; new streams
+    // are refused at creation instead of throttling down concurrency.
+    return Http2::max_concurrent_streams_in;
+  }
+
   int64_t current_client_streams = 0;
   RecGetRawStatSum(http2_rsb, HTTP2_STAT_CURRENT_CLIENT_STREAM_COUNT, &current_client_streams);
 

tests/gold_tests/h2/clients/h2_max_active_streams.py

@@ -0,0 +1,173 @@
+#!/usr/bin/env python3
+'''
+HTTP/2 client that opens N concurrent streams to test the global
+active-streams cap and the HPACK dynamic-table sync across refused streams.
+'''
+#  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 socket
+import ssl
+import sys
+from typing import Dict, Optional, Set, Tuple
+
+import hpack
+
+CONNECTION_PREFACE = b'PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n'
+
+FRAME_TYPE_DATA = 0x00
+FRAME_TYPE_HEADERS = 0x01
+FRAME_TYPE_RST_STREAM = 0x03
+FRAME_TYPE_SETTINGS = 0x04
+FRAME_TYPE_GOAWAY = 0x07
+
+FLAG_ACK = 0x01
+FLAG_END_STREAM = 0x01
+FLAG_END_HEADERS = 0x04
+
+ERROR_REFUSED_STREAM = 0x07
+
+# Reusable header value carried on streams >= the cap. Encoding the same
+# (name, value) pair on a later stream forces hpack to emit an indexed
+# reference into the dynamic-table entry added by the earlier encode, which
+# only resolves correctly if ATS decoded the earlier (refused) HEADERS frame.
+PROBE_HEADER_NAME = 'x-test-probe'
+PROBE_HEADER_VALUE = 'shared-probe-value'
+
+
+def make_socket(port: int) -> ssl.SSLSocket:
+    socket.setdefaulttimeout(15)
+    ctx = ssl.create_default_context()
+    ctx.check_hostname = False
+    ctx.verify_mode = ssl.CERT_NONE
+    ctx.set_alpn_protocols(['h2'])
+
+    raw = socket.create_connection(('127.0.0.1', port))
+    tls = ctx.wrap_socket(raw, server_hostname='localhost')
+    if tls.selected_alpn_protocol() != 'h2':
+        raise RuntimeError(f'failed to negotiate h2, got {tls.selected_alpn_protocol()!r}')
+    return tls
+
+
+def make_frame(frame_type: int, flags: int = 0, stream_id: int = 0, payload: bytes = b'') -> bytes:
+    return (len(payload).to_bytes(3, 'big') + bytes([frame_type, flags]) + (stream_id & 0x7fffffff).to_bytes(4, 'big') + payload)
+
+
+def read_exact(sock: ssl.SSLSocket, size: int) -> bytes:
+    chunks = []
+    remaining = size
+    while remaining > 0:
+        chunk = sock.recv(remaining)
+        if not chunk:
+            raise EOFError('socket closed')
+        chunks.append(chunk)
+        remaining -= len(chunk)
+    return b''.join(chunks)
+
+
+def read_frame(sock: ssl.SSLSocket) -> Tuple[int, int, int, bytes]:
+    header = read_exact(sock, 9)
+    length = int.from_bytes(header[0:3], 'big')
+    frame_type = header[3]
+    flags = header[4]
+    stream_id = int.from_bytes(header[5:9], 'big') & 0x7fffffff
+    payload = read_exact(sock, length)
+    return frame_type, flags, stream_id, payload
+
+
+def request_block(encoder: hpack.Encoder, stream_id: int, include_probe: bool) -> bytes:
+    headers = [
+        (':method', 'GET'),
+        (':scheme', 'https'),
+        (':authority', 'www.example.com'),
+        (':path', f'/stream/{stream_id}'),
+        ('uuid', f'max-active-streams-{stream_id}'),
+    ]
+    if include_probe:
+        headers.append((PROBE_HEADER_NAME, PROBE_HEADER_VALUE))
+    return encoder.encode(headers)
+
+
+def run(port: int, num_streams: int, probe_from: Optional[int]) -> int:
+    """Open @a num_streams concurrent streams.
+
+    Returns 0 on success.
+
+    Streams whose id is >= @a probe_from carry a fixed (name, value) header
+    so the second and later occurrences are encoded as indexed references
+    into the dynamic table established by the first occurrence. If ATS
+    fails to keep its decoder dynamic table in sync after refusing a
+    stream, the indexed reference fails to resolve and ATS sends a
+    COMPRESSION_ERROR GOAWAY, which this script reports as a failure.
+    """
+
+    stream_ids = [1 + 2 * i for i in range(num_streams)]
+    encoder = hpack.Encoder()
+    with make_socket(port) as sock:
+        sock.sendall(CONNECTION_PREFACE)
+        sock.sendall(make_frame(FRAME_TYPE_SETTINGS))
+        for sid in stream_ids:
+            include_probe = probe_from is not None and sid >= probe_from
+            block = request_block(encoder, sid, include_probe)
+            sock.sendall(make_frame(FRAME_TYPE_HEADERS, FLAG_END_HEADERS | FLAG_END_STREAM, sid, block))
+
+        ended: Set[int] = set()
+        statuses: Dict[int, str] = {}
+        try:
+            while ended != set(stream_ids):
+                frame_type, flags, stream_id, payload = read_frame(sock)
+                if frame_type == FRAME_TYPE_SETTINGS and not (flags & FLAG_ACK):
+                    sock.sendall(make_frame(FRAME_TYPE_SETTINGS, FLAG_ACK, 0))
+                    continue
+                if frame_type == FRAME_TYPE_GOAWAY:
+                    error_code = int.from_bytes(payload[4:8], 'big')
+                    print(f'GOAWAY error_code={error_code}')
+                    return 1
+                if stream_id in stream_ids:
+                    if frame_type == FRAME_TYPE_RST_STREAM:
+                        error_code = int.from_bytes(payload[0:4], 'big')
+                        statuses[stream_id] = f'rst:{error_code}'
+                        print(f'stream {stream_id}: RST_STREAM error_code={error_code}')
+                        ended.add(stream_id)
+                    elif frame_type in (FRAME_TYPE_DATA, FRAME_TYPE_HEADERS) and (flags & FLAG_END_STREAM):
+                        statuses[stream_id] = 'end_stream'
+                        print(f'stream {stream_id}: END_STREAM')
+                        ended.add(stream_id)
+        except (EOFError, socket.timeout) as exc:
+            print(f'socket terminated before all streams ended: {exc}', file=sys.stderr)
+            return 1
+
+    for sid, status in sorted(statuses.items()):
+        print(f'final stream {sid}: {status}')
+    return 0
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument('port', type=int, help='ATS TLS port')
+    parser.add_argument('--streams', type=int, default=4, help='number of concurrent streams to open')
+    parser.add_argument(
+        '--probe-from',
+        type=int,
+        default=None,
+        help='lowest stream id (odd) to carry the shared probe header; omit to disable probe')
+    args = parser.parse_args()
+    return run(args.port, args.streams, args.probe_from)
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())