From f09c345910007db50699ec3cbc4cf8251f36ada2 Mon Sep 17 00:00:00 2001 From: bneradt Date: Tue, 15 Sep 2026 18:03:13 -0500 Subject: [PATCH] Fix HTTP flow control policy reload HTTP/2 flow-control changes were reported as reloadable but had no effect because connections kept startup-only policy values. The reload callbacks also retained pointers to expired stack variables. This patch validates startup and runtime policy values and publishes them through persistent atomic storage for both connection directions. The regression coverage checks policy transitions on fresh connections through live updates and records.yaml reloads. Fixes: #13694 Co-authored-by: Codex Astra Medium --- include/proxy/http2/HTTP2.h | 50 +++--- src/proxy/http2/HTTP2.cc | 72 +++++---- src/proxy/http2/Http2ConnectionState.cc | 4 +- .../h2/http2_flow_control_reload.py | 142 ++++++++++++++++++ .../h2/http2_flow_control_reload.test.py | 70 +++++++++ 5 files changed, 281 insertions(+), 57 deletions(-) create mode 100644 tests/gold_tests/h2/http2_flow_control_reload.py create mode 100644 tests/gold_tests/h2/http2_flow_control_reload.test.py diff --git a/include/proxy/http2/HTTP2.h b/include/proxy/http2/HTTP2.h index 370bbbef929..3612aab2692 100644 --- a/include/proxy/http2/HTTP2.h +++ b/include/proxy/http2/HTTP2.h @@ -31,6 +31,8 @@ #include "tsutil/Metrics.h" +#include + using ts::Metrics; class HTTPHdr; @@ -411,30 +413,30 @@ enum class Http2FlowControlPolicy { class Http2 { public: - 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_in; - static Http2FlowControlPolicy flow_control_policy_in; - static uint32_t max_frame_size; - static uint32_t header_table_size; - static uint32_t max_header_list_size; - static uint32_t accept_no_activity_timeout; - static uint32_t no_activity_timeout_in; - static uint32_t active_timeout_in; - static uint32_t incomplete_header_timeout_in; - static uint32_t push_diary_size; - static uint32_t zombie_timeout_in; - - static uint32_t max_concurrent_streams_out; - static uint32_t min_concurrent_streams_out; - static uint32_t max_active_streams_out; - static uint32_t no_activity_timeout_out; - static uint32_t initial_window_size_out; - static Http2FlowControlPolicy flow_control_policy_out; + 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_in; + static std::atomic flow_control_policy_in; + static uint32_t max_frame_size; + static uint32_t header_table_size; + static uint32_t max_header_list_size; + static uint32_t accept_no_activity_timeout; + static uint32_t no_activity_timeout_in; + static uint32_t active_timeout_in; + static uint32_t incomplete_header_timeout_in; + static uint32_t push_diary_size; + static uint32_t zombie_timeout_in; + + static uint32_t max_concurrent_streams_out; + static uint32_t min_concurrent_streams_out; + static uint32_t max_active_streams_out; + static uint32_t no_activity_timeout_out; + static uint32_t initial_window_size_out; + static std::atomic flow_control_policy_out; static float stream_error_rate_threshold; static uint32_t stream_error_sampling_threshold; diff --git a/src/proxy/http2/HTTP2.cc b/src/proxy/http2/HTTP2.cc index e0e0d62d099..aee44d2d9a9 100644 --- a/src/proxy/http2/HTTP2.cc +++ b/src/proxy/http2/HTTP2.cc @@ -45,6 +45,28 @@ struct Http2HeaderName { static VersionConverter hvc; +void +establish_flow_control_policy(const char *name, std::atomic &policy) +{ + auto update = [](const char *name, RecDataT type, RecData data, void *cookie) -> int { + ink_assert(type == RECD_INT); + RecInt value = data.rec_int; + + if (value < 0 || value > 2) { + Error("Invalid value for %s: %" PRId64, name, value); + value = 0; + } + static_cast *>(cookie)->store(static_cast(value), + std::memory_order_relaxed); + return REC_ERR_OKAY; + }; + RecData data; + + RecRegisterConfigUpdateCb(name, update, &policy); + data.rec_int = RecGetRecordInt(name).value_or(0); + update(name, RECD_INT, data, &policy); +} + } // namespace // Statistics @@ -460,17 +482,17 @@ 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 = 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_in = 65535; -Http2FlowControlPolicy Http2::flow_control_policy_in = Http2FlowControlPolicy::STATIC_SESSION_AND_STATIC_STREAM; -uint32_t Http2::max_frame_size = 16384; -uint32_t Http2::header_table_size = 4096; -uint32_t Http2::max_header_list_size = 4294967295; +uint32_t Http2::max_concurrent_streams_in = 100; +uint32_t Http2::min_concurrent_streams_in = 10; +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_in = 65535; +std::atomic Http2::flow_control_policy_in{Http2FlowControlPolicy::STATIC_SESSION_AND_STATIC_STREAM}; +uint32_t Http2::max_frame_size = 16384; +uint32_t Http2::header_table_size = 4096; +uint32_t Http2::max_header_list_size = 4294967295; uint32_t Http2::accept_no_activity_timeout = 120; uint32_t Http2::no_activity_timeout_in = 120; @@ -479,12 +501,12 @@ uint32_t Http2::incomplete_header_timeout_in = 10; uint32_t Http2::push_diary_size = 256; uint32_t Http2::zombie_timeout_in = 0; -uint32_t Http2::max_concurrent_streams_out = 100; -uint32_t Http2::min_concurrent_streams_out = 10; -uint32_t Http2::max_active_streams_out = 0; -uint32_t Http2::initial_window_size_out = 65535; -Http2FlowControlPolicy Http2::flow_control_policy_out = Http2FlowControlPolicy::STATIC_SESSION_AND_STATIC_STREAM; -uint32_t Http2::no_activity_timeout_out = 120; +uint32_t Http2::max_concurrent_streams_out = 100; +uint32_t Http2::min_concurrent_streams_out = 10; +uint32_t Http2::max_active_streams_out = 0; +uint32_t Http2::initial_window_size_out = 65535; +std::atomic Http2::flow_control_policy_out{Http2FlowControlPolicy::STATIC_SESSION_AND_STATIC_STREAM}; +uint32_t Http2::no_activity_timeout_out = 120; float Http2::stream_error_rate_threshold = 0.1; uint32_t Http2::stream_error_sampling_threshold = 10; @@ -519,22 +541,10 @@ Http2::init() RecEstablishStaticConfigUInt32(stream_priority_enabled, "proxy.config.http2.stream_priority_enabled"); RecEstablishStaticConfigUInt32(initial_window_size_in, "proxy.config.http2.initial_window_size_in"); - uint32_t flow_control_policy_in_int = 0; - RecEstablishStaticConfigUInt32(flow_control_policy_in_int, "proxy.config.http2.flow_control.policy_in"); - if (flow_control_policy_in_int > 2) { - Error("Invalid value for proxy.config.http2.flow_control.policy_in: %d", flow_control_policy_in_int); - flow_control_policy_in_int = 0; - } - flow_control_policy_in = static_cast(flow_control_policy_in_int); + establish_flow_control_policy("proxy.config.http2.flow_control.policy_in", flow_control_policy_in); RecEstablishStaticConfigUInt32(initial_window_size_out, "proxy.config.http2.initial_window_size_out"); - uint32_t flow_control_policy_out_int = 0; - RecEstablishStaticConfigUInt32(flow_control_policy_out_int, "proxy.config.http2.flow_control.policy_out"); - if (flow_control_policy_out_int > 2) { - Error("Invalid value for proxy.config.http2.flow_control.policy_out: %d", flow_control_policy_out_int); - flow_control_policy_out_int = 0; - } - flow_control_policy_out = static_cast(flow_control_policy_out_int); + establish_flow_control_policy("proxy.config.http2.flow_control.policy_out", flow_control_policy_out); RecEstablishStaticConfigUInt32(max_frame_size, "proxy.config.http2.max_frame_size"); RecEstablishStaticConfigUInt32(header_table_size, "proxy.config.http2.header_table_size"); diff --git a/src/proxy/http2/Http2ConnectionState.cc b/src/proxy/http2/Http2ConnectionState.cc index 446dfc4dbdd..5eccce014b6 100644 --- a/src/proxy/http2/Http2ConnectionState.cc +++ b/src/proxy/http2/Http2ConnectionState.cc @@ -1236,9 +1236,9 @@ Http2ConnectionState::_get_configured_flow_control_policy() const { ink_assert(this->session != nullptr); if (this->session->is_outbound()) { - return Http2::flow_control_policy_out; + return Http2::flow_control_policy_out.load(std::memory_order_relaxed); } else { - return Http2::flow_control_policy_in; + return Http2::flow_control_policy_in.load(std::memory_order_relaxed); } } diff --git a/tests/gold_tests/h2/http2_flow_control_reload.py b/tests/gold_tests/h2/http2_flow_control_reload.py new file mode 100644 index 00000000000..69ed9f58454 --- /dev/null +++ b/tests/gold_tests/h2/http2_flow_control_reload.py @@ -0,0 +1,142 @@ +"""Probe HTTP/2 receive windows while reloading flow control policies.""" + +# 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 concurrent.futures +import http.client +import socket +import ssl +import subprocess +import time + +from h2.config import H2Configuration +from h2.connection import H2Connection +from h2.events import PingAckReceived, RequestReceived + + +class WindowProbe: + """Measure ATS receive windows from each side of a fresh connection.""" + + def __init__(self, args: argparse.Namespace) -> None: + self._args = args + self._client_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + self._client_context.check_hostname = False + self._client_context.verify_mode = ssl.CERT_NONE + self._client_context.set_alpn_protocols(['h2']) + self._server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + self._server_context.load_cert_chain(args.cert, args.key) + self._server_context.set_alpn_protocols(['h2']) + + def _window(self, sock: ssl.SSLSocket, is_client: bool) -> int: + assert sock.selected_alpn_protocol() == 'h2', 'Expected HTTP/2' + connection = H2Connection(config=H2Configuration(client_side=is_client)) + connection.initiate_connection() + connection.ping(b'reload!!') + sock.sendall(connection.data_to_send()) + # The PING acknowledgement is a barrier: ATS has sent its initial + # SETTINGS and connection WINDOW_UPDATE before acknowledging our PING. + acknowledged = False + received_request = is_client + while not (acknowledged and received_request): + data = sock.recv(65536) + assert data, 'ATS closed the connection before the probe completed' + for event in connection.receive_data(data): + if isinstance(event, PingAckReceived): + acknowledged = True + elif isinstance(event, RequestReceived): + received_request = True + connection.send_headers(event.stream_id, [(':status', '200'), ('content-length', '0')], end_stream=True) + sock.sendall(connection.data_to_send()) + return connection.outbound_flow_control_window + + def inbound(self) -> int: + with socket.create_connection(('127.0.0.1', self._args.https_port), timeout=5) as raw: + with self._client_context.wrap_socket(raw, server_hostname='localhost') as sock: + return self._window(sock, True) + + def _request(self) -> None: + connection = http.client.HTTPConnection('127.0.0.1', self._args.http_port, timeout=5) + try: + connection.request('GET', '/', headers={'Connection': 'close'}) + response = connection.getresponse() + assert response.status == 200, f'Unexpected origin response: {response.status}' + response.read() + finally: + connection.close() + + def outbound(self) -> int: + with socket.socket() as listener: + listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + listener.bind(('127.0.0.1', self._args.origin_port)) + listener.listen(1) + listener.settimeout(5) + with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor: + request = executor.submit(self._request) + raw, _ = listener.accept() + with raw: + raw.settimeout(5) + with self._server_context.wrap_socket(raw, server_side=True) as sock: + window = self._window(sock, False) + request.result(timeout=5) + return window + + def expect(self, inbound_policy: int, outbound_policy: int, wait: bool = False) -> None: + expected = tuple(65535 if policy == 0 else 6553500 for policy in (inbound_policy, outbound_policy)) + deadline = time.monotonic() + (20 if wait else 0) + while True: + actual = (self.inbound(), self.outbound()) + if actual == expected: + print(f'policies in={inbound_policy}, out={outbound_policy}: windows={actual}', flush=True) + return + if time.monotonic() >= deadline: + raise AssertionError( + f'policies in={inbound_policy}, out={outbound_policy}: expected windows {expected}, got {actual}') + time.sleep(0.25) + + def run(self) -> None: + self.expect(0, 0) + # Reset to policy 0 before policy 2 so the large window proves that + # both nonzero policy values actually take effect. Probe the opposite + # direction too, so crossed or shared callbacks cannot pass. + for direction in ('in', 'out'): + record = f'proxy.config.http2.flow_control.policy_{direction}' + for policy in (1, 0, 2, 0): + subprocess.run(['traffic_ctl', 'config', 'set', record, str(policy)], check=True, timeout=10) + self.expect(policy if direction == 'in' else 0, policy if direction == 'out' else 0, wait=True) + # Exercise the records.yaml reload path as well as live config set. + for policy in (1, 0): + for direction in ('in', 'out'): + record = f'proxy.config.http2.flow_control.policy_{direction}' + subprocess.run(['traffic_ctl', 'config', 'set', record, str(policy), '--cold'], check=True, timeout=10) + subprocess.run(['traffic_ctl', 'config', 'reload', '--monitor'], check=True, timeout=30) + self.expect(policy, policy, wait=True) + print('PASS: both flow control policies reloaded', flush=True) + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--https-port', type=int, required=True) + parser.add_argument('--http-port', type=int, required=True) + parser.add_argument('--origin-port', type=int, required=True) + parser.add_argument('--cert', required=True) + parser.add_argument('--key', required=True) + WindowProbe(parser.parse_args()).run() + + +if __name__ == '__main__': + main() diff --git a/tests/gold_tests/h2/http2_flow_control_reload.test.py b/tests/gold_tests/h2/http2_flow_control_reload.test.py new file mode 100644 index 00000000000..7f6f19f4b72 --- /dev/null +++ b/tests/gold_tests/h2/http2_flow_control_reload.test.py @@ -0,0 +1,70 @@ +"""Verify HTTP/2 flow control policies change without restarting ATS.""" + +# 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 +from ports import get_port + +Test.Summary = __doc__ + + +class FlowControlReloadTest: + """Observe both connection receive windows across runtime policy changes.""" + + def __init__(self) -> None: + self._ts = self._configure_trafficserver() + self._configure_probe() + + def _configure_trafficserver(self) -> 'Process': + ts = Test.MakeATSProcess('ts', enable_tls=True, enable_cache=False) + ts.addDefaultSSLFiles() + get_port(ts, 'origin_port') + ts.Disk.records_config.update( + { + 'proxy.config.ssl.server.cert.path': ts.Variables.SSLDir, + 'proxy.config.ssl.server.private_key.path': ts.Variables.SSLDir, + 'proxy.config.ssl.client.verify.server.policy': 'PERMISSIVE', + 'proxy.config.ssl.client.alpn_protocols': 'h2', + 'proxy.config.http2.initial_window_size_in': 65535, + 'proxy.config.http2.initial_window_size_out': 65535, + 'proxy.config.http2.max_concurrent_streams_in': 100, + 'proxy.config.http2.max_concurrent_streams_out': 100, + 'proxy.config.http2.flow_control.policy_in': 0, + 'proxy.config.http2.flow_control.policy_out': 0, + }) + ts.Disk.ssl_multicert_yaml.AddLines( + ['ssl_multicert:', ' - dest_ip: "*"', ' ssl_cert_name: server.pem', ' ssl_key_name: server.key']) + ts.Disk.remap_config.AddLine(f'map / https://127.0.0.1:{ts.Variables.origin_port}') + return ts + + def _configure_probe(self) -> None: + tr = Test.AddTestRun('Reload inbound and outbound flow control policies on the same ATS process') + tr.Processes.Default.StartBefore(self._ts) + tr.Processes.Default.Env = self._ts.Env + tr.Processes.Default.Command = ( + f'{sys.executable} {Test.TestDirectory}/http2_flow_control_reload.py ' + f'--https-port {self._ts.Variables.ssl_port} --http-port {self._ts.Variables.port} ' + f'--origin-port {self._ts.Variables.origin_port} ' + f'--cert {self._ts.Variables.SSLDir}/server.pem --key {self._ts.Variables.SSLDir}/server.key') + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.TimeOut = 180 + tr.Processes.Default.Streams.stdout = Testers.ContainsExpression( + 'PASS: both flow control policies reloaded', 'Both policies must affect new connections without a restart') + tr.StillRunningAfter = self._ts + + +FlowControlReloadTest()