Add keepalive_tolerance http2 option

The number of unacknowledged pings that can be tolerated
before the connection is forcefully closed.

When a keepalive ping is sent to the peer, a counter is
incremented and if this counter exceeds the tolerance limit,
the connection is forcefully closed. The counter is
decremented whenever a ping ack is received from the peer.

By default, the mechanism for closing the connection based
on ping and ping ack is disabled.

Loïc Hoguin: I have edited a lot of the code and renamed
a few things as well as simplified the docs and increased
test timeouts to avoid race conditions.

4 files changed, 86 insertions, 5 deletions

diff --git a/doc/src/manual/gun.asciidoc b/doc/src/manual/gun.asciidoc
index 40559fb..d511fdc 100644
--- a/doc/src/manual/gun.asciidoc
+++ b/doc/src/manual/gun.asciidoc
@@ -209,6 +209,7 @@ http2_opts() :: #{
     cookie_ignore_informational => boolean(),
     flow                        => pos_integer(),
     keepalive                   => timeout(),
+    keepalive_tolerance         => non_neg_integer(),
 
     %% HTTP/2 state machine configuration.
     connection_window_margin_size  => 0..16#7fffffff,
@@ -257,6 +258,13 @@ keepalive (infinity)::
 
 Time between pings in milliseconds.
 
+keepalive_tolerance - see below::
+
+The number of unacknowledged pings in flight that are
+tolerated before the connection is closed. By default
+this mechanism is disabled even if `keepalive` is
+enabled.
+
 === opts()
 
 [source,erlang]
diff --git a/src/gun.erl b/src/gun.erl
index b27ea6e..976a137 100644
--- a/src/gun.erl
+++ b/src/gun.erl
@@ -223,6 +223,7 @@
 	cookie_ignore_informational => boolean(),
 	flow => pos_integer(),
 	keepalive => timeout(),
+	keepalive_tolerance => non_neg_integer(),
 	notify_settings_changed => boolean(),
 
 	%% Options copied from cow_http2_machine.
diff --git a/src/gun_http2.erl b/src/gun_http2.erl
index 292504b..c37b48a 100644
--- a/src/gun_http2.erl
+++ b/src/gun_http2.erl
@@ -113,7 +113,12 @@
 	%% the idea, that's why the main map has the ID as key. Then we also
 	%% have a Ref->ID index for faster lookup when we only have the Ref.
 	streams = #{} :: #{cow_http2:streamid() => #stream{}},
-	stream_refs = #{} :: #{reference() => cow_http2:streamid()}
+	stream_refs = #{} :: #{reference() => cow_http2:streamid()},
+
+	%% Number of pings that have been sent but not yet acknowledged.
+	%% Used to determine whether the connection should be closed when
+	%% the keepalive_tolerance option is set.
+	pings_unack = 0 :: non_neg_integer()
 }).
 
 check_options(Opts) ->
@@ -139,6 +144,8 @@ do_check_options([{keepalive, infinity}|Opts]) ->
 	do_check_options(Opts);
 do_check_options([{keepalive, K}|Opts]) when is_integer(K), K > 0 ->
 	do_check_options(Opts);
+do_check_options([{keepalive_tolerance, K}|Opts]) when is_integer(K), K >= 0 ->
+	do_check_options(Opts);
 do_check_options([{notify_settings_changed, B}|Opts]) when is_boolean(B) ->
 	do_check_options(Opts);
 do_check_options([Opt={Name, _}|Opts]) ->
@@ -341,7 +348,8 @@ frame(State=#http2_state{http2_machine=HTTP2Machine0}, Frame, CookieStore, EvHan
 	end.
 
 maybe_ack_or_notify(State=#http2_state{reply_to=ReplyTo, socket=Socket,
-		transport=Transport, opts=Opts, http2_machine=HTTP2Machine}, Frame) ->
+		transport=Transport, opts=Opts, http2_machine=HTTP2Machine,
+		pings_unack=PingsUnack}, Frame) ->
 	case Frame of
 		{settings, _} ->
 			%% We notify remote settings changes only if the user requested it.
@@ -361,6 +369,8 @@ maybe_ack_or_notify(State=#http2_state{reply_to=ReplyTo, socket=Socket,
 				ok -> {state, State};
 				Error={error, _} -> Error
 			end;
+		{ping_ack, _Opaque} ->
+			{state, State#http2_state{pings_unack=PingsUnack - 1}};
 		_ ->
 			{state, State}
 	end.
@@ -908,10 +918,18 @@ close_stream(State, #stream{ref=StreamRef, reply_to=ReplyTo}, Reason) ->
 	ReplyTo ! {gun_error, self(), stream_ref(State, StreamRef), Reason},
 	ok.
 
-keepalive(#http2_state{socket=Socket, transport=Transport}, _, EvHandlerState) ->
+keepalive(State=#http2_state{pings_unack=PingsUnack, opts=Opts}, _, EvHandlerState)
+		when PingsUnack >= map_get(keepalive_tolerance, Opts) ->
+	{connection_error(State, {connection_error, no_error,
+		'The number of unacknowledged pings exceed the configured tolerance value.'}),
+		EvHandlerState};
+keepalive(State=#http2_state{socket=Socket, transport=Transport, pings_unack=PingsUnack},
+		_, EvHandlerState) ->
 	case Transport:send(Socket, cow_http2:ping(0)) of
-		ok -> {[], EvHandlerState};
-		Error={error, _} -> {Error, EvHandlerState}
+		ok ->
+			{{state, State#http2_state{pings_unack=PingsUnack + 1}}, EvHandlerState};
+		Error={error, _} ->
+			{Error, EvHandlerState}
 	end.
 
 headers(State=#http2_state{socket=Socket, transport=Transport, opts=Opts,
diff --git a/test/rfc7540_SUITE.erl b/test/rfc7540_SUITE.erl
index 159d036..a494128 100644
--- a/test/rfc7540_SUITE.erl
+++ b/test/rfc7540_SUITE.erl
@@ -435,6 +435,60 @@ settings_ack_timeout(_) ->
 	timer:sleep(6000),
 	gun:close(ConnPid).
 
+keepalive_tolerance_ping_ack_timeout(_) ->
+	doc("The PING frame may be used to easily test a connection. (RFC7540 8.1.4)"),
+	{ok, OriginPid, OriginPort} = init_origin(tcp, http2, do_ping_ack_loop_fun()),
+	{ok, Pid} = gun:open("localhost", OriginPort, #{
+		protocols => [http2],
+		http2_opts => #{keepalive => 100, keepalive_tolerance => 2}
+	}),
+	{ok, http2} = gun:await_up(Pid),
+	handshake_completed = receive_from(OriginPid),
+	%% When Gun sends the first ping, the server acks immediately.
+	receive ping_received -> OriginPid ! send_ping_ack end,
+	timer:sleep(250), %% Gun sends 2 pings while we sleep. 2 pings not acked.
+	%% Server acks one ping. One ping still not acked.
+	receive ping_received -> OriginPid ! send_ping_ack end,
+	timer:sleep(100), %% Gun sends 1 ping while we sleep. 2 pings not acked.
+	%% Server acks one ping. One ping still not acked.
+	receive ping_received -> OriginPid ! send_ping_ack end,
+	timer:sleep(100), %% Gun sends 1 ping while we sleep. 2 pings not acked.
+	%% Check that we haven't received a gun_down yet.
+	receive
+		GunDown when element(1, GunDown) =:= gun_down ->
+			error(unexpected)
+	after 0 ->
+		ok
+	end,
+	%% Within the next 10ms, Gun wants to send another ping, which would
+	%% result in 3 outstanding pings. Instead, Gun goes down.
+	receive
+		{gun_down, Pid, http2, {error, {connection_error, no_error, _}}, []} ->
+			gun:close(Pid)
+	after 100 ->
+		error(timeout)
+	end.
+
+do_ping_ack_loop_fun() ->
+	%% Receive ping, sync with parent, send ping ack, loop.
+	fun Loop(Parent, Socket, Transport) ->
+		{ok, Data} = Transport:recv(Socket, 9, infinity),
+		<<Len:24, 6:8, %% PING
+			0:8, %% Flags
+			0:1, 0:31>> = Data,
+		{ok, Payload} = Transport:recv(Socket, Len, 1000),
+		8 = Len = byte_size(Payload),
+		Parent ! ping_received,
+		receive
+			send_ping_ack ->
+				Ack = <<8:24, 6:8, %% PING
+					1:8, %% Ack flag
+					0:1, 0:31, Payload/binary>>,
+				ok = Transport:send(Socket, Ack)
+		end,
+		Loop(Parent, Socket, Transport)
+	end.
+
 connect_http_via_h2c(_) ->
 	doc("CONNECT can be used to establish a TCP connection "
 		"to an HTTP/1.1 server via a TCP HTTP/2 proxy. (RFC7540 8.3)"),