Remove stray HTTP/2 machine timeout messages on terminate

[smartinio]

What

• Add regression test for cow_http2_machine:terminate/1 removing stray HTTP/2 machine timeout messages
• Update terminate/1 to synchronously cancel and remove preface_timeout and settings_timeout messages, using the same pattern as Cowboy: https://github.com/ninenines/cowboy/blob/d386b0f62ddb57fee7d489b875239a812202956c/src/cowboy_http.erl#L348-L361

Why

cow_http2_machine starts timers that send {timeout, TRef, _} messages to the owning process. terminate/1 cancels these timers asynchronously and does not remove stray timeout messages from the mailbox. This sometimes leaves stale HTTP/2 machine timeout messages for the caller to receive after the machine has terminated.

This manifests in Gun as an error: Unexpected event in state :domain_lookup of type :info:. Here's a repro using Gun. Target the ebin of this fixed cowlib version vs latest using COWLIB_EBIN:

Details

#!/usr/bin/env escript
%%
%% Minimal public Gun repro for stale HTTP/2 setup timeout messages.
%%
%% The script starts a TLS server that negotiates ALPN "h2" and then does
%% not send HTTP/2 SETTINGS. It opens a Gun connection, waits through the
%% Cowlib HTTP/2 setup timeout, and counts Gun's "Unexpected event in state
%% domain_lookup" logs for stale cow_http2_machine timeout messages.
%%
%% Run from a built Gun checkout:
%%
%%   env ASDF_ERLANG_VERSION=27.0 \
%%     GUN_EBIN=$PWD/ebin \
%%     COWLIB_EBIN=$PWD/deps/cowlib/ebin \
%%     CT_HELPER_EBIN=$PWD/deps/ct_helper/ebin \
%%     EXPECT=stale \
%%     escript ./h2_stale_timeout_repro.escript
%%
%% Swap only COWLIB_EBIN to test a fixed Cowlib build:
%%
%%   env ASDF_ERLANG_VERSION=27.0 \
%%     GUN_EBIN=$PWD/ebin \
%%     COWLIB_EBIN=/path/to/fixed/cowlib/ebin \
%%     CT_HELPER_EBIN=$PWD/deps/ct_helper/ebin \
%%     EXPECT=clean \
%%     escript ./h2_stale_timeout_repro.escript
%%
%% Expected result with old Cowlib: stale_timeout_count > 0.
%% Expected result with fixed Cowlib: stale_timeout_count = 0.

main(_) ->
	GunEbin = getenv_path("GUN_EBIN"),
	CowlibEbin = getenv_path("COWLIB_EBIN"),
	CtHelperEbin = getenv_path("CT_HELPER_EBIN"),
	%% Keep COWLIB_EBIN explicit so the same script can A/B the dependency.
	code:add_patha(CtHelperEbin),
	code:add_patha(GunEbin),
	code:add_patha(CowlibEbin),
	{ok, _} = application:ensure_all_started(gun),
	RuntimeMs = getenv_int("RUNTIME_MS", 25000),
	StallMs = getenv_int("STALL_MS", 6000),
	H2Timeout = getenv_int("H2_TIMEOUT_MS", 5000),
	RetryMs = getenv_int("RETRY_MS", 1000),
	Self = self(),
	{ServerPid, Port} = start_tls_h2_stall_server(StallMs, Self),
	ok = logger:add_primary_filter(?MODULE, {fun logger_filter/2, Self}),
	{ok, ConnPid} = gun:open("localhost", Port, #{
		transport => tls,
		tls_opts => [{verify, verify_none}],
		protocols => [http2],
		retry => 100,
		http2_opts => #{
			preface_timeout => H2Timeout,
			settings_timeout => H2Timeout
		},
		retry_fun => fun(Retries, _) ->
			#{retries => max(0, Retries - 1), timeout => RetryMs}
		end
	}),
	Result = collect(#{
		deadline => erlang:monotonic_time(millisecond) + RuntimeMs,
		counts => #{},
		first => undefined
	}),
	catch gun:close(ConnPid),
	exit(ServerPid, kill),
	logger:remove_primary_filter(?MODULE),
	Count = total_count(maps:get(counts, Result)),
	io:format("gun=~s~n", [code:which(gun)]),
	io:format("cow_http2_machine=~s~n", [code:which(cow_http2_machine)]),
	io:format("runtime_ms=~B stall_ms=~B h2_timeout_ms=~B retry_ms=~B~n",
		[RuntimeMs, StallMs, H2Timeout, RetryMs]),
	io:format("stale_timeout_count=~B counts=~p first=~p~n",
		[Count, maps:get(counts, Result), maps:get(first, Result)]),
	case os:getenv("EXPECT") of
		"stale" when Count > 0 ->
			halt(0);
		"stale" ->
			halt(1);
		"clean" when Count =:= 0 ->
			halt(0);
		"clean" ->
			halt(1);
		false ->
			halt(0)
	end.

getenv_path(Name) ->
	case os:getenv(Name) of
		false ->
			error({missing_env, Name});
		Path ->
			case filelib:is_dir(Path) of
				true -> Path;
				false -> error({bad_path, Name, Path})
			end
	end.

getenv_int(Name, Default) ->
	case os:getenv(Name) of
		false -> Default;
		Value -> list_to_integer(Value)
	end.

start_tls_h2_stall_server(StallMs, Parent) ->
	Pid = spawn(fun() ->
		ok = ct_helper:make_certs_in_ets(),
		CertOpts = ct_helper:get_certs_from_ets(),
		{ok, ListenSocket} = ssl:listen(0, [binary, {active, false},
			{packet, raw}, {reuseaddr, true}, {ip, {127, 0, 0, 1}},
			{fail_if_no_peer_cert, false},
			{alpn_preferred_protocols, [<<"h2">>]}|CertOpts]),
		{ok, {_, Port}} = ssl:sockname(ListenSocket),
		Parent ! {self(), Port},
		tls_accept_loop(ListenSocket, StallMs)
	end),
	receive
		{Pid, Port} -> {Pid, Port}
	after 5000 ->
		error(server_start_timeout)
	end.

tls_accept_loop(ListenSocket, StallMs) ->
	case ssl:transport_accept(ListenSocket, 5000) of
		{ok, Socket0} ->
			spawn(fun() ->
				case ssl:handshake(Socket0, 5000) of
					{ok, Socket} ->
						timer:sleep(StallMs),
						ssl:close(Socket);
					{error, _} ->
						ssl:close(Socket0)
				end
			end),
			tls_accept_loop(ListenSocket, StallMs);
		{error, timeout} ->
			tls_accept_loop(ListenSocket, StallMs);
		{error, closed} ->
			ok
	end.

collect(State=#{deadline := Deadline, counts := Counts, first := First}) ->
	case erlang:monotonic_time(millisecond) >= Deadline of
		true ->
			State;
		false ->
			receive
				{stale_http2_timeout, ConnPid, StateName, TimeoutName, LogState} ->
					Key = {StateName, TimeoutName},
					Counts1 = Counts#{Key => maps:get(Key, Counts, 0) + 1},
					First1 = case First of
						undefined ->
							#{conn_pid => ConnPid, state => StateName,
								timeout => TimeoutName, gun_state => summarize_state(LogState)};
						_ ->
							First
					end,
					collect(State#{counts => Counts1, first => First1})
			after 100 ->
				collect(State)
			end
	end.

logger_filter(#{
		level := error,
		meta := #{pid := ConnPid},
		msg := {report, #{
			args := [StateName, info,
				{timeout, _, {cow_http2_machine, _, TimeoutName}}, State],
			format := "Unexpected event in state ~p of type ~p:~n~w~n~p~n",
			label := {error_logger, error_msg}
		}}
	}, TestPid) when TimeoutName =:= preface_timeout; TimeoutName =:= settings_timeout ->
	TestPid ! {stale_http2_timeout, ConnPid, StateName, TimeoutName, State},
	stop;
logger_filter(LogEvent, _) ->
	LogEvent.

summarize_state({state, _Owner, Status, Host, Port, _OriginScheme,
		_OriginHost, _OriginPort, _Intermediaries, _Opts, Socket, _CookieStore,
		Transport, Active, _Messages, Protocol, _ProtoState, _KeepaliveRef,
		_EventHandler, _EventHandlerState}) ->
	#{status => Status, host => Host, port => Port, socket => Socket,
		transport => Transport, active => Active, protocol => Protocol};
summarize_state(State) ->
	State.

total_count(Counts) ->
	maps:fold(fun(_, Value, Acc) -> Acc + Value end, 0, Counts).

Testing

Tests cover terminate/1 removing stray preface_timeout and settings_timeout messages.

[smartinio]

@essen Hi, could you have a look at this? Thanks in advance

[essen]

Already did but I don't have feedback yet.

[essen]

I will review this after today's release, haven't had time before.

[smartinio]

Thanks!

[essen]

I notice CI didn't run, and I don't have the button to approve the workflow for unknown reasons. If you can please close this PR and reopen another one with the same change and I will approve it.

[smartinio]

@essen Reopened at #161