fix(zombie): cap zombie detection by the latest send time (0.7.3, hot-patchable)

.github/workflows/erlang.yml

@@ -2,9 +2,9 @@ name: Erlang CI
 
 on:
   push:
-    branches: [ main ]
+    branches: [ main, 'patch-*' ]
   pull_request:
-    branches: [ main ]
+    branches: [ main, 'patch-*' ]
 
 jobs:
 

changelog.md

@@ -1,5 +1,17 @@
 # ehttpc changes
 
+## 0.7.3.1
+
+- Zombie-connection detection no longer waits for the full request timeout. A connection
+  with requests in flight is also reconnected when the latest send is older than
+  `max(max_inactive, 60 seconds)`. Previously a 600-second request timeout delayed detection
+  to 610 seconds, and requests with an `infinity` timeout disabled detection entirely.
+- Behaviour change: a connection whose response takes longer than 60 seconds is now killed
+  at 60 seconds. Set the `max_inactive` pool option to the slowest expected response time to
+  raise the bound.
+- The change keeps the worker state and message formats, so it can be applied by loading
+  the new `ehttpc` module into a running node.
+
 ## 0.7.3
 
 - Previously, we had a fixed restart intensity for the worker supervisor, meaning that if a

src/ehttpc.app.src

@@ -1,6 +1,6 @@
 {application, ehttpc, [
     {description, "HTTP Client for Erlang/OTP"},
-    {vsn, "0.7.3"},
+    {vsn, "0.7.3.1"},
     {registered, []},
     {applications, [
         kernel,

src/ehttpc.erl

@@ -82,6 +82,13 @@
 -define(undef, undefined).
 -define(IS_POOL(Pool), (not is_tuple(Pool) andalso not is_pid(Pool))).
 -define(DEFAULT_MAX_INACTIVE, 10_000).
+-ifdef(TEST).
+-define(INACTIVE_CAP, 2_000).
+-else.
+-define(INACTIVE_CAP, 60_000).
+-endif.
+%% Process dictionary key for the time of the latest request sent to gun.
+-define(LAST_SENT_TS, last_sent_ts).
 
 -define(IS_HEADERS_REQ(REQ),
     (tuple_size(REQ) =:= 2 andalso is_list(element(2, REQ)))
@@ -610,6 +617,7 @@ put_sent_req(
     } = Requests
 ) ->
     ?SENT_REQ(_, Expire, _) = Req,
+    _ = put(?LAST_SENT_TS, now_()),
     Requests#{
         sent := maps:put(StreamRef, Req, Sent),
         max_sent_expire := max_expire(T, Expire)
@@ -741,17 +749,15 @@ maybe_shoot(
             ?tp(cool_down, #{enable_pipelining => State#state.enable_pipelining}),
             State;
         reconnect ->
-            %% assert
-            true = (MaxExpire > 0),
             %% the connection has been inactive for too long
             log(
                 error,
                 #{
                     msg => "force_reconnecting_zombie_http_connection",
-                    last_request_expire => calendar:system_time_to_rfc3339(MaxExpire, [
-                        {unit, millisecond}
-                    ]),
+                    last_request_expire => format_ts(MaxExpire),
+                    last_request_sent_at => format_ts(get(?LAST_SENT_TS)),
                     inactive_duration_threshold => MaxInactive,
+                    inactive_duration_cap => max(MaxInactive, ?INACTIVE_CAP),
                     inflight_requests => SentCount,
                     connection_pid => Client,
                     pool => State#state.pool,
@@ -786,12 +792,38 @@ check_gun_pid(Pid) ->
             pause
     end.
 
-%% if there are sent requests, and the last reply is older than max_inactive,
-%% the connection is considered in zomebie state hence require a reconnect.
-check_gun_jammed(_SentCount, 0, _MaxInactiveDuration) ->
+format_ts(Ts) when is_integer(Ts), Ts > 0 ->
+    calendar:system_time_to_rfc3339(Ts, [{unit, millisecond}]);
+format_ts(_) ->
+    undefined.
+
+%% if there are sent requests, and the latest send is older than
+%% max(max_inactive, ?INACTIVE_CAP), or the latest request expiry is older than
+%% max_inactive, the connection is considered in zombie state hence require a reconnect.
+%% When no send time is recorded in the process dictionary, only the expiry is checked.
+check_gun_jammed(SentCount, MaxExpireTs, MaxInactiveDuration) ->
+    case get(?LAST_SENT_TS) of
+        LastSentTs when is_integer(LastSentTs) ->
+            check_gun_jammed(SentCount, MaxExpireTs, LastSentTs, MaxInactiveDuration);
+        _ ->
+            check_gun_jammed_expire(MaxExpireTs, MaxInactiveDuration)
+    end.
+
+check_gun_jammed(0, _MaxExpireTs, _LastSentTs, _MaxInactiveDuration) ->
+    %% no request in flight
+    ok;
+check_gun_jammed(_SentCount, MaxExpireTs, LastSentTs, MaxInactiveDuration) ->
+    case (now_() - LastSentTs) > max(MaxInactiveDuration, ?INACTIVE_CAP) of
+        true ->
+            reconnect;
+        false ->
+            check_gun_jammed_expire(MaxExpireTs, MaxInactiveDuration)
+    end.
+
+check_gun_jammed_expire(0, _MaxInactiveDuration) ->
     %% there was no expire time recorded before
     ok;
-check_gun_jammed(_SentCount, MaxExpireTs, MaxInactiveDuration) ->
+check_gun_jammed_expire(MaxExpireTs, MaxInactiveDuration) ->
     case (now_() - MaxExpireTs) > MaxInactiveDuration of
         true ->
             reconnect;

test/ehttpc_tests.erl

@@ -531,6 +531,37 @@ zombie_detect_inflight_full_test() ->
         end
     ).
 
+zombie_detect_long_timeout_test() ->
+    zombie_detect_capped(?PORT, ?FUNCTION_NAME, 30_000).
+
+zombie_detect_infinity_timeout_test() ->
+    zombie_detect_capped(?PORT, ?FUNCTION_NAME, infinity).
+
+%% With max_inactive = 1s and the inactive cap at 2s in the TEST build,
+%% a request timeout longer than the cap must not delay zombie detection past the cap.
+zombie_detect_capped(Port, Name, Timeout) ->
+    ServerOpts = #{
+        port => Port,
+        name => Name,
+        %% no response during this test
+        delay => 30_000,
+        oneoff => false
+    },
+    PoolOpts0 = pool_opts("127.0.0.1", Port, 5, _PrioritiseLatest = false),
+    PoolOpts = [{max_inactive, 1_000} | PoolOpts0],
+    ?WITH(
+        ServerOpts,
+        PoolOpts,
+        begin
+            T0 = erlang:monotonic_time(millisecond),
+            spawn_link(fun() -> ehttpc:request(?POOL, put, {<<"/">>, [], "foo"}, Timeout, 0) end),
+            {ok, _} = ?block_until(#{?snk_kind := reconnect}, 3_500, infinity),
+            Elapsed = erlang:monotonic_time(millisecond) - T0,
+            ?assert(Elapsed >= 2_000),
+            ok
+        end
+    ).
+
 head_request_test() ->
     Port = ?PORT,
     Host = "127.0.0.1",