From 381ce8ba475106236af1513f650c12ba52e53abb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miguel=20Rubinos=20Rodr=C3=ADguez?= Date: Mon, 21 Sep 2026 11:18:58 +0200 Subject: [PATCH 1/4] chore: nhttp_lib pin main branch --- CHANGELOG.md | 6 ++++++ rebar.config | 4 +++- rebar.lock | 7 ++++--- src/nhttp.app.src | 2 +- 4 files changed, 14 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0c1e02a..175b349 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## unreleased + +### Changed + +- Updates `nhttp_lib` reference + ## [1.1.1] - 2026-09-10 ### Changed diff --git a/rebar.config b/rebar.config index e3dfd44..fff255c 100644 --- a/rebar.config +++ b/rebar.config @@ -9,7 +9,9 @@ {deps, [ {opentelemetry_api, "1.4.0"}, {opentelemetry_api_experimental, "0.5.1"}, - {nhttp_lib, "1.1.1"}, + {nhttp_lib, + {git, "https://github.com/nomasystems/nhttp_lib.git", + {ref, "14e36f584e70022a240135b961d203e504693343"}}}, {nquic, "1.0.3"} ]}. diff --git a/rebar.lock b/rebar.lock index 02bef64..de4fbbe 100644 --- a/rebar.lock +++ b/rebar.lock @@ -1,5 +1,8 @@ {"1.2.0", -[{<<"nhttp_lib">>,{pkg,<<"nhttp_lib">>,<<"1.1.1">>},0}, +[{<<"nhttp_lib">>, + {git,"https://github.com/nomasystems/nhttp_lib.git", + {ref,"14e36f584e70022a240135b961d203e504693343"}}, + 0}, {<<"nquic">>,{pkg,<<"nquic">>,<<"1.0.3">>},0}, {<<"opentelemetry_api">>,{pkg,<<"opentelemetry_api">>,<<"1.4.0">>},0}, {<<"opentelemetry_api_experimental">>, @@ -7,12 +10,10 @@ 0}]}. [ {pkg_hash,[ - {<<"nhttp_lib">>, <<"B76860C6571AC8F3B2890B5FEFB5A4F593CBDAEF76A1783C6519D91AAE031904">>}, {<<"nquic">>, <<"1B25FD940CDF4BD7378B360484FA546D7967F60D8421CFDBEDD0774F6398D1D4">>}, {<<"opentelemetry_api">>, <<"63CA1742F92F00059298F478048DFB826F4B20D49534493D6919A0DB39B6DB04">>}, {<<"opentelemetry_api_experimental">>, <<"1B5AFACFCBD0834390336C845BC8AE08C8CF0D69BBED72EE53D178798B93E074">>}]}, {pkg_hash_ext,[ - {<<"nhttp_lib">>, <<"DD3790FEF5AC2F4494D8DDD373053069641B753F717F403F38DA3B817FD84F82">>}, {<<"nquic">>, <<"73E9675F9D1C1B2F182A961B3C689778CAD8BA178C63E57C6A23A1C5FDE427F2">>}, {<<"opentelemetry_api">>, <<"3DFBBFAA2C2ED3121C5C483162836C4F9027DEF469C41578AF5EF32589FCFC58">>}, {<<"opentelemetry_api_experimental">>, <<"10297057EADA47267D4F832011BECEF07D25690E6BF91FEBCCFC4E740DBA1A6F">>}]} diff --git a/src/nhttp.app.src b/src/nhttp.app.src index 3d65662..d7ed703 100644 --- a/src/nhttp.app.src +++ b/src/nhttp.app.src @@ -1,6 +1,6 @@ {application, nhttp, [ {description, "HTTP/1.1, HTTP/2, and HTTP/3 server for Erlang/OTP 27+"}, - {vsn, "1.1.1"}, + {vsn, "1.2.0"}, {registered, []}, {applications, [ kernel, From eb1744a64ad13b43abc94050ba5b6b412fe3fbe0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miguel=20Rubinos=20Rodr=C3=ADguez?= Date: Mon, 21 Sep 2026 11:19:48 +0200 Subject: [PATCH 2/4] fix(h2): flush buffered response bodies on a SETTINGS change --- src/nhttp_conn_h2.erl | 4 +++- test/nhttp_conn_h2_lifecycle_SUITE.erl | 21 +++++++++++++++++++++ test/nhttp_test_helpers.erl | 6 ++++++ 3 files changed, 30 insertions(+), 1 deletion(-) diff --git a/src/nhttp_conn_h2.erl b/src/nhttp_conn_h2.erl index 8c96f10..337c0f4 100644 --- a/src/nhttp_conn_h2.erl +++ b/src/nhttp_conn_h2.erl @@ -732,6 +732,8 @@ replenish_recv_window( end, #state{protocol_state = #h2_state{h2_conn = ConnAfter} = H2After} = State1, case nhttp_h2:send_window_update(ConnAfter, StreamId, Size) of + {ok, _UnknownStream, []} -> + State1; {ok, H2Conn2, Frame2} -> nhttp_conn:sock_send(State1, Frame2), State1#state{protocol_state = H2After#h2_state{h2_conn = H2Conn2}}; @@ -936,7 +938,7 @@ handle_h2_event( handle_h2_event(State, {goaway, _LastStreamId, ErrorCode, _DebugData}) -> nhttp_conn_ws_h2:notify_goaway(State, ErrorCode); handle_h2_event(State, {settings, _NewSettings}) -> - State; + flush_stream_buffer(State, 0); handle_h2_event(State, settings_ack) -> State; handle_h2_event(State, {window_update, StreamId, _Increment}) -> diff --git a/test/nhttp_conn_h2_lifecycle_SUITE.erl b/test/nhttp_conn_h2_lifecycle_SUITE.erl index 7b9069a..0a89a9c 100644 --- a/test/nhttp_conn_h2_lifecycle_SUITE.erl +++ b/test/nhttp_conn_h2_lifecycle_SUITE.erl @@ -28,6 +28,7 @@ h2_client_rst_during_stream/1, h2_connection_error_goaway/1, h2_drain_kills_active_worker/1, + h2_settings_window_raise_flushes_buffer/1, h2_streaming_body_too_large/1, h2_sys_messages/1, h2_uri_too_long/1, @@ -43,6 +44,8 @@ ]). -define(BIG_BODY, 200000). +-define(DEFAULT_WINDOW, 65535). +-define(SETTINGS_INITIAL_WINDOW_SIZE, 4). %%%----------------------------------------------------------------------------- %%% CT CALLBACKS @@ -53,6 +56,7 @@ all() -> h2_client_rst_during_stream, h2_connection_error_goaway, h2_drain_kills_active_worker, + h2_settings_window_raise_flushes_buffer, h2_streaming_body_too_large, h2_sys_messages, h2_uri_too_long, @@ -156,6 +160,23 @@ h2_window_exhaustion_then_update(Config) -> nhttp:stop(Pid), ok. +h2_settings_window_raise_flushes_buffer(Config) -> + {ok, Pid, Port} = start(Config, #{}), + {ok, Sock} = nhttp_test_helpers:h2_connect(Port), + ok = nhttp_test_helpers:h2_send_settings(Sock, [{?SETTINGS_INITIAL_WINDOW_SIZE, 0}]), + ?assert(lists:member({settings, 0, <<>>}, nhttp_test_helpers:h2_recv(Sock, 500))), + ok = nhttp_test_helpers:h2_send_request(Sock, 1, <<"/big">>), + Frames = nhttp_test_helpers:h2_recv(Sock, 500), + ?assert(lists:keymember(headers, 1, Frames)), + ?assertEqual(0, stream_data_size(Frames, 1)), + ok = nhttp_test_helpers:h2_send_settings(Sock, [ + {?SETTINGS_INITIAL_WINDOW_SIZE, ?DEFAULT_WINDOW} + ]), + ?assertEqual(?DEFAULT_WINDOW, stream_data_size(nhttp_test_helpers:h2_recv(Sock, 1000), 1)), + ssl:close(Sock), + nhttp:stop(Pid), + ok. + h2_worker_crash_mid_stream(Config) -> {ok, Pid, Port} = start(Config, #{}), {ok, Sock} = nhttp_test_helpers:h2_connect(Port), diff --git a/test/nhttp_test_helpers.erl b/test/nhttp_test_helpers.erl index 2264d62..f503abf 100644 --- a/test/nhttp_test_helpers.erl +++ b/test/nhttp_test_helpers.erl @@ -33,6 +33,7 @@ h2_send_raw/2, h2_send_request/3, h2_send_rst_stream/3, + h2_send_settings/2, h2_send_window_update/3, tcp_connect/1 ]). @@ -206,6 +207,11 @@ h2_send_raw(Sock, Bytes) -> h2_send_rst_stream(Sock, StreamId, ErrorCode) -> ssl:send(Sock, <<4:24, 3, 0, 0:1, StreamId:31, ErrorCode:32>>). +-spec h2_send_settings(ssl:sslsocket(), [{non_neg_integer(), non_neg_integer()}]) -> ok. +h2_send_settings(Sock, Settings) -> + Payload = <<<> || {Id, Value} <- Settings>>, + ssl:send(Sock, <<(byte_size(Payload)):24, 4, 0, 0:1, 0:31, Payload/binary>>). + -spec h2_send_window_update(ssl:sslsocket(), non_neg_integer(), non_neg_integer()) -> ok. h2_send_window_update(Sock, StreamId, Increment) -> ssl:send(Sock, <<4:24, 8, 0, 0:1, StreamId:31, 0:1, Increment:31>>). From 64322dea40f32ff26ddb6f7c15f5403d27935ca5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miguel=20Rubinos=20Rodr=C3=ADguez?= Date: Mon, 21 Sep 2026 13:14:47 +0200 Subject: [PATCH 3/4] feat(h2): add response delay and first-class window and frame options (#14) --- CHANGELOG.md | 5 + README.md | 16 ++ src/nhttp.erl | 13 + src/nhttp_conn.erl | 33 ++- src/nhttp_conn.hrl | 6 +- src/nhttp_conn_h2.erl | 126 ++++++++- src/nhttp_listener.erl | 46 ++++ test/nhttp_h2_credit_policy_SUITE.erl | 317 ++++++++++++++++++++++ test/nhttp_h2_streaming_body_SUITE.erl | 318 ++++++----------------- test/nhttp_listener_validation_SUITE.erl | 74 ++++++ test/nhttp_test_helpers.erl | 178 ++++++++++++- 11 files changed, 861 insertions(+), 271 deletions(-) create mode 100644 test/nhttp_h2_credit_policy_SUITE.erl diff --git a/CHANGELOG.md b/CHANGELOG.md index 175b349..43b0f86 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## unreleased +### Added + +- Listener options `h2_initial_window_size` and `h2_max_frame_size`, aliases of the `h2_settings` keys +- Listener option `h2_response_delay`, a fixed or uniform delay before the response headers of a `{reply, _, _}` result on HTTP/2 + ### Changed - Updates `nhttp_lib` reference diff --git a/README.md b/README.md index b24513b..1e6e032 100644 --- a/README.md +++ b/README.md @@ -81,6 +81,22 @@ HTTP/3 runs over QUIC on a UDP socket. HTTP/1.1 and HTTP/2 run over TCP on their - **Graceful shutdown** with connection draining - **One process per connection**, with HTTP/2 and HTTP/3 streams multiplexed inside it +### HTTP/2 test affordances + +These listener options exist so that a client can be tested against a +slow or stingy HTTP/2 peer. Each one defaults to the current behavior and +changes nothing on HTTP/1.1 or HTTP/3. + +| Option | Default | Effect | +|--------|---------|--------| +| `h2_initial_window_size` | 65535 | Alias of `initial_window_size` in `h2_settings`: the receive window that each new stream grants the peer. | +| `h2_max_frame_size` | 16384 | Alias of `max_frame_size` in `h2_settings`: the largest frame payload the server accepts. | +| `h2_response_delay` | 0 | Milliseconds to hold the response headers of a `{reply, _, _}` result. `{uniform, MinMs, MaxMs}` draws a value per response. | + +An alias must equal the `h2_settings` key when both are present. The +delay applies to `{reply, _, _}` results only. Error responses, producer +streams and WebSocket upgrades go out at once. + ## Documentation [nhttp on HexDocs](https://hexdocs.pm/nhttp) diff --git a/src/nhttp.erl b/src/nhttp.erl index 32a658f..0b7cb65 100644 --- a/src/nhttp.erl +++ b/src/nhttp.erl @@ -167,6 +167,7 @@ RFC 9001 §9.2). header_name/0, header_value/0, headers/0, + h2_response_delay/0, method/0, name/0, @@ -197,6 +198,9 @@ RFC 9001 §9.2). alt_svc => #{ma => non_neg_integer()} | false, backlog => pos_integer(), buffer => pos_integer(), + h2_initial_window_size => 1..16#7fffffff, + h2_max_frame_size => 16#4000..16#ffffff, + h2_response_delay => h2_response_delay(), h2_settings => nhttp_h2:settings(), handler := module(), handler_args => term(), @@ -212,6 +216,15 @@ RFC 9001 §9.2). proxy_protocol => boolean() | proxy_protocol_opts() }. +-doc """ +Delay before the response headers of a `{reply, _, _}` handler result on +HTTP/2, in milliseconds. `{uniform, MinMs, MaxMs}` draws a value in that +range for each response. Error responses, producer streams and WebSocket +upgrades are not delayed. Default 0. +""". +-type h2_response_delay() :: + non_neg_integer() | {uniform, non_neg_integer(), non_neg_integer()}. + -doc """ Connection timeouts (milliseconds, or `infinity`). diff --git a/src/nhttp_conn.erl b/src/nhttp_conn.erl index ebcde77..b4d3ea3 100644 --- a/src/nhttp_conn.erl +++ b/src/nhttp_conn.erl @@ -421,17 +421,34 @@ h1_state_from_opts(Opts) -> body_deadline = maps:get(body_deadline, Timeouts, infinity) }. +-doc """ +Build the HTTP/2 settings the codec advertises. The first-class options +`h2_initial_window_size` and `h2_max_frame_size` are aliases of the +`h2_settings` keys and are resolved here, the only place the codec is built. +""". +-spec h2_settings_from_opts(nhttp:opts()) -> nhttp_h2:settings(). +h2_settings_from_opts(Opts) -> + Settings = maps:fold( + fun put_h2_alias/3, + maps:get(h2_settings, Opts, #{}), + maps:with([h2_initial_window_size, h2_max_frame_size], Opts) + ), + Settings#{enable_connect_protocol => true}. + -spec init_protocol(#state{}) -> {ok, #state{}} | {error, nhttp_sock:socket_error()}. init_protocol(#state{family = http2, socket = Socket, peer = Peer, opts = Opts} = State) -> - UserH2Settings = maps:get(h2_settings, Opts, #{}), - H2Settings = UserH2Settings#{enable_connect_protocol => true}, - H2Conn0 = nhttp_h2:new(server, H2Settings), + H2Conn0 = nhttp_h2:new(server, h2_settings_from_opts(Opts)), H2Conn = nhttp_h2:set_peer(H2Conn0, Peer), Preface = nhttp_h2:preface(H2Conn), maybe ok ?= nhttp_sock:send(Socket, Preface), ok ?= nhttp_sock:setopts(Socket, [{active, once}]), - {ok, State#state{protocol_state = #h2_state{h2_conn = H2Conn}}} + {ok, State#state{ + protocol_state = #h2_state{ + h2_conn = H2Conn, + response_delay = maps:get(h2_response_delay, Opts, 0) + } + }} end; init_protocol(#state{family = http1, socket = Socket, opts = Opts} = State) -> case nhttp_sock:setopts(Socket, [{active, once}]) of @@ -439,6 +456,14 @@ init_protocol(#state{family = http1, socket = Socket, opts = Opts} = State) -> {error, _} = Error -> Error end. +-spec put_h2_alias( + h2_initial_window_size | h2_max_frame_size, pos_integer(), nhttp_h2:settings() +) -> nhttp_h2:settings(). +put_h2_alias(h2_initial_window_size, Value, Settings) -> + Settings#{initial_window_size => Value}; +put_h2_alias(h2_max_frame_size, Value, Settings) -> + Settings#{max_frame_size => Value}. + -spec version_to_family(nhttp_lib:version()) -> http1 | http2 | http3. version_to_family(http1_0) -> http1; version_to_family(http1_1) -> http1; diff --git a/src/nhttp_conn.hrl b/src/nhttp_conn.hrl index d1a1bec..11f92fe 100644 --- a/src/nhttp_conn.hrl +++ b/src/nhttp_conn.hrl @@ -44,7 +44,8 @@ req_span :: {nhttp_otel:span_ctx(), integer()} | undefined, bytes_sent = 0 :: non_neg_integer(), response_started = false :: boolean(), - request :: nhttp_lib:request() | undefined + request :: nhttp_lib:request() | undefined, + held_response :: {reference(), nhttp_lib:response()} | undefined }). -record(h1_push_ctx, { @@ -75,7 +76,8 @@ h2_conn :: nhttp_h2:conn(), h2_streams = #{} :: #{nhttp_lib:stream_id() => #h2_stream{}}, h2_workers = #{} :: #{pid() => nhttp_lib:stream_id()}, - drain_deadline :: integer() | undefined + drain_deadline :: integer() | undefined, + response_delay = 0 :: nhttp:h2_response_delay() }). -record(state, { diff --git a/src/nhttp_conn_h2.erl b/src/nhttp_conn_h2.erl index 337c0f4..d756bf2 100644 --- a/src/nhttp_conn_h2.erl +++ b/src/nhttp_conn_h2.erl @@ -43,6 +43,7 @@ %% LOCAL MACROS %%%----------------------------------------------------------------------------- -define(DRAIN_IDLE_WAKE_MS, 100). +-define(RESPONSE_DELAY_TAG, nhttp_h2_response_delay). %%%----------------------------------------------------------------------------- %% API @@ -167,6 +168,9 @@ h2_receive(Parent, Debug, State, Timeout) -> end; {'EXIT', Parent, Reason} -> nhttp_conn:stop_parent(Reason, State); + {timeout, TimerRef, {?RESPONSE_DELAY_TAG, StreamId}} -> + NewState = handle_h2_response_delay(State, StreamId, TimerRef), + h2_loop(Parent, Debug, NewState); Info -> h2_loop(Parent, Debug, nhttp_conn_ws_h2:handle_info(State, Info)) after Timeout -> @@ -247,26 +251,20 @@ apply_h2_request_result(State, StreamId, Result) -> -spec apply_request_result(#state{}, nhttp_lib:stream_id(), term()) -> #state{}. apply_request_result( - #state{protocol_state = #h2_state{} = H2} = State, + #state{protocol_state = #h2_state{response_delay = Delay}} = State, StreamId, - {reply, #{status := Status} = Response0, NewHState} + {reply, #{status := _} = Response0, NewHState} ) -> Stream = stream(State, StreamId), Request = stream_request(Stream), Response = nhttp_conn_compress:maybe_compress( Response0, Request, nhttp_conn:compress_config(State) ), - {NewH2Conn, NewStreams} = send_h2_response(State, StreamId, Response), - State1 = State#state{ - protocol_state = H2#h2_state{h2_conn = NewH2Conn, h2_streams = NewStreams} - }, - State2 = maybe_rst_stream_unread_body(State1, StreamId, Stream), - State3 = release_request_worker(State2, StreamId, Stream), - nhttp_conn:emit_request_stop(State3, Status, Stream#h2_stream.req_span), - State3#state{ - handler_state = NewHState, - requests_count = State3#state.requests_count + 1 - }; + State1 = State#state{handler_state = NewHState}, + case Delay of + 0 -> complete_h2_reply(State1, StreamId, Stream, Response); + _ -> hold_h2_reply(State1, StreamId, Stream, Response, Delay) + end; apply_request_result(State, StreamId, {stream, {producer, Status, Headers, _Producer}, NewHState}) -> start_h2_stream_push_response(State, StreamId, Status, Headers, NewHState); apply_request_result(State, StreamId, {accept_body, _BodyState, NewHState}) -> @@ -304,6 +302,35 @@ apply_request_result(State, StreamId, {nhttp_handler_exception, Class, Reason}) nhttp_conn:emit_request_stop(State1, ?HTTP_INTERNAL_SERVER_ERROR, Stream#h2_stream.req_span), State1#state{requests_count = State1#state.requests_count + 1}. +-spec cancel_held_response(#h2_stream{}) -> ok. +cancel_held_response(#h2_stream{held_response = undefined}) -> + ok; +cancel_held_response(#h2_stream{held_response = {TimerRef, _Response}}) -> + ok = erlang:cancel_timer(TimerRef, [{async, true}, {info, false}]), + ok. + +-doc """ +Send a `{reply, _, _}` response and close the request bookkeeping: the +unread-body RST_STREAM, the worker release and the request span. With a +response delay this runs when the hold timer fires. +""". +-spec complete_h2_reply(#state{}, nhttp_lib:stream_id(), #h2_stream{}, nhttp_lib:response()) -> + #state{}. +complete_h2_reply( + #state{protocol_state = #h2_state{} = H2} = State, + StreamId, + Stream, + #{status := Status} = Response +) -> + {NewH2Conn, NewStreams} = send_h2_response(State, StreamId, Response), + State1 = State#state{ + protocol_state = H2#h2_state{h2_conn = NewH2Conn, h2_streams = NewStreams} + }, + State2 = maybe_rst_stream_unread_body(State1, StreamId, Stream), + State3 = release_request_worker(State2, StreamId, Stream), + nhttp_conn:emit_request_stop(State3, Status, Stream#h2_stream.req_span), + State3#state{requests_count = State3#state.requests_count + 1}. + -spec dispatch_h2_request(#state{}, nhttp_lib:stream_id(), nhttp_lib:request()) -> #state{}. dispatch_h2_request(#state{limits = Limits} = State, StreamId, Request) -> case nhttp_limits:validate_request(Request, Limits) of @@ -458,6 +485,12 @@ apply_stream_push_validation( } end. +-spec draw_response_delay(nhttp:h2_response_delay()) -> non_neg_integer(). +draw_response_delay(Ms) when is_integer(Ms) -> + Ms; +draw_response_delay({uniform, MinMs, MaxMs}) -> + MinMs + rand:uniform(MaxMs - MinMs + 1) - 1. + -doc """ Bridge accept_body into the streaming body recv loop. When the worker returns `accept_body` from `handle_request/2`, flush any @@ -652,6 +685,22 @@ handle_h2_request_trailers( State#state{protocol_state = H2#h2_state{h2_streams = Streams#{StreamId => Stream1}}} end. +-doc """ +Hold timer fired for a delayed reply. A stream that was reset or closed +during the hold, or a timer reference that no longer matches, is ignored. +""". +-spec handle_h2_response_delay(#state{}, nhttp_lib:stream_id(), reference()) -> #state{}. +handle_h2_response_delay( + #state{protocol_state = #h2_state{h2_streams = Streams}} = State, StreamId, TimerRef +) -> + case maps:get(StreamId, Streams, undefined) of + #h2_stream{type = request, held_response = {TimerRef, Response}} = Stream -> + Released = Stream#h2_stream{held_response = undefined}, + complete_h2_reply(State, StreamId, Released, Response); + _ -> + State + end. + -spec handle_h2_streaming_body_too_large(#state{}, nhttp_lib:stream_id(), #h2_stream{}) -> #state{}. handle_h2_streaming_body_too_large(State, StreamId, Stream) -> @@ -666,6 +715,41 @@ handle_h2_streaming_body_too_large(State, StreamId, Stream) -> end, handle_h2_limit_error(State, StreamId, body_too_large). +-doc """ +Park a compressed `{reply, _, _}` response on its stream until the +response-delay timer fires. The worker exits right after it posts the +result, so it is released here and its `DOWN` is flushed. The stream +keeps `type = request` and drops any DATA that arrives during the hold. +""". +-spec hold_h2_reply( + #state{}, nhttp_lib:stream_id(), #h2_stream{}, nhttp_lib:response(), nhttp:h2_response_delay() +) -> #state{}. +hold_h2_reply( + #state{protocol_state = #h2_state{h2_streams = Streams, h2_workers = Workers} = H2} = State, + StreamId, + #h2_stream{worker = WPid} = Stream, + Response, + Delay +) -> + TimerRef = erlang:start_timer( + draw_response_delay(Delay), self(), {?RESPONSE_DELAY_TAG, StreamId} + ), + ok = demonitor_worker(Stream), + Held = Stream#h2_stream{ + worker = undefined, + worker_ref = undefined, + worker_mref = undefined, + pending_ack = undefined, + streaming_body = false, + held_response = {TimerRef, Response} + }, + State#state{ + protocol_state = H2#h2_state{ + h2_streams = Streams#{StreamId => Held}, + h2_workers = maps:remove(WPid, Workers) + } + }. + -doc """ After a terminal handler result (`reply` / `stream`) on a request whose body was still in flight, send RST_STREAM(NO_ERROR) to ask the peer to @@ -710,7 +794,8 @@ release_request_worker( worker_ref = undefined, worker_mref = undefined, pending_ack = undefined, - request = undefined + request = undefined, + held_response = undefined }, Streams#{StreamId => Cleared} end, @@ -779,6 +864,16 @@ stream(#state{protocol_state = #h2_state{h2_streams = Streams}}, StreamId) -> stream_request(#h2_stream{request = R}) when is_map(R) -> R. +-spec track_held_stream_end(#state{}, nhttp_lib:stream_id(), #h2_stream{}, nhttp_h2:fin()) -> + #state{}. +track_held_stream_end(State, _StreamId, _Stream, nofin) -> + State; +track_held_stream_end( + #state{protocol_state = #h2_state{h2_streams = Streams} = H2} = State, StreamId, Stream, fin +) -> + Ended = Stream#h2_stream{end_stream = true}, + State#state{protocol_state = H2#h2_state{h2_streams = Streams#{StreamId => Ended}}}. + %%%----------------------------------------------------------------------------- %% INTERNAL FUNCTIONS - WORKER MESSAGE HANDLERS %%%----------------------------------------------------------------------------- @@ -921,6 +1016,8 @@ handle_h2_event( State; #h2_stream{type = websocket} -> nhttp_conn_ws_h2:handle_data(State, StreamId, Data, Fin); + #h2_stream{type = request, held_response = {_, _}} = Stream -> + track_held_stream_end(State, StreamId, Stream, Fin); #h2_stream{type = request} = Stream -> handle_h2_request_data(State, StreamId, Stream, Data, Fin); _ -> @@ -1124,6 +1221,7 @@ handle_h2_worker_stream_reset( pending_ack = PendingRef } = Stream, emit_h2_stream_complete(State, StreamId, Stream, peer_reset), + ok = cancel_held_response(Stream), case PendingRef of undefined when WPid =/= undefined -> WPid ! {chunk_ack, Ref, {error, closed}}, diff --git a/src/nhttp_listener.erl b/src/nhttp_listener.erl index 9e3b86d..efe665f 100644 --- a/src/nhttp_listener.erl +++ b/src/nhttp_listener.erl @@ -467,6 +467,50 @@ validate_alt_svc(Map) when is_map(Map) -> validate_alt_svc(_Other) -> invalid_alt_svc_error(not_false_or_map). +-spec validate_h2_alias( + h2_initial_window_size | h2_max_frame_size, + initial_window_size | max_frame_size, + nhttp:opts(), + nhttp_h2:settings() +) -> ok | {error, term()}. +validate_h2_alias(AliasKey, Key, Opts, Settings) -> + case {maps:get(AliasKey, Opts, undefined), maps:get(Key, Settings, undefined)} of + {undefined, _} -> ok; + {V, undefined} -> validate_h2_setting(Key, V); + {V, V} -> validate_h2_setting(Key, V); + {V, _Other} -> {error, {invalid_h2_setting, Key, V, "must equal the h2_settings value"}} + end. + +-spec validate_h2_aliases(nhttp:opts()) -> ok | {error, term()}. +validate_h2_aliases(Opts) -> + Settings = maps:get(h2_settings, Opts, #{}), + maybe + ok ?= validate_h2_alias(h2_initial_window_size, initial_window_size, Opts, Settings), + ok ?= validate_h2_alias(h2_max_frame_size, max_frame_size, Opts, Settings), + ok + end. + +-spec validate_h2_response_delay(nhttp:h2_response_delay() | undefined) -> ok | {error, term()}. +validate_h2_response_delay(undefined) -> + ok; +validate_h2_response_delay(Ms) when is_integer(Ms), Ms >= 0 -> + ok; +validate_h2_response_delay({uniform, Min, Max}) when + is_integer(Min), is_integer(Max), Min >= 0, Min =< Max +-> + ok; +validate_h2_response_delay(V) -> + {error, + {invalid_h2_response_delay, V, + "must be non_neg_integer() or {uniform, MinMs, MaxMs} with MinMs =< MaxMs"}}. + +-spec validate_h2_setting(initial_window_size | max_frame_size, pos_integer()) -> + ok | {error, term()}. +validate_h2_setting(initial_window_size, V) -> + validate_initial_window_size(V); +validate_h2_setting(max_frame_size, V) -> + validate_max_frame_size(V). + -spec validate_h2_settings(nhttp_h2:settings()) -> ok | {error, term()}. validate_h2_settings(Settings) -> maybe @@ -515,6 +559,8 @@ validate_opts(Opts) -> ok ?= validate_required_opts(Opts), ok ?= validate_versions(Opts), ok ?= validate_h2_settings(maps:get(h2_settings, Opts, #{})), + ok ?= validate_h2_aliases(Opts), + ok ?= validate_h2_response_delay(maps:get(h2_response_delay, Opts, undefined)), ok ?= validate_proxy_protocol(maps:get(proxy_protocol, Opts, false)), ok ?= validate_acceptor_count(maps:get(acceptor_count, Opts, undefined)), ok ?= validate_alt_svc(maps:get(alt_svc, Opts, #{})), diff --git a/test/nhttp_h2_credit_policy_SUITE.erl b/test/nhttp_h2_credit_policy_SUITE.erl new file mode 100644 index 0000000..32dc39b --- /dev/null +++ b/test/nhttp_h2_credit_policy_SUITE.erl @@ -0,0 +1,317 @@ +%%%----------------------------------------------------------------------------- +%%% HTTP/2 credit-policy affordances: the listener options that make nhttp +%%% behave like a slow or stingy peer, observed on the wire with the raw +%%% HTTP/2 client in `nhttp_test_helpers'. +%%%----------------------------------------------------------------------------- +-module(nhttp_h2_credit_policy_SUITE). + +-include_lib("common_test/include/ct.hrl"). +-include_lib("stdlib/include/assert.hrl"). + +-export([ + all/0, + init_per_suite/1, + end_per_suite/1, + init_per_testcase/2, + end_per_testcase/2 +]). + +-export([ + initial_window_size_alias_reaches_codec/1, + max_frame_size_alias_reaches_codec/1, + response_delay_holds_concurrent_streams/1, + response_delay_holds_reply/1, + response_delay_reply_only/1, + response_delay_reset_drops_held_reply/1, + response_delay_uniform_draws_per_response/1 +]). + +-behaviour(nhttp_handler). +-export([init/1, handle_request/2, handle_request_body/3]). + +-define(CANCEL, 8). +-define(DELAY_MS, 300). +-define(FRAME_SIZE_ERROR, 6). +-define(MEASURE_SLACK_MS, 50). +-define(RECV_TIMEOUT, 3000). +-define(RESET_AFTER_MS, 100). +-define(SETTINGS_INITIAL_WINDOW_SIZE, 4). +-define(UNIFORM_DRAWS, 20). +-define(UNIFORM_MAX_MS, 200). +-define(UNIFORM_MIN_MS, 100). +-define(WINDOW, 1000). + +%%%----------------------------------------------------------------------------- +%%% TEST HANDLER +%%%----------------------------------------------------------------------------- + +init(_Args) -> + {ok, #{}}. + +handle_request(#{path := <<"/echo">>}, State) -> + {accept_body, [], State}; +handle_request(#{path := <<"/stream">>}, State) -> + {stream, nhttp_stream:producer(200, [], fun send_one_chunk/1), State}; +handle_request(_Req, State) -> + {reply, nhttp_resp:ok(<<"hello">>), State}. + +handle_request_body({data, Chunk}, Acc, State) -> + {accept_body, [Chunk | Acc], State}; +handle_request_body({fin, _Trailers}, Acc, State) -> + {reply, nhttp_resp:ok(iolist_to_binary(lists:reverse(Acc))), State}; +handle_request_body({abort, Reason}, _Acc, State) -> + {abort, Reason, State}. + +send_one_chunk(SendChunk) -> + SendChunk(<<"chunk">>). + +%%%----------------------------------------------------------------------------- +%%% SUITE +%%%----------------------------------------------------------------------------- + +all() -> + [ + initial_window_size_alias_reaches_codec, + max_frame_size_alias_reaches_codec, + response_delay_holds_concurrent_streams, + response_delay_holds_reply, + response_delay_reply_only, + response_delay_reset_drops_held_reply, + response_delay_uniform_draws_per_response + ]. + +init_per_suite(Config) -> + _ = application:ensure_all_started(ssl), + {CertFile, _KeyFile} = nhttp_test_helpers:certs(), + case filelib:is_regular(CertFile) of + true -> Config; + false -> {skip, "SSL certificates not found"} + end. + +end_per_suite(_Config) -> + ok. + +init_per_testcase(_TC, Config) -> + process_flag(trap_exit, true), + Config. + +end_per_testcase(_TC, _Config) -> + ok. + +%%%----------------------------------------------------------------------------- +%%% TEST CASES +%%%----------------------------------------------------------------------------- + +initial_window_size_alias_reaches_codec(_Config) -> + {PidAlias, PortAlias} = nhttp_test_helpers:h2_start_server(?MODULE, #{ + h2_initial_window_size => ?WINDOW + }), + {PidSetting, PortSetting} = nhttp_test_helpers:h2_start_server(?MODULE, #{ + h2_settings => #{initial_window_size => ?WINDOW} + }), + try + Advertised = nhttp_test_helpers:h2_server_settings(PortAlias), + ?assertEqual([?WINDOW], [V || {?SETTINGS_INITIAL_WINDOW_SIZE, V} <- Advertised]), + ?assertEqual(nhttp_test_helpers:h2_server_settings(PortSetting), Advertised), + {ok, Sock} = nhttp_test_helpers:h2_connect(PortAlias), + Fits = binary:copy(<<$a>>, ?WINDOW), + ok = nhttp_test_helpers:h2_send_post(Sock, 1, <<"/echo">>, Fits), + Echoed = nhttp_test_helpers:h2_recv_stream(Sock, 1, ?RECV_TIMEOUT), + ?assertEqual(<<"200">>, nhttp_test_helpers:h2_response_status(Echoed)), + ?assertEqual(Fits, nhttp_test_helpers:h2_response_body(Echoed, 1)), + ssl:close(Sock) + after + nhttp:stop(PidAlias), + nhttp:stop(PidSetting) + end. + +max_frame_size_alias_reaches_codec(_Config) -> + {PidMin, PortMin} = nhttp_test_helpers:h2_start_server(?MODULE, #{ + h2_max_frame_size => 16384 + }), + try + {ok, SockMin} = nhttp_test_helpers:h2_connect(PortMin), + TooBig = binary:copy(<<$a>>, 16385), + ok = nhttp_test_helpers:h2_send_post(SockMin, 1, <<"/echo">>, TooBig), + Refused = nhttp_test_helpers:h2_recv_stream(SockMin, 1, ?RECV_TIMEOUT), + ?assertEqual(connection_error, error_scope(Refused, 1, ?FRAME_SIZE_ERROR)), + ssl:close(SockMin) + after + nhttp:stop(PidMin) + end, + {PidBig, PortBig} = nhttp_test_helpers:h2_start_server(?MODULE, #{ + h2_max_frame_size => 32768 + }), + try + {ok, SockBig} = nhttp_test_helpers:h2_connect(PortBig), + Body = binary:copy(<<$b>>, 20000), + ok = nhttp_test_helpers:h2_send_post(SockBig, 1, <<"/echo">>, Body), + Echoed = nhttp_test_helpers:h2_recv_stream(SockBig, 1, ?RECV_TIMEOUT), + ?assertEqual(<<"200">>, nhttp_test_helpers:h2_response_status(Echoed)), + ?assertEqual(Body, nhttp_test_helpers:h2_response_body(Echoed, 1)), + ssl:close(SockBig) + after + nhttp:stop(PidBig) + end. + +response_delay_holds_concurrent_streams(_Config) -> + {Pid, Port} = nhttp_test_helpers:h2_start_server(?MODULE, #{h2_response_delay => ?DELAY_MS}), + try + {ok, Sock} = nhttp_test_helpers:h2_connect(Port), + T1 = now_ms(), + ok = nhttp_test_helpers:h2_send_post(Sock, 1, <<"/echo">>, <<"one">>), + T3 = now_ms(), + ok = nhttp_test_helpers:h2_send_post(Sock, 3, <<"/echo">>, <<"three">>), + {Frames, #{1 := At1, 3 := At3}, _Rest} = await_streams(Sock, [1, 3], <<>>, ?RECV_TIMEOUT), + ?assertEqual([], rst_streams(Frames)), + ?assert(At1 - T1 >= ?DELAY_MS), + ?assert(At3 - T3 >= ?DELAY_MS), + ?assert(max(At1, At3) - T1 < 2 * ?DELAY_MS), + ?assertEqual(<<"one">>, nhttp_test_helpers:h2_response_body(Frames, 1)), + ?assertEqual(<<"three">>, nhttp_test_helpers:h2_response_body(Frames, 3)), + ssl:close(Sock) + after + nhttp:stop(Pid) + end. + +response_delay_holds_reply(_Config) -> + {Pid, Port} = nhttp_test_helpers:h2_start_server(?MODULE, #{h2_response_delay => ?DELAY_MS}), + try + {ok, Sock} = nhttp_test_helpers:h2_connect(Port), + Body = <<"held-until-the-delay-elapses">>, + ok = nhttp_test_helpers:h2_send_headers(Sock, 1, <<"/echo">>, byte_size(Body), false), + T0 = now_ms(), + ok = nhttp_test_helpers:h2_send_data(Sock, 1, Body, true), + {Frames, #{1 := At}, _Rest} = await_streams(Sock, [1], <<>>, ?RECV_TIMEOUT), + ?assertEqual([], rst_streams(frames_before_headers(Frames, 1))), + ?assert(At - T0 >= ?DELAY_MS), + ?assertEqual(<<"200">>, nhttp_test_helpers:h2_response_status(Frames)), + ?assertEqual(Body, nhttp_test_helpers:h2_response_body(Frames, 1)), + ssl:close(Sock) + after + nhttp:stop(Pid) + end. + +response_delay_reply_only(_Config) -> + {Pid, Port} = nhttp_test_helpers:h2_start_server(?MODULE, #{ + h2_response_delay => ?DELAY_MS, max_body_size => 8 + }), + try + {ok, Sock} = nhttp_test_helpers:h2_connect(Port), + TooLarge = binary:copy(<<$x>>, 32), + TError = now_ms(), + ok = nhttp_test_helpers:h2_send_post(Sock, 1, <<"/echo">>, TooLarge), + {Error, #{1 := AtError}, Rest1} = await_streams(Sock, [1], <<>>, ?RECV_TIMEOUT), + ?assertEqual(<<"413">>, nhttp_test_helpers:h2_response_status(Error)), + ?assert(AtError - TError < ?DELAY_MS), + TStream = now_ms(), + ok = nhttp_test_helpers:h2_send_request(Sock, 3, <<"/stream">>), + {Stream, #{3 := AtStream}, Rest3} = await_streams(Sock, [3], Rest1, ?RECV_TIMEOUT), + ?assertEqual(<<"chunk">>, nhttp_test_helpers:h2_response_body(Stream, 3)), + ?assert(AtStream - TStream < ?DELAY_MS), + TReply = now_ms(), + ok = nhttp_test_helpers:h2_send_request(Sock, 5, <<"/hello">>), + {Reply, #{5 := AtReply}, _Rest5} = await_streams(Sock, [5], Rest3, ?RECV_TIMEOUT), + ?assertEqual(<<"hello">>, nhttp_test_helpers:h2_response_body(Reply, 5)), + ?assert(AtReply - TReply >= ?DELAY_MS), + ssl:close(Sock) + after + nhttp:stop(Pid) + end. + +response_delay_reset_drops_held_reply(_Config) -> + {Pid, Port} = nhttp_test_helpers:h2_start_server(?MODULE, #{h2_response_delay => ?DELAY_MS}), + try + {ok, Sock} = nhttp_test_helpers:h2_connect(Port), + ok = nhttp_test_helpers:h2_send_post(Sock, 1, <<"/echo">>, <<"cancelled">>), + timer:sleep(?RESET_AFTER_MS), + ok = nhttp_test_helpers:h2_send_rst_stream(Sock, 1, ?CANCEL), + Frames = nhttp_test_helpers:h2_recv(Sock, 2 * ?DELAY_MS), + ?assertEqual([], [F || {headers, 1, _, _} = F <- Frames]), + ok = nhttp_test_helpers:h2_send_request(Sock, 3, <<"/hello">>), + Later = nhttp_test_helpers:h2_recv_stream(Sock, 3, ?RECV_TIMEOUT), + ?assertEqual(<<"200">>, nhttp_test_helpers:h2_response_status(Later)), + ?assertEqual([], [F || {headers, 1, _, _} = F <- Later]), + ssl:close(Sock) + after + nhttp:stop(Pid) + end. + +response_delay_uniform_draws_per_response(_Config) -> + {Pid, Port} = nhttp_test_helpers:h2_start_server(?MODULE, #{ + h2_response_delay => {uniform, ?UNIFORM_MIN_MS, ?UNIFORM_MAX_MS} + }), + try + {ok, Sock} = nhttp_test_helpers:h2_connect(Port), + StreamIds = [2 * N - 1 || N <- lists:seq(1, ?UNIFORM_DRAWS)], + {Sock, _Rest, Delays} = lists:foldl(fun measure_reply/2, {Sock, <<>>, []}, StreamIds), + ?assertEqual([], [D || D <- Delays, D < ?UNIFORM_MIN_MS]), + ?assertEqual([], [D || D <- Delays, D > ?UNIFORM_MAX_MS + ?MEASURE_SLACK_MS]), + ?assert(length(lists:usort(Delays)) >= 2), + ssl:close(Sock) + after + nhttp:stop(Pid) + end. + +%%%----------------------------------------------------------------------------- +%%% HELPERS +%%%----------------------------------------------------------------------------- + +%% Receive until every stream in StreamIds is done. Returns the frames, a +%% map from stream id to the time its first HEADERS frame was decoded, and +%% the undecoded tail of the socket buffer. +await_streams(Sock, StreamIds, Buf, Timeout) -> + await_streams(Sock, StreamIds, Buf, [], #{}, Timeout). + +await_streams(Sock, StreamIds, Buf, Acc, Seen, Timeout) -> + case lists:all(fun(Id) -> nhttp_test_helpers:h2_stream_done(Acc, Id) end, StreamIds) of + true -> + {Acc, Seen, Buf}; + false -> + T0 = now_ms(), + {ok, Data} = ssl:recv(Sock, 0, Timeout), + {Frames, Rest} = nhttp_test_helpers:decode_h2_frames(<>), + Seen1 = note_headers_seen(Frames, StreamIds, Seen, now_ms()), + await_streams(Sock, StreamIds, Rest, Acc ++ Frames, Seen1, Timeout - (now_ms() - T0)) + end. + +frames_before_headers(Frames, StreamId) -> + lists:takewhile(fun(F) -> not is_headers(F, StreamId) end, Frames). + +error_scope(Frames, StreamId, Code) -> + Goaway = [C || {goaway, 0, <<_Last:32, C:32, _/binary>>} <- Frames, C =:= Code], + Rst = [C || {rst_stream, SId, C} <- Frames, SId =:= StreamId, C =:= Code], + case {Goaway, Rst} of + {[_ | _], _} -> connection_error; + {[], [_ | _]} -> stream_error; + {[], []} -> {no_error_seen, Frames} + end. + +is_headers({headers, SId, _, _}, StreamId) -> SId =:= StreamId; +is_headers(_, _) -> false. + +measure_reply(StreamId, {Sock, Buf, Delays}) -> + T0 = now_ms(), + ok = nhttp_test_helpers:h2_send_request(Sock, StreamId, <<"/hello">>), + {_Frames, #{StreamId := At}, Rest} = await_streams(Sock, [StreamId], Buf, ?RECV_TIMEOUT), + {Sock, Rest, [At - T0 | Delays]}. + +note_headers_seen(Frames, StreamIds, Seen, At) -> + lists:foldl( + fun(Id, Acc) -> + case + maps:is_key(Id, Acc) orelse not lists:any(fun(F) -> is_headers(F, Id) end, Frames) + of + true -> Acc; + false -> Acc#{Id => At} + end + end, + Seen, + StreamIds + ). + +now_ms() -> + erlang:monotonic_time(millisecond). + +rst_streams(Frames) -> + [F || {rst_stream, _, _} = F <- Frames]. diff --git a/test/nhttp_h2_streaming_body_SUITE.erl b/test/nhttp_h2_streaming_body_SUITE.erl index b007a77..02586be 100644 --- a/test/nhttp_h2_streaming_body_SUITE.erl +++ b/test/nhttp_h2_streaming_body_SUITE.erl @@ -134,154 +134,160 @@ end_per_testcase(_TC, _Config) -> %%% TEST CASES %%%----------------------------------------------------------------------------- -echo_post(Config) -> - {Pid, Port} = start_server(Config, #{}), +echo_post(_Config) -> + {Pid, Port} = nhttp_test_helpers:h2_start_server(?MODULE, #{}), try Body = <<"hello-streamed-h2-body">>, - {ok, Sock} = h2_connect(Port), - ok = send_post(Sock, 1, <<"/echo">>, Body), - Frames = recv_stream(Sock, 1, 3000), - ?assertEqual(<<"200">>, response_status(Frames)), - ?assertEqual(Body, response_body(Frames, 1)), + {ok, Sock} = nhttp_test_helpers:h2_connect(Port), + ok = nhttp_test_helpers:h2_send_post(Sock, 1, <<"/echo">>, Body), + Frames = nhttp_test_helpers:h2_recv_stream(Sock, 1, 3000), + ?assertEqual(<<"200">>, nhttp_test_helpers:h2_response_status(Frames)), + ?assertEqual(Body, nhttp_test_helpers:h2_response_body(Frames, 1)), ssl:close(Sock) after nhttp:stop(Pid) end. -echo_post_split_data(Config) -> - {Pid, Port} = start_server(Config, #{}), +echo_post_split_data(_Config) -> + {Pid, Port} = nhttp_test_helpers:h2_start_server(?MODULE, #{}), try Part1 = <<"first-part-">>, Part2 = <<"second-part">>, Body = <>, - {ok, Sock} = h2_connect(Port), - ok = send_headers(Sock, 1, <<"/echo">>, byte_size(Body), false), - ok = send_data(Sock, 1, Part1, false), + {ok, Sock} = nhttp_test_helpers:h2_connect(Port), + ok = nhttp_test_helpers:h2_send_headers(Sock, 1, <<"/echo">>, byte_size(Body), false), + ok = nhttp_test_helpers:h2_send_data(Sock, 1, Part1, false), timer:sleep(50), - ok = send_data(Sock, 1, Part2, true), - Frames = recv_stream(Sock, 1, 3000), - ?assertEqual(<<"200">>, response_status(Frames)), - ?assertEqual(Body, response_body(Frames, 1)), + ok = nhttp_test_helpers:h2_send_data(Sock, 1, Part2, true), + Frames = nhttp_test_helpers:h2_recv_stream(Sock, 1, 3000), + ?assertEqual(<<"200">>, nhttp_test_helpers:h2_response_status(Frames)), + ?assertEqual(Body, nhttp_test_helpers:h2_response_body(Frames, 1)), ssl:close(Sock) after nhttp:stop(Pid) end. -echo_post_many_data_frames(Config) -> - {Pid, Port} = start_server(Config, #{}), +echo_post_many_data_frames(_Config) -> + {Pid, Port} = nhttp_test_helpers:h2_start_server(?MODULE, #{}), try Parts = [ <<"chunk-", (integer_to_binary(N))/binary, "|">> || N <- lists:seq(1, 128) ], Body = iolist_to_binary(Parts), - {ok, Sock} = h2_connect(Port), - ok = send_headers(Sock, 1, <<"/echo">>, byte_size(Body), false), + {ok, Sock} = nhttp_test_helpers:h2_connect(Port), + ok = nhttp_test_helpers:h2_send_headers(Sock, 1, <<"/echo">>, byte_size(Body), false), ok = send_data_frames(Sock, 1, Parts), - Frames = recv_stream(Sock, 1, 5000), - ?assertEqual(<<"200">>, response_status(Frames)), - ?assertEqual(Body, response_body(Frames, 1)), + Frames = nhttp_test_helpers:h2_recv_stream(Sock, 1, 5000), + ?assertEqual(<<"200">>, nhttp_test_helpers:h2_response_status(Frames)), + ?assertEqual(Body, nhttp_test_helpers:h2_response_body(Frames, 1)), ssl:close(Sock) after nhttp:stop(Pid) end. -reply_mid_body_rst_no_error(Config) -> - {Pid, Port} = start_server(Config, #{}), +reply_mid_body_rst_no_error(_Config) -> + {Pid, Port} = nhttp_test_helpers:h2_start_server(?MODULE, #{}), try - {ok, Sock} = h2_connect(Port), + {ok, Sock} = nhttp_test_helpers:h2_connect(Port), Body = <<"the-handler-replies-on-the-first-chunk">>, - ok = send_headers(Sock, 1, <<"/reply-mid-body">>, byte_size(Body), false), + ok = nhttp_test_helpers:h2_send_headers( + Sock, 1, <<"/reply-mid-body">>, byte_size(Body), false + ), Chunk1 = binary:part(Body, 0, 8), - ok = send_data(Sock, 1, Chunk1, false), - Frames = recv_stream(Sock, 1, 3000), - ?assertEqual(<<"200">>, response_status(Frames)), - ?assertEqual(<<"early">>, response_body(Frames, 1)), + ok = nhttp_test_helpers:h2_send_data(Sock, 1, Chunk1, false), + Frames = nhttp_test_helpers:h2_recv_stream(Sock, 1, 3000), + ?assertEqual(<<"200">>, nhttp_test_helpers:h2_response_status(Frames)), + ?assertEqual(<<"early">>, nhttp_test_helpers:h2_response_body(Frames, 1)), ?assert(has_rst_no_error(Frames, 1)), ssl:close(Sock) after nhttp:stop(Pid) end. -max_body_size_413(Config) -> - {Pid, Port} = start_server(Config, #{max_body_size => 8}), +max_body_size_413(_Config) -> + {Pid, Port} = nhttp_test_helpers:h2_start_server(?MODULE, #{max_body_size => 8}), try - {ok, Sock} = h2_connect(Port), + {ok, Sock} = nhttp_test_helpers:h2_connect(Port), Body = <<"way-too-many-bytes-for-the-cap">>, - ok = send_headers(Sock, 1, <<"/echo">>, byte_size(Body), false), - ok = send_data(Sock, 1, Body, true), - Frames = recv_stream(Sock, 1, 3000), - ?assertEqual(<<"413">>, response_status(Frames)), + ok = nhttp_test_helpers:h2_send_headers(Sock, 1, <<"/echo">>, byte_size(Body), false), + ok = nhttp_test_helpers:h2_send_data(Sock, 1, Body, true), + Frames = nhttp_test_helpers:h2_recv_stream(Sock, 1, 3000), + ?assertEqual(<<"413">>, nhttp_test_helpers:h2_response_status(Frames)), ssl:close(Sock) after nhttp:stop(Pid) end. -trailers_streaming(Config) -> - {Pid, Port} = start_server(Config, #{}), +trailers_streaming(_Config) -> + {Pid, Port} = nhttp_test_helpers:h2_start_server(?MODULE, #{}), try - {ok, Sock} = h2_connect(Port), + {ok, Sock} = nhttp_test_helpers:h2_connect(Port), Body = <<"hello">>, - ok = send_headers(Sock, 1, <<"/trailers-stream">>, byte_size(Body), false), + ok = nhttp_test_helpers:h2_send_headers( + Sock, 1, <<"/trailers-stream">>, byte_size(Body), false + ), timer:sleep(100), - ok = send_data(Sock, 1, Body, false), + ok = nhttp_test_helpers:h2_send_data(Sock, 1, Body, false), timer:sleep(50), ok = send_trailers(Sock, 1, [{<<"x-trailer">>, <<"abc">>}]), - Frames = recv_stream(Sock, 1, 5000), - ?assertEqual(<<"200">>, response_status(Frames)), - ?assertEqual(<<"hello|x-trailer=abc">>, response_body(Frames, 1)), + Frames = nhttp_test_helpers:h2_recv_stream(Sock, 1, 5000), + ?assertEqual(<<"200">>, nhttp_test_helpers:h2_response_status(Frames)), + ?assertEqual(<<"hello|x-trailer=abc">>, nhttp_test_helpers:h2_response_body(Frames, 1)), ssl:close(Sock) after nhttp:stop(Pid) end. -trailers_buffered(Config) -> - {Pid, Port} = start_server(Config, #{}), +trailers_buffered(_Config) -> + {Pid, Port} = nhttp_test_helpers:h2_start_server(?MODULE, #{}), try - {ok, Sock} = h2_connect(Port), + {ok, Sock} = nhttp_test_helpers:h2_connect(Port), Body = <<"hello">>, - ok = send_headers(Sock, 1, <<"/trailers-buffered">>, byte_size(Body), false), - ok = send_data(Sock, 1, Body, false), + ok = nhttp_test_helpers:h2_send_headers( + Sock, 1, <<"/trailers-buffered">>, byte_size(Body), false + ), + ok = nhttp_test_helpers:h2_send_data(Sock, 1, Body, false), ok = send_trailers(Sock, 1, [{<<"x-trailer">>, <<"abc">>}]), - Frames = recv_stream(Sock, 1, 5000), - ?assertEqual(<<"200">>, response_status(Frames)), - ?assertEqual(<<"hello|x-trailer=abc">>, response_body(Frames, 1)), + Frames = nhttp_test_helpers:h2_recv_stream(Sock, 1, 5000), + ?assertEqual(<<"200">>, nhttp_test_helpers:h2_response_status(Frames)), + ?assertEqual(<<"hello|x-trailer=abc">>, nhttp_test_helpers:h2_response_body(Frames, 1)), ssl:close(Sock) after nhttp:stop(Pid) end. -bad_handler_return(Config) -> - {Pid, Port} = start_server(Config, #{}), +bad_handler_return(_Config) -> + {Pid, Port} = nhttp_test_helpers:h2_start_server(?MODULE, #{}), try - {ok, Sock} = h2_connect(Port), - ok = send_headers(Sock, 1, <<"/bad-return">>, 0, true), - Frames = recv_stream(Sock, 1, 3000), + {ok, Sock} = nhttp_test_helpers:h2_connect(Port), + ok = nhttp_test_helpers:h2_send_headers(Sock, 1, <<"/bad-return">>, 0, true), + Frames = nhttp_test_helpers:h2_recv_stream(Sock, 1, 3000), ?assert(has_rst_stream(Frames, 1)), ssl:close(Sock) after nhttp:stop(Pid) end. -ws_upgrade_on_h2_rejected(Config) -> - {Pid, Port} = start_server(Config, #{}), +ws_upgrade_on_h2_rejected(_Config) -> + {Pid, Port} = nhttp_test_helpers:h2_start_server(?MODULE, #{}), try - {ok, Sock} = h2_connect(Port), - ok = send_headers(Sock, 1, <<"/ws-upgrade">>, 0, true), - Frames = recv_stream(Sock, 1, 3000), - ?assertEqual(<<"500">>, response_status(Frames)), + {ok, Sock} = nhttp_test_helpers:h2_connect(Port), + ok = nhttp_test_helpers:h2_send_headers(Sock, 1, <<"/ws-upgrade">>, 0, true), + Frames = nhttp_test_helpers:h2_recv_stream(Sock, 1, 3000), + ?assertEqual(<<"500">>, nhttp_test_helpers:h2_response_status(Frames)), ssl:close(Sock) after nhttp:stop(Pid) end. -ws_upgrade_on_h2_session_rejected(Config) -> - {Pid, Port} = start_server(Config, #{}), +ws_upgrade_on_h2_session_rejected(_Config) -> + {Pid, Port} = nhttp_test_helpers:h2_start_server(?MODULE, #{}), try - {ok, Sock} = h2_connect(Port), - ok = send_headers(Sock, 1, <<"/ws-upgrade-session">>, 0, true), - Frames = recv_stream(Sock, 1, 3000), - ?assertEqual(<<"500">>, response_status(Frames)), + {ok, Sock} = nhttp_test_helpers:h2_connect(Port), + ok = nhttp_test_helpers:h2_send_headers(Sock, 1, <<"/ws-upgrade-session">>, 0, true), + Frames = nhttp_test_helpers:h2_recv_stream(Sock, 1, 3000), + ?assertEqual(<<"500">>, nhttp_test_helpers:h2_response_status(Frames)), ssl:close(Sock) after nhttp:stop(Pid) @@ -300,23 +306,6 @@ has_rst_stream(Frames, StreamId) -> %%% SERVER HELPERS %%%----------------------------------------------------------------------------- -start_server(Config, Extra) -> - CertFile = ?config(certfile, Config), - KeyFile = ?config(keyfile, Config), - Opts = maps:merge( - #{ - port => 0, - handler => ?MODULE, - tls => #{certfile => CertFile, keyfile => KeyFile}, - versions => [http2], - timeouts => #{idle => 5000} - }, - Extra - ), - {ok, Pid} = nhttp:start_link(Opts), - {ok, Port} = nhttp:get_port(Pid), - {Pid, Port}. - find_test_conf_dir() -> ModPath = code:which(?MODULE), TestDir = filename:dirname(ModPath), @@ -326,60 +315,10 @@ find_test_conf_dir() -> %%% H2 CLIENT HELPERS %%%----------------------------------------------------------------------------- -h2_connect(Port) -> - {ok, Sock} = ssl:connect( - "127.0.0.1", - Port, - [ - binary, - {active, false}, - {verify, verify_none}, - {alpn_advertised_protocols, [<<"h2">>]} - ], - 5000 - ), - ok = ssl:send(Sock, <<"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n">>), - ok = ssl:send(Sock, <<0, 0, 0, 4, 0, 0, 0, 0, 0>>), - _ = ssl:recv(Sock, 0, 1000), - ok = ssl:send(Sock, <<0, 0, 0, 4, 1, 0, 0, 0, 0>>), - {ok, Sock}. - -send_post(Sock, StreamId, Path, Body) -> - ok = send_headers(Sock, StreamId, Path, byte_size(Body), false), - send_data(Sock, StreamId, Body, true). - -send_headers(Sock, StreamId, Path, Length, EndStream) -> - {ok, Enc} = nhttp_hpack:new(), - Headers = [ - {<<":method">>, <<"POST">>}, - {<<":scheme">>, <<"https">>}, - {<<":authority">>, <<"localhost">>}, - {<<":path">>, Path}, - {<<"content-length">>, integer_to_binary(Length)} - ], - {ok, IOList, _Enc1} = nhttp_hpack:encode(Headers, Enc), - Block = iolist_to_binary(IOList), - Flags = - case EndStream of - true -> 16#05; - false -> 16#04 - end, - Frame = <<(byte_size(Block)):24, 1, Flags, 0:1, StreamId:31, Block/binary>>, - ssl:send(Sock, Frame). - -send_data(Sock, StreamId, Data, EndStream) -> - Flags = - case EndStream of - true -> 16#01; - false -> 16#00 - end, - Frame = <<(byte_size(Data)):24, 0, Flags, 0:1, StreamId:31, Data/binary>>, - ssl:send(Sock, Frame). - send_data_frames(Sock, StreamId, [Last]) -> - send_data(Sock, StreamId, Last, true); + nhttp_test_helpers:h2_send_data(Sock, StreamId, Last, true); send_data_frames(Sock, StreamId, [Part | Rest]) -> - ok = send_data(Sock, StreamId, Part, false), + ok = nhttp_test_helpers:h2_send_data(Sock, StreamId, Part, false), send_data_frames(Sock, StreamId, Rest). send_trailers(Sock, StreamId, Trailers) -> @@ -390,108 +329,11 @@ send_trailers(Sock, StreamId, Trailers) -> Frame = <<(byte_size(Block)):24, 1, Flags, 0:1, StreamId:31, Block/binary>>, ssl:send(Sock, Frame). -recv_stream(Sock, StreamId, Timeout) -> - Frames = recv_stream_until_done(Sock, StreamId, Timeout, <<>>, []), - Tail = recv_stream_tail(Sock, 200, <<>>, []), - Frames ++ Tail. - -recv_stream_until_done(Sock, StreamId, Timeout, Buf, Acc) -> - case stream_done(Acc, StreamId) of - true -> - Acc; - false -> - case ssl:recv(Sock, 0, Timeout) of - {ok, Data} -> - NewBuf = <>, - {Frames, Rest} = decode_frames(NewBuf, []), - recv_stream_until_done(Sock, StreamId, Timeout, Rest, Acc ++ Frames); - {error, _} -> - Acc - end - end. - -recv_stream_tail(Sock, Timeout, Buf, Acc) -> - case ssl:recv(Sock, 0, Timeout) of - {ok, Data} -> - NewBuf = <>, - {Frames, Rest} = decode_frames(NewBuf, []), - recv_stream_tail(Sock, Timeout, Rest, Acc ++ Frames); - {error, _} -> - Acc - end. - -stream_done(Frames, StreamId) -> - lists:any( - fun - ({data, SId, _, fin}) when SId =:= StreamId -> true; - ({headers, SId, _, fin}) when SId =:= StreamId -> true; - ({rst_stream, SId, _}) when SId =:= StreamId -> true; - (_) -> false - end, - Frames - ). - -response_status(Frames) -> - case [P || {headers, _, P, _} <- Frames] of - [Block | _] -> - {ok, Dec} = nhttp_hpack:new(), - case nhttp_hpack:decode(Block, Dec) of - {ok, Headers, _} -> proplists:get_value(<<":status">>, Headers); - _ -> undefined - end; - [] -> - undefined - end. - -response_body(Frames, StreamId) -> - iolist_to_binary([P || {data, SId, P, _} <- Frames, SId =:= StreamId]). - has_rst_no_error(Frames, StreamId) -> lists:any( fun - ({rst_stream, SId, no_error}) when SId =:= StreamId -> true; + ({rst_stream, SId, 0}) when SId =:= StreamId -> true; (_) -> false end, Frames ). - -decode_frames(<>, Acc) -> - Frame = decode_frame(Type, Flags, StreamId, Payload), - decode_frames(Rest, [Frame | Acc]); -decode_frames(Other, Acc) -> - {lists:reverse(Acc), Other}. - -decode_frame(0, Flags, StreamId, Payload) -> - Fin = - case Flags band 1 of - 1 -> fin; - 0 -> nofin - end, - {data, StreamId, Payload, Fin}; -decode_frame(1, Flags, StreamId, Payload) -> - Fin = - case Flags band 1 of - 1 -> fin; - 0 -> nofin - end, - {headers, StreamId, Payload, Fin}; -decode_frame(3, _Flags, StreamId, <>) -> - {rst_stream, StreamId, error_code(Code)}; -decode_frame(Type, _Flags, StreamId, Payload) -> - {other, StreamId, Type, Payload}. - -error_code(0) -> no_error; -error_code(1) -> protocol_error; -error_code(2) -> internal_error; -error_code(3) -> flow_control_error; -error_code(4) -> settings_timeout; -error_code(5) -> stream_closed; -error_code(6) -> frame_size_error; -error_code(7) -> refused_stream; -error_code(8) -> cancel; -error_code(9) -> compression_error; -error_code(10) -> connect_error; -error_code(11) -> enhance_your_calm; -error_code(12) -> inadequate_security; -error_code(13) -> http_1_1_required; -error_code(N) -> N. diff --git a/test/nhttp_listener_validation_SUITE.erl b/test/nhttp_listener_validation_SUITE.erl index b4331fd..2364add 100644 --- a/test/nhttp_listener_validation_SUITE.erl +++ b/test/nhttp_listener_validation_SUITE.erl @@ -27,6 +27,9 @@ -export([ acceptor_sys_lifecycle/1, acceptor_sys_terminate/1, + listener_h2_alias_conflicts_with_settings/1, + listener_h2_alias_out_of_range/1, + listener_h2_response_delay_invalid/1, listener_invalid_tls/1, listener_invalid_versions/1, listener_listen_failed/1, @@ -41,6 +44,9 @@ all() -> [ acceptor_sys_lifecycle, acceptor_sys_terminate, + listener_h2_alias_conflicts_with_settings, + listener_h2_alias_out_of_range, + listener_h2_response_delay_invalid, listener_invalid_tls, listener_invalid_versions, listener_listen_failed, @@ -74,6 +80,74 @@ handle_request(_Request, State) -> %%% TEST CASES %%%----------------------------------------------------------------------------- +listener_h2_alias_conflicts_with_settings(_Config) -> + assert_error_contains( + "invalid_h2_setting", + nhttp:start_link(#{ + port => 0, + handler => ?MODULE, + versions => [http1_1], + h2_max_frame_size => 16384, + h2_settings => #{max_frame_size => 32768} + }) + ), + assert_error_contains( + "invalid_h2_setting", + nhttp:start_link(#{ + port => 0, + handler => ?MODULE, + versions => [http1_1], + h2_initial_window_size => 1000, + h2_settings => #{initial_window_size => 2000} + }) + ), + {ok, Pid} = nhttp:start_link(#{ + port => 0, + handler => ?MODULE, + versions => [http1_1], + h2_initial_window_size => 1000, + h2_max_frame_size => 32768, + h2_settings => #{initial_window_size => 1000, max_frame_size => 32768} + }), + nhttp:stop(Pid), + ok. + +listener_h2_alias_out_of_range(_Config) -> + Base = #{port => 0, handler => ?MODULE, versions => [http1_1]}, + ?assertEqual( + nhttp:start_link(Base#{h2_settings => #{max_frame_size => 16383}}), + nhttp:start_link(Base#{h2_max_frame_size => 16383}) + ), + assert_error_contains("max_frame_size", nhttp:start_link(Base#{h2_max_frame_size => 16383})), + ?assertEqual( + nhttp:start_link(Base#{h2_settings => #{initial_window_size => 0}}), + nhttp:start_link(Base#{h2_initial_window_size => 0}) + ), + assert_error_contains( + "initial_window_size", nhttp:start_link(Base#{h2_initial_window_size => 0}) + ), + ok. + +listener_h2_response_delay_invalid(_Config) -> + Base = #{port => 0, handler => ?MODULE, versions => [http1_1]}, + Invalid = [-1, 1.5, ms, {uniform, 200, 100}, {uniform, -1, 5}, {uniform, 1}], + lists:foreach( + fun(Delay) -> + assert_error_contains( + "invalid_h2_response_delay", nhttp:start_link(Base#{h2_response_delay => Delay}) + ) + end, + Invalid + ), + lists:foreach( + fun(Delay) -> + {ok, Pid} = nhttp:start_link(Base#{h2_response_delay => Delay}), + nhttp:stop(Pid) + end, + [0, 250, {uniform, 5, 5}, {uniform, 100, 200}] + ), + ok. + listener_invalid_versions(_Config) -> assert_error_contains( "invalid_versions", diff --git a/test/nhttp_test_helpers.erl b/test/nhttp_test_helpers.erl index f503abf..852951e 100644 --- a/test/nhttp_test_helpers.erl +++ b/test/nhttp_test_helpers.erl @@ -30,16 +30,28 @@ h2_connect/1, h2_open_stream/3, h2_recv/2, + h2_recv_stream/3, + h2_response_body/2, + h2_response_status/1, + h2_send_data/4, + h2_send_headers/5, + h2_send_post/4, h2_send_raw/2, h2_send_request/3, h2_send_rst_stream/3, h2_send_settings/2, h2_send_window_update/3, + h2_server_settings/1, + h2_start_server/2, + h2_stream_done/2, tcp_connect/1 ]). -define(DEFAULT_TIMEOUT, 5000). +-define(H2_IDLE_TIMEOUT, 5000). -define(POLL_INTERVAL, 25). +-define(SETTINGS_RECV_MS, 1000). +-define(STREAM_TAIL_MS, 200). -type h2_frame() :: {data, non_neg_integer(), binary(), fin | nofin} @@ -122,6 +134,23 @@ start(Opts) -> {ok, Port} = nhttp:get_port(Pid), {ok, Pid, Port}. +-doc "Start an HTTP/2-only TLS listener on a free port with the test certificates.". +-spec h2_start_server(module(), nhttp:opts()) -> {pid(), inet:port_number()}. +h2_start_server(Handler, Extra) -> + {CertFile, KeyFile} = certs(), + {ok, Pid, Port} = start( + maps:merge( + #{ + handler => Handler, + tls => #{certfile => CertFile, keyfile => KeyFile}, + versions => [http2], + timeouts => #{idle => ?H2_IDLE_TIMEOUT} + }, + Extra + ) + ), + {Pid, Port}. + -spec conn_pids(pid()) -> [pid()]. conn_pids(ListenerPid) -> lists:append([conns_in(TSup) || TSup <- transport_sups(ListenerPid)]). @@ -170,23 +199,22 @@ drain_tcp(Sock, Acc) -> -spec h2_connect(inet:port_number()) -> {ok, ssl:sslsocket()}. h2_connect(Port) -> - {ok, Sock} = ssl:connect( - "127.0.0.1", - Port, - [ - binary, - {active, false}, - {verify, verify_none}, - {alpn_advertised_protocols, [<<"h2">>]} - ], - 5000 - ), - ok = ssl:send(Sock, <<"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n">>), - ok = ssl:send(Sock, <<0, 0, 0, 4, 0, 0, 0, 0, 0>>), + {ok, Sock} = h2_connect_preface(Port), _ = ssl:recv(Sock, 0, 2000), ok = ssl:send(Sock, <<0, 0, 0, 4, 1, 0, 0, 0, 0>>), {ok, Sock}. +-doc """ +Open a connection and return the parameters of the SETTINGS frame the +server sends in its preface, as `{Identifier, Value}` pairs. +""". +-spec h2_server_settings(inet:port_number()) -> [{non_neg_integer(), non_neg_integer()}]. +h2_server_settings(Port) -> + {ok, Sock} = h2_connect_preface(Port), + Frames = h2_recv(Sock, ?SETTINGS_RECV_MS), + ok = ssl:close(Sock), + [{Id, Value} || {settings, 0, Payload} <- Frames, <> <= Payload]. + -spec h2_send_request(ssl:sslsocket(), non_neg_integer(), binary()) -> ok. h2_send_request(Sock, StreamId, Path) -> Header = h2_header_block(Path), @@ -220,6 +248,86 @@ h2_send_window_update(Sock, StreamId, Increment) -> h2_recv(Sock, Timeout) -> h2_recv(Sock, Timeout, <<>>, []). +-doc """ +Receive frames until the stream ends (END_STREAM or RST_STREAM), then +drain the socket for a short tail so frames sent right after are seen. +""". +-spec h2_recv_stream(ssl:sslsocket(), non_neg_integer(), timeout()) -> [h2_frame()]. +h2_recv_stream(Sock, StreamId, Timeout) -> + Frames = recv_stream_until_done(Sock, StreamId, Timeout, <<>>, []), + Frames ++ recv_stream_tail(Sock, <<>>, []). + +-spec h2_response_body([h2_frame()], non_neg_integer()) -> binary(). +h2_response_body(Frames, StreamId) -> + iolist_to_binary([P || {data, SId, P, _} <- Frames, SId =:= StreamId]). + +-doc "The `:status` of the first HEADERS frame, decoded with a fresh HPACK table.". +-spec h2_response_status([h2_frame()]) -> binary() | undefined. +h2_response_status(Frames) -> + case [P || {headers, _, P, _} <- Frames] of + [Block | _] -> + {ok, Dec} = nhttp_hpack:new(), + case nhttp_hpack:decode(Block, Dec) of + {ok, Headers, _} -> proplists:get_value(<<":status">>, Headers); + {error, _} -> undefined + end; + [] -> + undefined + end. + +-doc "True once `Frames` carries END_STREAM or RST_STREAM for `StreamId`.". +-spec h2_stream_done([h2_frame()], non_neg_integer()) -> boolean(). +h2_stream_done(Frames, StreamId) -> + lists:any( + fun + ({data, SId, _, fin}) when SId =:= StreamId -> true; + ({headers, SId, _, fin}) when SId =:= StreamId -> true; + ({rst_stream, SId, _}) when SId =:= StreamId -> true; + (_) -> false + end, + Frames + ). + +-spec h2_send_data(ssl:sslsocket(), non_neg_integer(), binary(), boolean()) -> + ok | {error, term()}. +h2_send_data(Sock, StreamId, Data, EndStream) -> + Flags = + case EndStream of + true -> 16#01; + false -> 16#00 + end, + Frame = <<(byte_size(Data)):24, 0, Flags, 0:1, StreamId:31, Data/binary>>, + ssl:send(Sock, Frame). + +-doc "Send a POST HEADERS frame with `content-length` set to `Length`.". +-spec h2_send_headers( + ssl:sslsocket(), non_neg_integer(), binary(), non_neg_integer(), boolean() +) -> ok | {error, term()}. +h2_send_headers(Sock, StreamId, Path, Length, EndStream) -> + {ok, Enc} = nhttp_hpack:new(), + Headers = [ + {<<":method">>, <<"POST">>}, + {<<":scheme">>, <<"https">>}, + {<<":authority">>, <<"localhost">>}, + {<<":path">>, Path}, + {<<"content-length">>, integer_to_binary(Length)} + ], + {ok, IOList, _Enc1} = nhttp_hpack:encode(Headers, Enc), + Block = iolist_to_binary(IOList), + Flags = + case EndStream of + true -> 16#05; + false -> 16#04 + end, + Frame = <<(byte_size(Block)):24, 1, Flags, 0:1, StreamId:31, Block/binary>>, + ssl:send(Sock, Frame). + +-spec h2_send_post(ssl:sslsocket(), non_neg_integer(), binary(), binary()) -> + ok | {error, term()}. +h2_send_post(Sock, StreamId, Path, Body) -> + ok = h2_send_headers(Sock, StreamId, Path, byte_size(Body), false), + h2_send_data(Sock, StreamId, Body, true). + -spec decode_h2_frames(binary()) -> {[h2_frame()], binary()}. decode_h2_frames(Bin) -> decode_h2_frames(Bin, []). @@ -274,6 +382,50 @@ decode_h2_frame(8, _Flags, StreamId, <>) -> decode_h2_frame(Type, _Flags, StreamId, Payload) -> {other, StreamId, Type, Payload}. +-spec h2_connect_preface(inet:port_number()) -> {ok, ssl:sslsocket()}. +h2_connect_preface(Port) -> + {ok, Sock} = ssl:connect( + "127.0.0.1", + Port, + [ + binary, + {active, false}, + {verify, verify_none}, + {alpn_advertised_protocols, [<<"h2">>]} + ], + 5000 + ), + ok = ssl:send(Sock, <<"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n">>), + ok = ssl:send(Sock, <<0, 0, 0, 4, 0, 0, 0, 0, 0>>), + {ok, Sock}. + -spec h2_fin(non_neg_integer()) -> fin | nofin. h2_fin(Flags) when Flags band 1 =:= 1 -> fin; h2_fin(_Flags) -> nofin. + +-spec recv_stream_until_done( + ssl:sslsocket(), non_neg_integer(), timeout(), binary(), [h2_frame()] +) -> [h2_frame()]. +recv_stream_until_done(Sock, StreamId, Timeout, Buf, Acc) -> + case h2_stream_done(Acc, StreamId) of + true -> + Acc; + false -> + case ssl:recv(Sock, 0, Timeout) of + {ok, Data} -> + {Frames, Rest} = decode_h2_frames(<>), + recv_stream_until_done(Sock, StreamId, Timeout, Rest, Acc ++ Frames); + {error, _} -> + Acc + end + end. + +-spec recv_stream_tail(ssl:sslsocket(), binary(), [h2_frame()]) -> [h2_frame()]. +recv_stream_tail(Sock, Buf, Acc) -> + case ssl:recv(Sock, 0, ?STREAM_TAIL_MS) of + {ok, Data} -> + {Frames, Rest} = decode_h2_frames(<>), + recv_stream_tail(Sock, Rest, Acc ++ Frames); + {error, _} -> + Acc + end. From 9564e4627372c0cab500e52e2dc357edcaeae0bd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miguel=20Rubinos=20Rodr=C3=ADguez?= Date: Tue, 22 Sep 2026 10:22:09 +0200 Subject: [PATCH 4/4] feat: h2 window credit policy (#15) --- CHANGELOG.md | 2 + README.md | 33 ++ src/nhttp.erl | 19 + src/nhttp_conn.erl | 21 +- src/nhttp_conn.hrl | 21 +- src/nhttp_conn_h2.erl | 473 ++++++++++++++++++++--- src/nhttp_listener.erl | 54 +++ test/nhttp_h2_credit_policy_SUITE.erl | 420 +++++++++++++++++++- test/nhttp_listener_validation_SUITE.erl | 82 ++++ 9 files changed, 1068 insertions(+), 57 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 43b0f86..9d047bc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Listener options `h2_initial_window_size` and `h2_max_frame_size`, aliases of the `h2_settings` keys - Listener option `h2_response_delay`, a fixed or uniform delay before the response headers of a `{reply, _, _}` result on HTTP/2 +- Listener options `h2_connection_window_policy` and `h2_stream_window_policy`, credit policies for the HTTP/2 receive windows with the shapes `eager`, `{threshold, N}`, `{delay, Ms}` and `never` +- The `on_response` shape of both credit policies, which sends the credit for a request ahead of the HEADERS of its `{reply, _, _}` response, and the listener option `h2_credit_batch` that batches the connection credit ### Changed diff --git a/README.md b/README.md index 1e6e032..8935e22 100644 --- a/README.md +++ b/README.md @@ -92,11 +92,44 @@ changes nothing on HTTP/1.1 or HTTP/3. | `h2_initial_window_size` | 65535 | Alias of `initial_window_size` in `h2_settings`: the receive window that each new stream grants the peer. | | `h2_max_frame_size` | 16384 | Alias of `max_frame_size` in `h2_settings`: the largest frame payload the server accepts. | | `h2_response_delay` | 0 | Milliseconds to hold the response headers of a `{reply, _, _}` result. `{uniform, MinMs, MaxMs}` draws a value per response. | +| `h2_connection_window_policy` | `eager` | Credit policy for the connection receive window. | +| `h2_stream_window_policy` | `eager` | Credit policy for each stream receive window. | +| `h2_credit_batch` | 0 | Octets per WINDOW_UPDATE under the `on_response` connection policy. 0 sends the whole accumulator with each response. Valid only when a policy is `on_response`. | An alias must equal the `h2_settings` key when both are present. The delay applies to `{reply, _, _}` results only. Error responses, producer streams and WebSocket upgrades go out at once. +A credit policy has five shapes. `eager` sends a WINDOW_UPDATE for each +body chunk as soon as the handler consumed it. `{threshold, N}` holds the +credit until the uncredited octets reach `N`, then sends the accumulated +total. `{delay, Ms}` sends the credit for each chunk `Ms` milliseconds +after the handler consumed it. `on_response` holds the credit of a request +until the HEADERS of its `{reply, _, _}` result go out. The credit goes in +the same socket write, ahead of the HEADERS. `never` sends no credit. + +The two policies are independent. No policy credits more than the handler +consumed. Error responses, `{abort, _, _}` results and stream resets +release no `on_response` credit. A stream that closes before its delayed +credit is due gets no stream WINDOW_UPDATE. The connection credit for +those octets still goes out. + +Under the `on_response` connection policy the responded octets move into +one accumulator. Each time the accumulator reaches `h2_credit_batch`, one +WINDOW_UPDATE of exactly that size goes out with the response, and the +remainder carries. This example imitates APNs, which credits half of its +window at a time, sends no stream credit, and answers 129 to 388 ms after +the request: + +```erlang +#{ + h2_connection_window_policy => on_response, + h2_credit_batch => 32830, + h2_stream_window_policy => never, + h2_response_delay => {uniform, 129, 388} +} +``` + ## Documentation [nhttp on HexDocs](https://hexdocs.pm/nhttp) diff --git a/src/nhttp.erl b/src/nhttp.erl index 0b7cb65..2fd891d 100644 --- a/src/nhttp.erl +++ b/src/nhttp.erl @@ -168,6 +168,7 @@ RFC 9001 §9.2). header_value/0, headers/0, h2_response_delay/0, + h2_window_policy/0, method/0, name/0, @@ -198,10 +199,13 @@ RFC 9001 §9.2). alt_svc => #{ma => non_neg_integer()} | false, backlog => pos_integer(), buffer => pos_integer(), + h2_connection_window_policy => h2_window_policy(), + h2_credit_batch => non_neg_integer(), h2_initial_window_size => 1..16#7fffffff, h2_max_frame_size => 16#4000..16#ffffff, h2_response_delay => h2_response_delay(), h2_settings => nhttp_h2:settings(), + h2_stream_window_policy => h2_window_policy(), handler := module(), handler_args => term(), compression => boolean(), @@ -225,6 +229,21 @@ upgrades are not delayed. Default 0. -type h2_response_delay() :: non_neg_integer() | {uniform, non_neg_integer(), non_neg_integer()}. +-doc """ +Credit policy for an HTTP/2 receive window. `eager` sends a WINDOW_UPDATE +for each body chunk as soon as the handler consumed it. `{threshold, N}` +holds the credit until the uncredited octets reach `N`, then sends the +accumulated total. `{delay, Ms}` sends the credit for each chunk `Ms` +milliseconds after the handler consumed it. `on_response` sends the +credit for the body of a request ahead of the response headers of its +`{reply, _, _}` result. On the connection window `h2_credit_batch` +holds that credit until the responded octets reach the batch, then +sends one WINDOW_UPDATE of exactly the batch. `never` sends no credit. +Default `eager`. +""". +-type h2_window_policy() :: + eager | {threshold, pos_integer()} | {delay, non_neg_integer()} | on_response | never. + -doc """ Connection timeouts (milliseconds, or `infinity`). diff --git a/src/nhttp_conn.erl b/src/nhttp_conn.erl index b4d3ea3..f4e6575 100644 --- a/src/nhttp_conn.erl +++ b/src/nhttp_conn.erl @@ -421,6 +421,13 @@ h1_state_from_opts(Opts) -> body_deadline = maps:get(body_deadline, Timeouts, infinity) }. +-spec h2_conn_credit(nhttp:h2_window_policy(), non_neg_integer()) -> h2_conn_credit(). +h2_conn_credit(eager, _Batch) -> eager; +h2_conn_credit(never, _Batch) -> never; +h2_conn_credit({threshold, N}, _Batch) -> {threshold, N, 0}; +h2_conn_credit({delay, Ms}, _Batch) -> {delay, Ms, queue:new()}; +h2_conn_credit(on_response, Batch) -> {on_response, Batch, 0}. + -doc """ Build the HTTP/2 settings the codec advertises. The first-class options `h2_initial_window_size` and `h2_max_frame_size` are aliases of the @@ -435,6 +442,13 @@ h2_settings_from_opts(Opts) -> ), Settings#{enable_connect_protocol => true}. +-spec h2_stream_credit(nhttp:h2_window_policy()) -> h2_stream_credit(). +h2_stream_credit(eager) -> eager; +h2_stream_credit(never) -> never; +h2_stream_credit({threshold, N}) -> {threshold, N}; +h2_stream_credit({delay, Ms}) -> {delay, Ms, queue:new()}; +h2_stream_credit(on_response) -> on_response. + -spec init_protocol(#state{}) -> {ok, #state{}} | {error, nhttp_sock:socket_error()}. init_protocol(#state{family = http2, socket = Socket, peer = Peer, opts = Opts} = State) -> H2Conn0 = nhttp_h2:new(server, h2_settings_from_opts(Opts)), @@ -446,7 +460,12 @@ init_protocol(#state{family = http2, socket = Socket, peer = Peer, opts = Opts} {ok, State#state{ protocol_state = #h2_state{ h2_conn = H2Conn, - response_delay = maps:get(h2_response_delay, Opts, 0) + conn_credit = h2_conn_credit( + maps:get(h2_connection_window_policy, Opts, eager), + maps:get(h2_credit_batch, Opts, 0) + ), + response_delay = maps:get(h2_response_delay, Opts, 0), + stream_credit = h2_stream_credit(maps:get(h2_stream_window_policy, Opts, eager)) } }} end; diff --git a/src/nhttp_conn.hrl b/src/nhttp_conn.hrl index 11f92fe..038e049 100644 --- a/src/nhttp_conn.hrl +++ b/src/nhttp_conn.hrl @@ -40,6 +40,8 @@ streaming_body = false :: boolean(), pending_trailers :: nhttp_lib:headers() | undefined, body_window_pending = queue:new() :: queue:queue(non_neg_integer()), + uncredited = 0 :: non_neg_integer(), + conn_uncredited = 0 :: non_neg_integer(), status :: nhttp_lib:status() | undefined, req_span :: {nhttp_otel:span_ctx(), integer()} | undefined, bytes_sent = 0 :: non_neg_integer(), @@ -72,12 +74,29 @@ body_deadline_at :: integer() | undefined }). +-type h2_credit_due() :: queue:queue({integer(), nhttp_lib:stream_id(), non_neg_integer()}). +-type h2_conn_credit() :: + eager + | never + | {threshold, pos_integer(), non_neg_integer()} + | {delay, non_neg_integer(), h2_credit_due()} + | {on_response, non_neg_integer(), non_neg_integer()}. +-type h2_stream_credit() :: + eager + | never + | {threshold, pos_integer()} + | {delay, non_neg_integer(), h2_credit_due()} + | on_response. + -record(h2_state, { h2_conn :: nhttp_h2:conn(), h2_streams = #{} :: #{nhttp_lib:stream_id() => #h2_stream{}}, h2_workers = #{} :: #{pid() => nhttp_lib:stream_id()}, drain_deadline :: integer() | undefined, - response_delay = 0 :: nhttp:h2_response_delay() + conn_credit = eager :: h2_conn_credit(), + credit_timer :: {reference(), integer()} | undefined, + response_delay = 0 :: nhttp:h2_response_delay(), + stream_credit = eager :: h2_stream_credit() }). -record(state, { diff --git a/src/nhttp_conn_h2.erl b/src/nhttp_conn_h2.erl index d756bf2..0c7c618 100644 --- a/src/nhttp_conn_h2.erl +++ b/src/nhttp_conn_h2.erl @@ -42,6 +42,7 @@ %%%----------------------------------------------------------------------------- %% LOCAL MACROS %%%----------------------------------------------------------------------------- +-define(CREDIT_DELAY_TAG, nhttp_h2_credit_delay). -define(DRAIN_IDLE_WAKE_MS, 100). -define(RESPONSE_DELAY_TAG, nhttp_h2_response_delay). @@ -171,6 +172,9 @@ h2_receive(Parent, Debug, State, Timeout) -> {timeout, TimerRef, {?RESPONSE_DELAY_TAG, StreamId}} -> NewState = handle_h2_response_delay(State, StreamId, TimerRef), h2_loop(Parent, Debug, NewState); + {timeout, TimerRef, ?CREDIT_DELAY_TAG} -> + NewState = handle_h2_credit_delay(State, TimerRef), + h2_loop(Parent, Debug, NewState); Info -> h2_loop(Parent, Debug, nhttp_conn_ws_h2:handle_info(State, Info)) after Timeout -> @@ -312,24 +316,30 @@ cancel_held_response(#h2_stream{held_response = {TimerRef, _Response}}) -> -doc """ Send a `{reply, _, _}` response and close the request bookkeeping: the unread-body RST_STREAM, the worker release and the request span. With a -response delay this runs when the hold timer fires. +response delay this runs when the hold timer fires. An `on_response` +policy releases the credit of the request ahead of the HEADERS, in the +same socket write. """. -spec complete_h2_reply(#state{}, nhttp_lib:stream_id(), #h2_stream{}, nhttp_lib:response()) -> #state{}. complete_h2_reply( - #state{protocol_state = #h2_state{} = H2} = State, + #state{protocol_state = #h2_state{conn_credit = {on_response, _, _}}} = State, StreamId, Stream, - #{status := Status} = Response + Response ) -> - {NewH2Conn, NewStreams} = send_h2_response(State, StreamId, Response), - State1 = State#state{ - protocol_state = H2#h2_state{h2_conn = NewH2Conn, h2_streams = NewStreams} - }, - State2 = maybe_rst_stream_unread_body(State1, StreamId, Stream), - State3 = release_request_worker(State2, StreamId, Stream), - nhttp_conn:emit_request_stop(State3, Status, Stream#h2_stream.req_span), - State3#state{requests_count = State3#state.requests_count + 1}. + {State1, Credit} = release_response_credit(State, StreamId, Stream), + finish_h2_reply(State1, StreamId, Stream, Response, Credit); +complete_h2_reply( + #state{protocol_state = #h2_state{stream_credit = on_response}} = State, + StreamId, + Stream, + Response +) -> + {State1, Credit} = release_response_credit(State, StreamId, Stream), + finish_h2_reply(State1, StreamId, Stream, Response, Credit); +complete_h2_reply(State, StreamId, Stream, Response) -> + finish_h2_reply(State, StreamId, Stream, Response, []). -spec dispatch_h2_request(#state{}, nhttp_lib:stream_id(), nhttp_lib:request()) -> #state{}. dispatch_h2_request(#state{limits = Limits} = State, StreamId, Request) -> @@ -357,6 +367,25 @@ dispatch_h2_streaming_request(#state{limits = Limits} = State, StreamId, Request handle_h2_limit_error(State, StreamId, LimitError) end. +-spec finish_h2_reply( + #state{}, nhttp_lib:stream_id(), #h2_stream{}, nhttp_lib:response(), iodata() +) -> #state{}. +finish_h2_reply( + #state{protocol_state = #h2_state{} = H2} = State, + StreamId, + Stream, + #{status := Status} = Response, + Credit +) -> + {NewH2Conn, NewStreams} = send_h2_response(State, StreamId, Response, Credit), + State1 = State#state{ + protocol_state = H2#h2_state{h2_conn = NewH2Conn, h2_streams = NewStreams} + }, + State2 = maybe_rst_stream_unread_body(State1, StreamId, Stream), + State3 = release_request_worker(State2, StreamId, Stream), + nhttp_conn:emit_request_stop(State3, Status, Stream#h2_stream.req_span), + State3#state{requests_count = State3#state.requests_count + 1}. + -spec spawn_h2_request_worker( #state{}, nhttp_lib:stream_id(), nhttp_lib:request(), boolean() ) -> #state{}. @@ -419,6 +448,112 @@ abort_stream_worker( State end. +-spec apply_conn_credit(#state{}, nhttp_lib:stream_id(), pos_integer()) -> #state{}. +apply_conn_credit( + #state{protocol_state = #h2_state{conn_credit = eager}} = State, _StreamId, Size +) -> + credit_connection(State, Size); +apply_conn_credit( + #state{protocol_state = #h2_state{conn_credit = never}} = State, _StreamId, _Size +) -> + State; +apply_conn_credit( + #state{protocol_state = #h2_state{conn_credit = {threshold, N, Acc}} = H2} = State, + _StreamId, + Size +) -> + case Acc + Size of + Total when Total >= N -> + Reset = State#state{protocol_state = H2#h2_state{conn_credit = {threshold, N, 0}}}, + credit_connection(Reset, Total); + Total -> + State#state{protocol_state = H2#h2_state{conn_credit = {threshold, N, Total}}} + end; +apply_conn_credit( + #state{protocol_state = #h2_state{conn_credit = {delay, Ms, Due}} = H2} = State, + StreamId, + Size +) -> + Now = erlang:monotonic_time(millisecond), + DueAt = Now + Ms, + Queued = State#state{ + protocol_state = H2#h2_state{ + conn_credit = {delay, Ms, queue:in({DueAt, StreamId, Size}, Due)} + } + }, + arm_credit_timer(Queued, DueAt, Now); +apply_conn_credit( + #state{ + protocol_state = #h2_state{conn_credit = {on_response, _, _}, h2_streams = Streams} = H2 + } = State, + StreamId, + Size +) -> + #h2_stream{conn_uncredited = Acc} = Stream = maps:get(StreamId, Streams), + State#state{ + protocol_state = H2#h2_state{ + h2_streams = Streams#{StreamId => Stream#h2_stream{conn_uncredited = Acc + Size}} + } + }. + +-spec apply_stream_credit(#state{}, nhttp_lib:stream_id(), pos_integer()) -> #state{}. +apply_stream_credit( + #state{protocol_state = #h2_state{stream_credit = eager}} = State, StreamId, Size +) -> + credit_stream(State, StreamId, Size); +apply_stream_credit( + #state{protocol_state = #h2_state{stream_credit = never}} = State, _StreamId, _Size +) -> + State; +apply_stream_credit( + #state{ + protocol_state = #h2_state{stream_credit = {threshold, N}, h2_streams = Streams} = H2 + } = State, + StreamId, + Size +) -> + #h2_stream{uncredited = Acc} = Stream = maps:get(StreamId, Streams), + case Acc + Size of + Total when Total >= N -> + Reset = State#state{ + protocol_state = H2#h2_state{ + h2_streams = Streams#{StreamId => Stream#h2_stream{uncredited = 0}} + } + }, + credit_stream(Reset, StreamId, Total); + Total -> + State#state{ + protocol_state = H2#h2_state{ + h2_streams = Streams#{StreamId => Stream#h2_stream{uncredited = Total}} + } + } + end; +apply_stream_credit( + #state{protocol_state = #h2_state{stream_credit = {delay, Ms, Due}} = H2} = State, + StreamId, + Size +) -> + Now = erlang:monotonic_time(millisecond), + DueAt = Now + Ms, + Queued = State#state{ + protocol_state = H2#h2_state{ + stream_credit = {delay, Ms, queue:in({DueAt, StreamId, Size}, Due)} + } + }, + arm_credit_timer(Queued, DueAt, Now); +apply_stream_credit( + #state{protocol_state = #h2_state{stream_credit = on_response, h2_streams = Streams} = H2} = + State, + StreamId, + Size +) -> + #h2_stream{uncredited = Acc} = Stream = maps:get(StreamId, Streams), + State#state{ + protocol_state = H2#h2_state{ + h2_streams = Streams#{StreamId => Stream#h2_stream{uncredited = Acc + Size}} + } + }. + -spec apply_stream_push_validation( #state{}, nhttp_lib:stream_id(), @@ -485,6 +620,79 @@ apply_stream_push_validation( } end. +-doc """ +Arm the one credit timer of the connection for `DueAt` unless it is +already armed for an earlier or equal instant. A later timer is +cancelled first. Its stale message is ignored by the reference check in +`handle_h2_credit_delay/2`. +""". +-spec arm_credit_timer(#state{}, integer(), integer()) -> #state{}. +arm_credit_timer( + #state{protocol_state = #h2_state{credit_timer = {_Ref, Armed}}} = State, DueAt, _Now +) when + Armed =< DueAt +-> + State; +arm_credit_timer( + #state{protocol_state = #h2_state{credit_timer = Timer} = H2} = State, DueAt, Now +) -> + ok = cancel_credit_timer(Timer), + Ref = erlang:start_timer(max(0, DueAt - Now), self(), ?CREDIT_DELAY_TAG), + State#state{protocol_state = H2#h2_state{credit_timer = {Ref, DueAt}}}. + +-spec cancel_credit_timer({reference(), integer()} | undefined) -> ok. +cancel_credit_timer(undefined) -> + ok; +cancel_credit_timer({Ref, _DueAt}) -> + ok = erlang:cancel_timer(Ref, [{async, true}, {info, false}]), + ok. + +-spec credit_connection(#state{}, pos_integer()) -> #state{}. +credit_connection(#state{protocol_state = #h2_state{h2_conn = H2Conn} = H2} = State, Size) -> + case nhttp_h2:send_window_update(H2Conn, connection, Size) of + {ok, H2Conn1, Frame} -> + ok = nhttp_conn:sock_send(State, Frame), + State#state{protocol_state = H2#h2_state{h2_conn = H2Conn1}}; + {error, _} -> + State + end. + +-doc """ +Turn every full batch of the `on_response` connection accumulator into +one WINDOW_UPDATE frame. A batch of 0 sends the whole accumulator. +""". +-spec credit_connection_batches(#state{}, iodata()) -> {#state{}, iodata()}. +credit_connection_batches( + #state{protocol_state = #h2_state{conn_credit = {on_response, Batch, Acc}} = H2} = State, + Frames +) when Acc > 0, Acc >= Batch -> + Increment = + case Batch of + 0 -> Acc; + _ -> Batch + end, + Popped = State#state{ + protocol_state = H2#h2_state{conn_credit = {on_response, Batch, Acc - Increment}} + }, + {State1, Frame} = window_update_frame(Popped, connection, Increment), + credit_connection_batches(State1, [Frames, Frame]); +credit_connection_batches(State, Frames) -> + {State, Frames}. + +-spec credit_stream(#state{}, nhttp_lib:stream_id(), pos_integer()) -> #state{}. +credit_stream( + #state{protocol_state = #h2_state{h2_conn = H2Conn} = H2} = State, StreamId, Size +) -> + case nhttp_h2:send_window_update(H2Conn, StreamId, Size) of + {ok, _UnknownStream, []} -> + State; + {ok, H2Conn1, Frame} -> + ok = nhttp_conn:sock_send(State, Frame), + State#state{protocol_state = H2#h2_state{h2_conn = H2Conn1}}; + {error, _} -> + State + end. + -spec draw_response_delay(nhttp:h2_response_delay()) -> non_neg_integer(). draw_response_delay(Ms) when is_integer(Ms) -> Ms; @@ -553,15 +761,15 @@ forward_terminal_body_event(WPid, Ref, true, Trailers) -> ok. -doc """ -Worker acked a body chunk. Pop the oldest pending byte size and emit -WINDOW_UPDATE for that many bytes on both the connection-level and the -stream-level flow-control windows so the peer can resume sending. +Worker acked a body chunk. Pop the oldest pending byte size and hand it +to the credit policies of the connection and the stream receive windows. Acks for `fin` / `abort` events carry no flow-control debt and are absorbed silently when the pending queue is empty. """. -spec handle_h2_body_chunk_ack(#state{}, pid(), reference()) -> #state{}. handle_h2_body_chunk_ack( - #state{protocol_state = #h2_state{h2_streams = Streams, h2_workers = Workers}} = State, + #state{protocol_state = #h2_state{h2_streams = Streams, h2_workers = Workers} = H2} = + State, WPid, Ref ) -> @@ -576,14 +784,12 @@ handle_h2_body_chunk_ack( case queue:out(Pending) of {{value, Size}, Rest} -> Stream1 = Stream#h2_stream{body_window_pending = Rest}, - State1 = replenish_recv_window(State, StreamId, Size), - #state{protocol_state = #h2_state{h2_streams = Streams1} = H2After} = - State1, - State1#state{ - protocol_state = H2After#h2_state{ - h2_streams = Streams1#{StreamId => Stream1} + Popped = State#state{ + protocol_state = H2#h2_state{ + h2_streams = Streams#{StreamId => Stream1} } - }; + }, + replenish_recv_window(Popped, StreamId, Size); {empty, _Rest} -> State end; @@ -593,6 +799,22 @@ handle_h2_body_chunk_ack( end ). +-doc """ +Credit timer fired. Sends every delayed credit that is due on both +windows, oldest first, and re-arms the timer for the earliest entry +left. A timer reference that no longer matches is ignored. +""". +-spec handle_h2_credit_delay(#state{}, reference()) -> #state{}. +handle_h2_credit_delay( + #state{protocol_state = #h2_state{credit_timer = {TimerRef, _DueAt}} = H2} = State, TimerRef +) -> + Now = erlang:monotonic_time(millisecond), + Disarmed = State#state{protocol_state = H2#h2_state{credit_timer = undefined}}, + Released = release_due_stream_credit(release_due_conn_credit(Disarmed, Now), Now), + rearm_credit_timer(Released, Now); +handle_h2_credit_delay(State, _TimerRef) -> + State. + -doc """ DATA event on a stream that has been dispatched to a worker. Buffers into `body_acc` while the worker is still in `handle_request/2`. Once @@ -715,6 +937,25 @@ handle_h2_streaming_body_too_large(State, StreamId, Stream) -> end, handle_h2_limit_error(State, StreamId, body_too_large). +-spec head_credit_due(h2_conn_credit() | h2_stream_credit()) -> integer() | undefined. +head_credit_due(eager) -> + undefined; +head_credit_due(never) -> + undefined; +head_credit_due({threshold, _N}) -> + undefined; +head_credit_due({threshold, _N, _Acc}) -> + undefined; +head_credit_due(on_response) -> + undefined; +head_credit_due({on_response, _Batch, _Acc}) -> + undefined; +head_credit_due({delay, _Ms, Due}) -> + case queue:peek(Due) of + {value, {DueAt, _StreamId, _Size}} -> DueAt; + empty -> undefined + end. + -doc """ Park a compressed `{reply, _, _}` response on its stream until the response-delay timer fires. The worker exits right after it posts the @@ -750,6 +991,12 @@ hold_h2_reply( } }. +-spec join_credit(iodata(), iodata()) -> [iodata()]. +join_credit([], []) -> + []; +join_credit(ConnFrames, StreamFrames) -> + [ConnFrames, StreamFrames]. + -doc """ After a terminal handler result (`reply` / `stream`) on a request whose body was still in flight, send RST_STREAM(NO_ERROR) to ask the peer to @@ -763,6 +1010,69 @@ maybe_rst_stream_unread_body(State, StreamId, #h2_stream{end_stream = false}) -> send_h2_rst_stream_with_code(State, StreamId, no_error), State. +-spec next_credit_due(integer() | undefined, integer() | undefined) -> integer() | undefined. +next_credit_due(undefined, StreamDue) -> + StreamDue; +next_credit_due(ConnDue, undefined) -> + ConnDue; +next_credit_due(ConnDue, StreamDue) -> + min(ConnDue, StreamDue). + +-spec rearm_credit_timer(#state{}, integer()) -> #state{}. +rearm_credit_timer( + #state{protocol_state = #h2_state{conn_credit = Conn, stream_credit = Stream}} = State, Now +) -> + case next_credit_due(head_credit_due(Conn), head_credit_due(Stream)) of + undefined -> State; + DueAt -> arm_credit_timer(State, DueAt, Now) + end. + +-spec release_conn_response_credit(#state{}, non_neg_integer()) -> {#state{}, iodata()}. +release_conn_response_credit( + #state{protocol_state = #h2_state{conn_credit = {on_response, Batch, Acc}} = H2} = State, + Octets +) -> + Queued = State#state{ + protocol_state = H2#h2_state{conn_credit = {on_response, Batch, Acc + Octets}} + }, + credit_connection_batches(Queued, []); +release_conn_response_credit(State, _Octets) -> + {State, []}. + +-spec release_due_conn_credit(#state{}, integer()) -> #state{}. +release_due_conn_credit( + #state{protocol_state = #h2_state{conn_credit = {delay, Ms, Due}} = H2} = State, Now +) -> + case queue:out(Due) of + {{value, {DueAt, _StreamId, Size}}, Rest} when DueAt =< Now -> + Popped = State#state{protocol_state = H2#h2_state{conn_credit = {delay, Ms, Rest}}}, + release_due_conn_credit(credit_connection(Popped, Size), Now); + {{value, _NotDue}, _Rest} -> + State; + {empty, _Due} -> + State + end; +release_due_conn_credit(State, _Now) -> + State. + +-spec release_due_stream_credit(#state{}, integer()) -> #state{}. +release_due_stream_credit( + #state{protocol_state = #h2_state{stream_credit = {delay, Ms, Due}} = H2} = State, Now +) -> + case queue:out(Due) of + {{value, {DueAt, StreamId, Size}}, Rest} when DueAt =< Now -> + Popped = State#state{ + protocol_state = H2#h2_state{stream_credit = {delay, Ms, Rest}} + }, + release_due_stream_credit(credit_stream(Popped, StreamId, Size), Now); + {{value, _NotDue}, _Rest} -> + State; + {empty, _Due} -> + State + end; +release_due_stream_credit(State, _Now) -> + State. + -spec release_request_worker(#state{}, nhttp_lib:stream_id(), #h2_stream{}) -> #state{}. release_request_worker( #state{protocol_state = #h2_state{h2_streams = Streams, h2_workers = Workers} = H2} = State, @@ -801,30 +1111,48 @@ release_request_worker( end, State#state{protocol_state = H2#h2_state{h2_streams = Streams1, h2_workers = Workers1}}. +-doc """ +Release the credit an `on_response` policy held for the request of +`Stream`. The connection octets move into the batch accumulator and +every full batch becomes one WINDOW_UPDATE. The stream octets become one +WINDOW_UPDATE. The frames go ahead of the response HEADERS, because +HEADERS with END_STREAM closes the stream and the codec refuses credit +on a closed stream. +""". +-spec release_response_credit(#state{}, nhttp_lib:stream_id(), #h2_stream{}) -> + {#state{}, iodata()}. +release_response_credit( + State, StreamId, #h2_stream{uncredited = StreamOctets, conn_uncredited = ConnOctets} +) -> + {State1, ConnFrames} = release_conn_response_credit(State, ConnOctets), + {State2, StreamFrames} = release_stream_response_credit(State1, StreamId, StreamOctets), + {State2, join_credit(ConnFrames, StreamFrames)}. + +-spec release_stream_response_credit(#state{}, nhttp_lib:stream_id(), non_neg_integer()) -> + {#state{}, iodata()}. +release_stream_response_credit( + #state{protocol_state = #h2_state{stream_credit = on_response}} = State, StreamId, Octets +) when Octets > 0 -> + window_update_frame(State, StreamId, Octets); +release_stream_response_credit(State, _StreamId, _Octets) -> + {State, []}. + +-doc """ +Credit the peer for `Size` consumed octets on both receive windows +through the resolved policies. The `eager` pair is the default and sends +both WINDOW_UPDATE frames at once. +""". -spec replenish_recv_window(#state{}, nhttp_lib:stream_id(), non_neg_integer()) -> #state{}. replenish_recv_window(State, _StreamId, 0) -> State; replenish_recv_window( - #state{protocol_state = #h2_state{h2_conn = H2Conn} = H2} = State, StreamId, Size + #state{protocol_state = #h2_state{conn_credit = eager, stream_credit = eager}} = State, + StreamId, + Size ) -> - State1 = - case nhttp_h2:send_window_update(H2Conn, connection, Size) of - {ok, H2Conn1, Frame1} -> - nhttp_conn:sock_send(State, Frame1), - State#state{protocol_state = H2#h2_state{h2_conn = H2Conn1}}; - {error, _} -> - State - end, - #state{protocol_state = #h2_state{h2_conn = ConnAfter} = H2After} = State1, - case nhttp_h2:send_window_update(ConnAfter, StreamId, Size) of - {ok, _UnknownStream, []} -> - State1; - {ok, H2Conn2, Frame2} -> - nhttp_conn:sock_send(State1, Frame2), - State1#state{protocol_state = H2After#h2_state{h2_conn = H2Conn2}}; - {error, _} -> - State1 - end. + credit_stream(credit_connection(State, Size), StreamId, Size); +replenish_recv_window(State, StreamId, Size) -> + apply_stream_credit(apply_conn_credit(State, StreamId, Size), StreamId, Size). -spec send_h2_rst_stream_with_code(#state{}, nhttp_lib:stream_id(), atom()) -> ok. send_h2_rst_stream_with_code(State, StreamId, ErrorCode) -> @@ -874,6 +1202,25 @@ track_held_stream_end( Ended = Stream#h2_stream{end_stream = true}, State#state{protocol_state = H2#h2_state{h2_streams = Streams#{StreamId => Ended}}}. +-doc """ +Build one WINDOW_UPDATE for `Target` without sending it. A stream the +codec no longer has yields no frame. An overflow keeps the codec state +and yields no frame, which is the codec contract. +""". +-spec window_update_frame(#state{}, connection | nhttp_lib:stream_id(), pos_integer()) -> + {#state{}, iodata()}. +window_update_frame( + #state{protocol_state = #h2_state{h2_conn = H2Conn} = H2} = State, Target, Size +) -> + case nhttp_h2:send_window_update(H2Conn, Target, Size) of + {ok, _UnknownStream, []} -> + {State, []}; + {ok, H2Conn1, Frame} -> + {State#state{protocol_state = H2#h2_state{h2_conn = H2Conn1}}, Frame}; + {error, _} -> + {State, []} + end. + %%%----------------------------------------------------------------------------- %% INTERNAL FUNCTIONS - WORKER MESSAGE HANDLERS %%%----------------------------------------------------------------------------- @@ -1287,6 +1634,12 @@ post_buffer_drain( response_to_headers(#{status := Status, headers := Headers}) -> [{<<":status">>, integer_to_binary(Status)} | Headers]. +-spec send_credit_only(#state{}, iodata()) -> ok. +send_credit_only(_State, []) -> + ok; +send_credit_only(State, Credit) -> + nhttp_conn:sock_send(State, Credit). + -spec send_h2_data_unbuffered( #state{}, nhttp_lib:stream_id(), iodata(), nhttp_h2:fin() ) -> #state{}. @@ -1356,12 +1709,19 @@ send_h2_headers( H2Conn end. --spec send_h2_response(#state{}, nhttp_lib:stream_id(), nhttp_lib:response()) -> +-doc """ +Send a response with `Credit`, the WINDOW_UPDATE frames of an +`on_response` policy, ahead of it in the same socket write. When the +codec refuses the HEADERS the credit still goes out, because the codec +already counted it. +""". +-spec send_h2_response(#state{}, nhttp_lib:stream_id(), nhttp_lib:response(), iodata()) -> {nhttp_h2:conn(), #{nhttp_lib:stream_id() => #h2_stream{}}}. send_h2_response( #state{protocol_state = #h2_state{h2_conn = H2Conn, h2_streams = Streams}} = State, StreamId, - Response + Response, + Credit ) -> Headers = nhttp_conn:alt_svc_headers(State, response_to_headers(Response)), Body = maps:get(body, Response, <<>>), @@ -1369,22 +1729,26 @@ send_h2_response( <<>> -> case nhttp_h2:send_headers(H2Conn, StreamId, Headers, fin) of {ok, NewH2Conn, Frame} -> - nhttp_conn:sock_send(State, Frame), + ok = send_with_credit(State, Credit, Frame), {NewH2Conn, Streams}; {error, {stream_closed, _}} -> + ok = send_credit_only(State, Credit), {H2Conn, maps:remove(StreamId, Streams)}; {error, connection_closing} -> + ok = send_credit_only(State, Credit), {H2Conn, Streams} end; _ -> case nhttp_h2:send_headers(H2Conn, StreamId, Headers, nofin) of {ok, H2Conn1, HeaderFrame} -> send_h2_response_with_body( - State, H2Conn1, Streams, StreamId, HeaderFrame, Body + State, H2Conn1, Streams, StreamId, HeaderFrame, Body, Credit ); {error, {stream_closed, _}} -> + ok = send_credit_only(State, Credit), {H2Conn, maps:remove(StreamId, Streams)}; {error, connection_closing} -> + ok = send_credit_only(State, Credit), {H2Conn, Streams} end end. @@ -1396,15 +1760,16 @@ send_h2_response( #{nhttp_lib:stream_id() => #h2_stream{}}, nhttp_lib:stream_id(), iodata(), - binary() + binary(), + iodata() ) -> {nhttp_h2:conn(), #{nhttp_lib:stream_id() => #h2_stream{}}}. -send_h2_response_with_body(State, H2Conn, Streams, StreamId, HeaderFrame, Body) -> +send_h2_response_with_body(State, H2Conn, Streams, StreamId, HeaderFrame, Body, Credit) -> case nhttp_h2:send_data(H2Conn, StreamId, Body, fin) of {ok, NewH2Conn, DataFrame} -> - nhttp_conn:sock_send(State, [HeaderFrame, DataFrame]), + ok = send_with_credit(State, Credit, [HeaderFrame, DataFrame]), {NewH2Conn, Streams}; {partial, NewH2Conn, DataFrame, Remaining, PendingEndStream, _Window} -> - nhttp_conn:sock_send(State, [HeaderFrame, DataFrame]), + ok = send_with_credit(State, Credit, [HeaderFrame, DataFrame]), Stream = maps:get(StreamId, Streams, #h2_stream{}), NewStream = Stream#h2_stream{ send_buffer = Remaining, @@ -1412,10 +1777,10 @@ send_h2_response_with_body(State, H2Conn, Streams, StreamId, HeaderFrame, Body) }, {NewH2Conn, Streams#{StreamId => NewStream}}; {error, {stream_closed, _}} -> - nhttp_conn:sock_send(State, HeaderFrame), + ok = send_with_credit(State, Credit, HeaderFrame), {H2Conn, maps:remove(StreamId, Streams)}; {error, {unknown_stream, _}} -> - nhttp_conn:sock_send(State, HeaderFrame), + ok = send_with_credit(State, Credit, HeaderFrame), {H2Conn, Streams} end. @@ -1424,3 +1789,9 @@ send_h2_rst_stream(State, StreamId) -> {ok, Frame} = nhttp_h2_frame:rst_stream(StreamId, internal_error), nhttp_conn:sock_send(State, Frame), ok. + +-spec send_with_credit(#state{}, iodata(), iodata()) -> ok. +send_with_credit(State, [], Frames) -> + nhttp_conn:sock_send(State, Frames); +send_with_credit(State, Credit, Frames) -> + nhttp_conn:sock_send(State, [Credit, Frames]). diff --git a/src/nhttp_listener.erl b/src/nhttp_listener.erl index efe665f..11b93f0 100644 --- a/src/nhttp_listener.erl +++ b/src/nhttp_listener.erl @@ -522,6 +522,59 @@ validate_h2_settings(Settings) -> ok end. +-spec validate_h2_credit_batch(nhttp:opts()) -> ok | {error, term()}. +validate_h2_credit_batch(Opts) -> + OnResponse = + maps:get(h2_connection_window_policy, Opts, eager) =:= on_response orelse + maps:get(h2_stream_window_policy, Opts, eager) =:= on_response, + case maps:get(h2_credit_batch, Opts, undefined) of + undefined -> + ok; + V when is_integer(V), V >= 0, V =< ?H2_MAX_WINDOW_SIZE, OnResponse -> + ok; + V when is_integer(V), V >= 0, V =< ?H2_MAX_WINDOW_SIZE -> + {error, + {invalid_h2_credit_batch, V, + "needs h2_connection_window_policy or h2_stream_window_policy on_response"}}; + V -> + {error, {invalid_h2_credit_batch, V, "must be 0..2147483647"}} + end. + +-spec validate_h2_window_policies(nhttp:opts()) -> ok | {error, term()}. +validate_h2_window_policies(Opts) -> + maybe + ok ?= + validate_h2_window_policy( + h2_connection_window_policy, maps:get(h2_connection_window_policy, Opts, eager) + ), + ok ?= + validate_h2_window_policy( + h2_stream_window_policy, maps:get(h2_stream_window_policy, Opts, eager) + ), + ok ?= validate_h2_credit_batch(Opts), + ok + end. + +-spec validate_h2_window_policy( + h2_connection_window_policy | h2_stream_window_policy, nhttp:h2_window_policy() | term() +) -> ok | {error, term()}. +validate_h2_window_policy(_Key, eager) -> + ok; +validate_h2_window_policy(_Key, never) -> + ok; +validate_h2_window_policy(_Key, {threshold, N}) when + is_integer(N), N >= 1, N =< ?H2_MAX_WINDOW_SIZE +-> + ok; +validate_h2_window_policy(_Key, {delay, Ms}) when is_integer(Ms), Ms >= 0 -> + ok; +validate_h2_window_policy(_Key, on_response) -> + ok; +validate_h2_window_policy(Key, V) -> + {error, + {invalid_h2_window_policy, Key, V, + "must be eager, {threshold, 1..2147483647}, {delay, Ms}, on_response or never"}}. + -spec validate_header_table_size(non_neg_integer() | undefined) -> ok | {error, term()}. validate_header_table_size(undefined) -> ok; @@ -561,6 +614,7 @@ validate_opts(Opts) -> ok ?= validate_h2_settings(maps:get(h2_settings, Opts, #{})), ok ?= validate_h2_aliases(Opts), ok ?= validate_h2_response_delay(maps:get(h2_response_delay, Opts, undefined)), + ok ?= validate_h2_window_policies(Opts), ok ?= validate_proxy_protocol(maps:get(proxy_protocol, Opts, false)), ok ?= validate_acceptor_count(maps:get(acceptor_count, Opts, undefined)), ok ?= validate_alt_svc(maps:get(alt_svc, Opts, #{})), diff --git a/test/nhttp_h2_credit_policy_SUITE.erl b/test/nhttp_h2_credit_policy_SUITE.erl index 32dc39b..ba6e0fa 100644 --- a/test/nhttp_h2_credit_policy_SUITE.erl +++ b/test/nhttp_h2_credit_policy_SUITE.erl @@ -17,25 +17,54 @@ ]). -export([ + connection_never_credits_nothing/1, + connection_threshold_credits_accumulated_total/1, + credit_sum_bounded_by_connection_policy/1, + credit_sum_bounded_by_stream_policy/1, + delay_credits_after_the_delay/1, initial_window_size_alias_reaches_codec/1, max_frame_size_alias_reaches_codec/1, + on_response_batch_credits_with_the_batch_response/1, + on_response_credits_each_response/1, response_delay_holds_concurrent_streams/1, response_delay_holds_reply/1, response_delay_reply_only/1, response_delay_reset_drops_held_reply/1, - response_delay_uniform_draws_per_response/1 + response_delay_uniform_draws_per_response/1, + stream_and_connection_policies_are_independent/1 ]). -behaviour(nhttp_handler). -export([init/1, handle_request/2, handle_request_body/3]). +-define(BATCH, 32830). +-define(BATCH_POSTS, 734). +-define(BATCH_RESPONSES, (?BATCH div ?CHUNK)). -define(CANCEL, 8). +-define(CHUNK, 134). +-define(CLIENT_WINDOW_RAISE, 2147418112). +-define(CONN_WINDOW, 65535). +-define(CHUNK_SPACING_MS, 40). +-define(CREDIT_DELAY_MS, 200). +-define(DELAY_CHUNKS, 5). -define(DELAY_MS, 300). -define(FRAME_SIZE_ERROR, 6). +-define(MAX_STREAMS, 1000). -define(MEASURE_SLACK_MS, 50). +-define(NEVER_POSTS, 200). +-define(NO_CREDIT_MS, 100). +-define(POLICY_POSTS, 20). -define(RECV_TIMEOUT, 3000). -define(RESET_AFTER_MS, 100). -define(SETTINGS_INITIAL_WINDOW_SIZE, 4). +-define(SETTLE_MS, 300). +-define(SUM_CHUNK, 100). +-define(SUM_CHUNKS, 4). +-define(SUM_DELAY_MS, 100). +-define(SUM_STREAMS, 5). +-define(SUM_THRESHOLD, 200). +-define(THRESHOLD_CYCLES, 2). +-define(THRESHOLD_POSTS, 10). -define(UNIFORM_DRAWS, 20). -define(UNIFORM_MAX_MS, 200). -define(UNIFORM_MIN_MS, 100). @@ -71,13 +100,21 @@ send_one_chunk(SendChunk) -> all() -> [ + connection_never_credits_nothing, + connection_threshold_credits_accumulated_total, + credit_sum_bounded_by_connection_policy, + credit_sum_bounded_by_stream_policy, + delay_credits_after_the_delay, initial_window_size_alias_reaches_codec, max_frame_size_alias_reaches_codec, + on_response_batch_credits_with_the_batch_response, + on_response_credits_each_response, response_delay_holds_concurrent_streams, response_delay_holds_reply, response_delay_reply_only, response_delay_reset_drops_held_reply, - response_delay_uniform_draws_per_response + response_delay_uniform_draws_per_response, + stream_and_connection_policies_are_independent ]. init_per_suite(Config) -> @@ -102,6 +139,70 @@ end_per_testcase(_TC, _Config) -> %%% TEST CASES %%%----------------------------------------------------------------------------- +connection_never_credits_nothing(_Config) -> + Body = binary:copy(<<$n>>, ?CHUNK), + StreamIds = stream_ids(?NEVER_POSTS), + All = all_frames(run_posts(#{h2_connection_window_policy => never}, StreamIds, Body)), + ?assertEqual([], window_updates(All, 0)), + ?assertEqual([], [Id || Id <- StreamIds, response_body(All, Id) =/= Body]). + +connection_threshold_credits_accumulated_total(_Config) -> + Threshold = ?THRESHOLD_POSTS * ?CHUNK, + Body = binary:copy(<<$t>>, ?CHUNK), + StreamIds = stream_ids(?THRESHOLD_CYCLES * ?THRESHOLD_POSTS), + {PerStream, Tail} = run_posts( + #{h2_connection_window_policy => {threshold, Threshold}}, StreamIds, Body + ), + Credited = [{Id, window_updates(Frames, 0)} || {Id, Frames} <- PerStream], + ?assertEqual( + [{stream_id(N * ?THRESHOLD_POSTS), [Threshold]} || N <- lists:seq(1, ?THRESHOLD_CYCLES)], + [Entry || {_, Increments} = Entry <- Credited, Increments =/= []] + ), + ?assertEqual([], window_updates(Tail, 0)). + +credit_sum_bounded_by_connection_policy(_Config) -> + lists:foreach( + fun(Shape) -> + assert_credit_sums(policy_opts(h2_connection_window_policy, Shape), Shape, eager) + end, + shapes() + ). + +credit_sum_bounded_by_stream_policy(_Config) -> + lists:foreach( + fun(Shape) -> + assert_credit_sums(policy_opts(h2_stream_window_policy, Shape), eager, Shape) + end, + shapes() + ). + +delay_credits_after_the_delay(_Config) -> + {Pid, Port} = nhttp_test_helpers:h2_start_server(?MODULE, #{ + h2_connection_window_policy => {delay, ?CREDIT_DELAY_MS}, + h2_stream_window_policy => {delay, ?CREDIT_DELAY_MS} + }), + try + {ok, Sock} = nhttp_test_helpers:h2_connect(Port), + Chunk = binary:copy(<<$d>>, ?CHUNK), + Total = ?DELAY_CHUNKS * ?CHUNK, + ok = nhttp_test_helpers:h2_send_headers(Sock, 1, <<"/echo">>, Total, false), + SentAt = send_spaced_chunks(Sock, 1, Chunk, ?DELAY_CHUNKS), + {Timed, Rest} = collect_window_updates(Sock, <<>>, [0, 1], ?DELAY_CHUNKS, ?RECV_TIMEOUT), + ConnAt = [At || {At, {window_update, 0, _}} <- Timed], + StreamAt = [At || {At, {window_update, 1, _}} <- Timed], + ?assertEqual([], early_credits(ConnAt, SentAt)), + ?assertEqual([], early_credits(StreamAt, SentAt)), + Frames = [F || {_, F} <- Timed], + ?assertEqual(Total, lists:sum(window_updates(Frames, 0))), + ?assertEqual(Total, lists:sum(window_updates(Frames, 1))), + ok = nhttp_test_helpers:h2_send_data(Sock, 1, <<>>, true), + {Reply, _, _} = await_streams(Sock, [1], Rest, ?RECV_TIMEOUT), + ?assertEqual(binary:copy(Chunk, ?DELAY_CHUNKS), response_body(Reply, 1)), + ssl:close(Sock) + after + nhttp:stop(Pid) + end. + initial_window_size_alias_reaches_codec(_Config) -> {PidAlias, PortAlias} = nhttp_test_helpers:h2_start_server(?MODULE, #{ h2_initial_window_size => ?WINDOW @@ -154,6 +255,72 @@ max_frame_size_alias_reaches_codec(_Config) -> nhttp:stop(PidBig) end. +on_response_batch_credits_with_the_batch_response(_Config) -> + {Pid, Port} = nhttp_test_helpers:h2_start_server(?MODULE, #{ + h2_connection_window_policy => on_response, + h2_credit_batch => ?BATCH, + h2_stream_window_policy => never, + h2_response_delay => ?CREDIT_DELAY_MS, + h2_settings => #{max_concurrent_streams => ?MAX_STREAMS} + }), + try + {ok, Sock} = nhttp_test_helpers:h2_connect(Port), + ok = nhttp_test_helpers:h2_send_window_update(Sock, 0, ?CLIENT_WINDOW_RAISE), + Body = binary:copy(<<$b>>, ?CHUNK), + StreamIds = stream_ids(?BATCH_POSTS), + {Reads, Rest} = post_within_window(Sock, StreamIds, Body, ?CONN_WINDOW), + {Tail, _} = collect(Sock, Rest, ?SETTLE_MS), + All = lists:append(Reads) ++ Tail, + ?assertEqual([], [Id || Id <- StreamIds, response_body(All, Id) =/= Body]), + ?assertEqual([?BATCH, ?BATCH], window_updates(All, 0)), + ?assertEqual([], [F || {window_update, SId, _} = F <- All, SId > 0]), + ?assertEqual( + [{window_update, 0, ?BATCH}, {window_update, 0, ?BATCH}], + [frame_before_nth_headers(Reads, N) || N <- [?BATCH_RESPONSES, 2 * ?BATCH_RESPONSES]] + ), + ?assertEqual( + (?BATCH_POSTS - 2 * ?BATCH_RESPONSES) * ?CHUNK, + ?BATCH_POSTS * ?CHUNK - lists:sum(window_updates(All, 0)) + ), + ssl:close(Sock) + after + nhttp:stop(Pid) + end. + +on_response_credits_each_response(_Config) -> + {Pid, Port} = nhttp_test_helpers:h2_start_server(?MODULE, #{ + h2_connection_window_policy => on_response, + h2_stream_window_policy => on_response, + h2_response_delay => ?CREDIT_DELAY_MS + }), + try + {ok, Sock} = nhttp_test_helpers:h2_connect(Port), + Body = binary:copy(<<$e>>, ?CHUNK), + StreamIds = stream_ids(?POLICY_POSTS), + lists:foreach( + fun(Id) -> ok = nhttp_test_helpers:h2_send_post(Sock, Id, <<"/echo">>, Body) end, + StreamIds + ), + {Early, Rest} = collect(Sock, <<>>, ?NO_CREDIT_MS), + ?assertEqual([], [F || F <- Early, not is_settings(F)]), + {Reads, _} = await_stream_reads(Sock, StreamIds, Rest, ?RECV_TIMEOUT), + All = lists:append(Reads), + ?assertEqual([], [Id || Id <- StreamIds, response_body(All, Id) =/= Body]), + ?assertEqual(lists:duplicate(?POLICY_POSTS, ?CHUNK), window_updates(All, 0)), + ?assertEqual( + [], + [ + Id + || Id <- StreamIds, + credits_ahead_of_headers(Reads, Id) =/= + [{window_update, 0, ?CHUNK}, {window_update, Id, ?CHUNK}] + ] + ), + ssl:close(Sock) + after + nhttp:stop(Pid) + end. + response_delay_holds_concurrent_streams(_Config) -> {Pid, Port} = nhttp_test_helpers:h2_start_server(?MODULE, #{h2_response_delay => ?DELAY_MS}), try @@ -253,10 +420,82 @@ response_delay_uniform_draws_per_response(_Config) -> nhttp:stop(Pid) end. +stream_and_connection_policies_are_independent(_Config) -> + Body = binary:copy(<<$i>>, ?CHUNK), + StreamIds = stream_ids(?POLICY_POSTS), + StreamNever = all_frames(run_posts(#{h2_stream_window_policy => never}, StreamIds, Body)), + ?assertEqual([], [F || {window_update, SId, _} = F <- StreamNever, SId > 0]), + ?assertEqual(lists:duplicate(?POLICY_POSTS, ?CHUNK), window_updates(StreamNever, 0)), + ConnNever = all_frames( + run_posts( + #{h2_connection_window_policy => never, h2_stream_window_policy => eager}, + StreamIds, + Body + ) + ), + ?assertEqual([], window_updates(ConnNever, 0)), + ?assertEqual([], [Id || Id <- StreamIds, window_updates(ConnNever, Id) =/= [?CHUNK]]). + %%%----------------------------------------------------------------------------- %%% HELPERS %%%----------------------------------------------------------------------------- +all_frames({PerStream, Tail}) -> + lists:append([Frames || {_, Frames} <- PerStream]) ++ Tail. + +assert_credit(Window, Shape, Consumed, Sum) -> + ?assert(Sum =< Consumed), + ?assertEqual(expected_credit(Window, Shape, Consumed), Sum). + +%% Open ?SUM_STREAMS streams, send ?SUM_CHUNKS chunks on each, let the +%% credit settle while the streams are still open, then close them and +%% check the WINDOW_UPDATE sums per window against the shapes. +assert_credit_sums(Opts, ConnShape, StreamShape) -> + {Pid, Port} = nhttp_test_helpers:h2_start_server(?MODULE, Opts), + try + {ok, Sock} = nhttp_test_helpers:h2_connect(Port), + StreamIds = stream_ids(?SUM_STREAMS), + PerStream = ?SUM_CHUNKS * ?SUM_CHUNK, + Chunk = binary:copy(<<$s>>, ?SUM_CHUNK), + lists:foreach( + fun(Id) -> + ok = nhttp_test_helpers:h2_send_headers(Sock, Id, <<"/echo">>, PerStream, false) + end, + StreamIds + ), + lists:foreach( + fun(_Round) -> + lists:foreach( + fun(Id) -> ok = nhttp_test_helpers:h2_send_data(Sock, Id, Chunk, false) end, + StreamIds + ) + end, + lists:seq(1, ?SUM_CHUNKS) + ), + {Credits, Rest} = collect(Sock, <<>>, ?SETTLE_MS), + lists:foreach( + fun(Id) -> ok = nhttp_test_helpers:h2_send_data(Sock, Id, <<>>, true) end, + StreamIds + ), + {Replies, _, Rest1} = await_streams(Sock, StreamIds, Rest, ?RECV_TIMEOUT), + {Tail, _} = collect(Sock, Rest1, ?SETTLE_MS), + All = Credits ++ Replies ++ Tail, + Echoed = binary:copy(Chunk, ?SUM_CHUNKS), + ?assertEqual([], [Id || Id <- StreamIds, response_body(All, Id) =/= Echoed]), + assert_credit( + connection, ConnShape, ?SUM_STREAMS * PerStream, lists:sum(window_updates(All, 0)) + ), + lists:foreach( + fun(Id) -> + assert_credit(stream, StreamShape, PerStream, lists:sum(window_updates(All, Id))) + end, + StreamIds + ), + ssl:close(Sock) + after + nhttp:stop(Pid) + end. + %% Receive until every stream in StreamIds is done. Returns the frames, a %% map from stream id to the time its first HEADERS frame was decoded, and %% the undecoded tail of the socket buffer. @@ -275,8 +514,66 @@ await_streams(Sock, StreamIds, Buf, Acc, Seen, Timeout) -> await_streams(Sock, StreamIds, Rest, Acc ++ Frames, Seen1, Timeout - (now_ms() - T0)) end. -frames_before_headers(Frames, StreamId) -> - lists:takewhile(fun(F) -> not is_headers(F, StreamId) end, Frames). +%% Receive until every stream in StreamIds is done. Returns the frames +%% grouped by socket read, in order, and the undecoded tail of the +%% socket buffer. A frame split across two reads belongs to the later one. +await_stream_reads(Sock, StreamIds, Buf, Timeout) -> + await_stream_reads(Sock, StreamIds, Buf, Timeout, []). + +await_stream_reads(_Sock, [], Buf, _Timeout, Reads) -> + {lists:reverse(Reads), Buf}; +await_stream_reads(Sock, Pending, Buf, Timeout, Reads) -> + T0 = now_ms(), + {ok, Data} = ssl:recv(Sock, 0, Timeout), + {New, Rest} = nhttp_test_helpers:decode_h2_frames(<>), + Left = [Id || Id <- Pending, not nhttp_test_helpers:h2_stream_done(New, Id)], + await_stream_reads(Sock, Left, Rest, Timeout - (now_ms() - T0), [New | Reads]). + +%% Receive whatever arrives within Timeout. Returns the frames and the +%% undecoded tail of the socket buffer. +collect(Sock, Buf, Timeout) -> + collect(Sock, Buf, Timeout, []). + +collect(_Sock, Buf, Timeout, Acc) when Timeout =< 0 -> + {Acc, Buf}; +collect(Sock, Buf, Timeout, Acc) -> + T0 = now_ms(), + case ssl:recv(Sock, 0, Timeout) of + {ok, Data} -> + {Frames, Rest} = nhttp_test_helpers:decode_h2_frames(<>), + collect(Sock, Rest, Timeout - (now_ms() - T0), Acc ++ Frames); + {error, timeout} -> + {Acc, Buf} + end. + +%% Receive until every stream in StreamIds has N WINDOW_UPDATE frames. +%% Returns each frame with the time its socket read completed. +collect_window_updates(Sock, Buf, StreamIds, N, Timeout) -> + collect_window_updates(Sock, Buf, StreamIds, N, Timeout, []). + +collect_window_updates(Sock, Buf, StreamIds, N, Timeout, Acc) -> + Frames = [F || {_, F} <- Acc], + case lists:all(fun(Id) -> length(window_updates(Frames, Id)) >= N end, StreamIds) of + true -> + {Acc, Buf}; + false -> + T0 = now_ms(), + {ok, Data} = ssl:recv(Sock, 0, Timeout), + {New, Rest} = nhttp_test_helpers:decode_h2_frames(<>), + At = now_ms(), + Timed = [{At, F} || F <- New], + collect_window_updates(Sock, Rest, StreamIds, N, Timeout - (At - T0), Acc ++ Timed) + end. + +%% The run of WINDOW_UPDATE frames right ahead of the HEADERS frame of +%% StreamId inside the socket read that carried it. +credits_ahead_of_headers(Reads, StreamId) -> + [Read] = [R || R <- Reads, lists:any(fun(F) -> is_headers(F, StreamId) end, R)], + Before = lists:reverse(frames_before_headers(Read, StreamId)), + lists:reverse(lists:takewhile(fun is_window_update/1, Before)). + +early_credits(CreditedAt, SentAt) -> + [At - T || {At, T} <- lists:zip(CreditedAt, SentAt), At - T < ?CREDIT_DELAY_MS]. error_scope(Frames, StreamId, Code) -> Goaway = [C || {goaway, 0, <<_Last:32, C:32, _/binary>>} <- Frames, C =:= Code], @@ -287,9 +584,33 @@ error_scope(Frames, StreamId, Code) -> {[], []} -> {no_error_seen, Frames} end. +expected_credit(_Window, never, _Consumed) -> + 0; +expected_credit(connection, {on_response, Batch}, Consumed) when Batch > 0 -> + Consumed - Consumed rem Batch; +expected_credit(_Window, _Shape, Consumed) -> + Consumed. + +%% The frame that precedes the Nth HEADERS frame of the run inside the +%% socket read that carried it, or `first_in_read' when none does. +frame_before_nth_headers([Read | Reads], N) -> + case nth_headers(Read, N, first_in_read) of + {found, Before} -> Before; + {remaining, Left} -> frame_before_nth_headers(Reads, Left) + end. + +frames_before_headers(Frames, StreamId) -> + lists:takewhile(fun(F) -> not is_headers(F, StreamId) end, Frames). + is_headers({headers, SId, _, _}, StreamId) -> SId =:= StreamId; is_headers(_, _) -> false. +is_settings({settings, 0, _}) -> true; +is_settings(_) -> false. + +is_window_update({window_update, _, _}) -> true; +is_window_update(_) -> false. + measure_reply(StreamId, {Sock, Buf, Delays}) -> T0 = now_ms(), ok = nhttp_test_helpers:h2_send_request(Sock, StreamId, <<"/hello">>), @@ -313,5 +634,96 @@ note_headers_seen(Frames, StreamIds, Seen, At) -> now_ms() -> erlang:monotonic_time(millisecond). +nth_headers([], N, _Prev) -> + {remaining, N}; +nth_headers([{headers, _, _, _} | _], 1, Prev) -> + {found, Prev}; +nth_headers([{headers, _, _, _} = F | Rest], N, _Prev) -> + nth_headers(Rest, N - 1, F); +nth_headers([F | Rest], N, _Prev) -> + nth_headers(Rest, N, F). + +policy_opts(Key, {on_response, Batch}) -> + #{Key => on_response, h2_credit_batch => Batch}; +policy_opts(Key, Shape) -> + #{Key => Shape}. + +%% POST Body on each stream in turn and await its response. Returns the +%% frames per stream and the undecoded tail of the socket buffer. +post_each(Sock, StreamIds, Body) -> + post_each(Sock, StreamIds, Body, <<>>, []). + +post_each(_Sock, [], _Body, Buf, Acc) -> + {lists:reverse(Acc), Buf}; +post_each(Sock, [Id | Ids], Body, Buf, Acc) -> + ok = nhttp_test_helpers:h2_send_post(Sock, Id, <<"/echo">>, Body), + {Frames, _Seen, Rest} = await_streams(Sock, [Id], Buf, ?RECV_TIMEOUT), + post_each(Sock, Ids, Body, Rest, [{Id, Frames} | Acc]). + +%% POST Body on the streams in waves. A wave holds as many bodies as the +%% connection window allows. The responses of a wave are awaited before +%% the next wave, and the connection increments they carry refill the +%% window. Returns the frames grouped by socket read and the tail. +post_within_window(Sock, StreamIds, Body, Window) -> + post_within_window(Sock, StreamIds, Body, Window, <<>>, []). + +post_within_window(_Sock, [], _Body, _Window, Buf, Reads) -> + {Reads, Buf}; +post_within_window(Sock, StreamIds, Body, Window, Buf, Reads) -> + {[_ | _] = Wave, Later} = lists:split( + min(length(StreamIds), Window div byte_size(Body)), StreamIds + ), + lists:foreach( + fun(Id) -> ok = nhttp_test_helpers:h2_send_post(Sock, Id, <<"/echo">>, Body) end, Wave + ), + {WaveReads, Rest} = await_stream_reads(Sock, Wave, Buf, ?RECV_TIMEOUT), + Refill = lists:sum(window_updates(lists:append(WaveReads), 0)), + Window1 = Window - length(Wave) * byte_size(Body) + Refill, + post_within_window(Sock, Later, Body, Window1, Rest, Reads ++ WaveReads). + +response_body(Frames, StreamId) -> + nhttp_test_helpers:h2_response_body(Frames, StreamId). + rst_streams(Frames) -> [F || {rst_stream, _, _} = F <- Frames]. + +%% Start a server with Opts, POST Body on every stream in turn, then drain +%% the socket for ?SETTLE_MS. Returns the frames per stream and the tail. +run_posts(Opts, StreamIds, Body) -> + {Pid, Port} = nhttp_test_helpers:h2_start_server(?MODULE, Opts), + try + {ok, Sock} = nhttp_test_helpers:h2_connect(Port), + {PerStream, Rest} = post_each(Sock, StreamIds, Body), + {Tail, _} = collect(Sock, Rest, ?SETTLE_MS), + ssl:close(Sock), + {PerStream, Tail} + after + nhttp:stop(Pid) + end. + +send_spaced_chunks(_Sock, _StreamId, _Chunk, 0) -> + []; +send_spaced_chunks(Sock, StreamId, Chunk, N) -> + T = now_ms(), + ok = nhttp_test_helpers:h2_send_data(Sock, StreamId, Chunk, false), + timer:sleep(?CHUNK_SPACING_MS), + [T | send_spaced_chunks(Sock, StreamId, Chunk, N - 1)]. + +shapes() -> + [ + eager, + {threshold, ?SUM_THRESHOLD}, + {delay, ?SUM_DELAY_MS}, + {on_response, 0}, + {on_response, ?BATCH}, + never + ]. + +stream_id(N) -> + 2 * N - 1. + +stream_ids(Count) -> + [stream_id(N) || N <- lists:seq(1, Count)]. + +window_updates(Frames, StreamId) -> + [Inc || {window_update, SId, Inc} <- Frames, SId =:= StreamId]. diff --git a/test/nhttp_listener_validation_SUITE.erl b/test/nhttp_listener_validation_SUITE.erl index 2364add..302f7a3 100644 --- a/test/nhttp_listener_validation_SUITE.erl +++ b/test/nhttp_listener_validation_SUITE.erl @@ -29,7 +29,9 @@ acceptor_sys_terminate/1, listener_h2_alias_conflicts_with_settings/1, listener_h2_alias_out_of_range/1, + listener_h2_credit_batch_invalid/1, listener_h2_response_delay_invalid/1, + listener_h2_window_policy_invalid/1, listener_invalid_tls/1, listener_invalid_versions/1, listener_listen_failed/1, @@ -46,7 +48,9 @@ all() -> acceptor_sys_terminate, listener_h2_alias_conflicts_with_settings, listener_h2_alias_out_of_range, + listener_h2_credit_batch_invalid, listener_h2_response_delay_invalid, + listener_h2_window_policy_invalid, listener_invalid_tls, listener_invalid_versions, listener_listen_failed, @@ -148,6 +152,81 @@ listener_h2_response_delay_invalid(_Config) -> ), ok. +listener_h2_credit_batch_invalid(_Config) -> + Base = #{port => 0, handler => ?MODULE, versions => [http1_1]}, + Invalid = [ + #{h2_credit_batch => 0}, + #{h2_credit_batch => 32830, h2_connection_window_policy => eager}, + #{h2_credit_batch => 32830, h2_stream_window_policy => {threshold, 1}}, + #{h2_credit_batch => -1, h2_connection_window_policy => on_response}, + #{h2_credit_batch => 1.5, h2_connection_window_policy => on_response}, + #{h2_credit_batch => 2147483648, h2_connection_window_policy => on_response}, + #{h2_credit_batch => big, h2_stream_window_policy => on_response} + ], + Valid = [ + #{h2_credit_batch => 0, h2_connection_window_policy => on_response}, + #{h2_credit_batch => 32830, h2_stream_window_policy => on_response}, + #{ + h2_credit_batch => 2147483647, + h2_connection_window_policy => on_response, + h2_stream_window_policy => on_response + } + ], + lists:foreach( + fun(Opts) -> + assert_error_contains( + "invalid_h2_credit_batch", nhttp:start_link(maps:merge(Base, Opts)) + ) + end, + Invalid + ), + lists:foreach( + fun(Opts) -> + {ok, Pid} = nhttp:start_link(maps:merge(Base, Opts)), + nhttp:stop(Pid) + end, + Valid + ), + ok. + +listener_h2_window_policy_invalid(_Config) -> + Base = #{port => 0, handler => ?MODULE, versions => [http1_1]}, + Invalid = [ + lazy, + {threshold, 0}, + {threshold, -1}, + {threshold, 2147483648}, + {threshold, n}, + {delay, -1}, + {delay, 1.5}, + {uniform, 1, 2} + ], + Valid = [ + eager, + never, + on_response, + {threshold, 1}, + {threshold, 2147483647}, + {delay, 0}, + {delay, 250} + ], + lists:foreach( + fun({Key, Policy}) -> + assert_error_contains( + "invalid_h2_window_policy", nhttp:start_link(Base#{Key => Policy}) + ) + end, + [{Key, Policy} || Key <- policy_keys(), Policy <- Invalid] + ), + lists:foreach( + fun({Key, Policy}) -> + {ok, Pid} = nhttp:start_link(Base#{Key => Policy}), + nhttp:stop(Pid) + end, + [{Key, Policy} || Key <- policy_keys(), Policy <- Valid] + ), + ok. + listener_invalid_versions(_Config) -> assert_error_contains( "invalid_versions", @@ -224,6 +303,9 @@ assert_error_contains(Substr, Result) -> Flat = lists:flatten(io_lib:format("~p", [Result])), ?assertNotEqual(nomatch, string:find(Flat, Substr)). +policy_keys() -> + [h2_connection_window_policy, h2_stream_window_policy]. + acceptor_sys_lifecycle(_Config) -> {ok, Pid} = nhttp:start_link(#{ port => 0, handler => ?MODULE, versions => [http1_1], acceptor_count => 1