From 24cfb43e86d19a3f04312ae6076a4a74d68d8921 Mon Sep 17 00:00:00 2001 From: zmstone Date: Fri, 25 Sep 2026 21:20:45 +0200 Subject: [PATCH 1/4] refactor: carry the call time and timeout in ExpireAt ExpireAt was an absolute deadline computed in the caller. It becomes {CalledAt, Timeout}, so the worker can read the request timeout itself. The zombie-detection threshold needs that value to clamp detection time for long timeouts. deadline/1, timeout/1 and is_expired/2 interpret the tuple. The sites that only pass ExpireAt through are unchanged. No behaviour change. --- src/ehttpc.erl | 31 ++++++++++++++++++++----------- 1 file changed, 20 insertions(+), 11 deletions(-) diff --git a/src/ehttpc.erl b/src/ehttpc.erl index 57b7fb4..adaedac 100644 --- a/src/ehttpc.erl +++ b/src/ehttpc.erl @@ -616,10 +616,21 @@ cancel_stream(nofin, Client, StreamRef) -> _ = gun:cancel(Client, StreamRef), ok. -timeout(infinity = _ExpireAt) -> +%% ExpireAt is {CalledAt, Timeout}: the caller's timestamp and its request timeout. +timeout({_CalledAt, infinity}) -> infinity; timeout(ExpireAt) -> - max(ExpireAt - now_(), 0). + max(deadline(ExpireAt) - now_(), 0). + +deadline({_CalledAt, infinity}) -> + infinity; +deadline({CalledAt, Timeout}) -> + CalledAt + Timeout. + +is_expired({_CalledAt, infinity}, _Now) -> + false; +is_expired(ExpireAt, Now) -> + Now > deadline(ExpireAt). now_() -> erlang:system_time(millisecond). @@ -636,10 +647,10 @@ put_sent_req( max_sent_expire := T } = Requests ) -> - ?SENT_REQ(_, Expire, _) = Req, + ?SENT_REQ(_, ExpireAt, _) = Req, Requests#{ sent := maps:put(StreamRef, Req, Sent), - max_sent_expire := max_expire(T, Expire) + max_sent_expire := max_expire(T, deadline(ExpireAt)) }. %% if a request has infinity timeout, ignore it @@ -670,12 +681,12 @@ take_sent_req(StreamRef, #{sent := Sent, max_sent_expire := T} = Requests) -> end end. -is_sent_req_expired(?SENT_REQ(_From, infinity = _ExpireAt, _), _Now) -> +is_sent_req_expired(?SENT_REQ(_From, {_CalledAt, infinity}, _), _Now) -> false; is_sent_req_expired(?SENT_REQ({Pid, _Ref}, ExpireAt, _), Now) when is_pid(Pid) -> %% for gen_server:call, it is aborted after timeout, there is no need to send %% reply to the caller - Now > ExpireAt orelse (not erlang:is_process_alive(Pid)); + is_expired(ExpireAt, Now) orelse (not erlang:is_process_alive(Pid)); is_sent_req_expired(?SENT_REQ(_, _, _), _) -> %% for async requests, there is no way to tell if the caller %% the provided result-callback should be evaluated or not, @@ -712,7 +723,7 @@ drop_expired(#{pending_count := 0} = Requests, _Now) -> drop_expired(#{pending := Pending, pending_count := PC} = Requests, Now) -> {PeekFun, OutFun} = peek_oldest_fn(Requests), {value, ?PEND_REQ(ReplyTo, ?REQ(_, _, ExpireAt))} = PeekFun(Pending), - case is_integer(ExpireAt) andalso Now > ExpireAt of + case is_expired(ExpireAt, Now) of true -> {_, NewPendings} = OutFun(Pending), NewRequests = Requests#{pending => NewPendings, pending_count => PC - 1}, @@ -1129,10 +1140,8 @@ enqueue_latest_fn(#{prioritise_latest := true}) -> enqueue_latest_fn(_) -> fun queue:in/2. -fresh_expire_at(infinity = _Timeout) -> - infinity; -fresh_expire_at(Timeout) when is_integer(Timeout) -> - now_() + Timeout. +fresh_expire_at(Timeout) when Timeout =:= infinity; is_integer(Timeout) -> + {now_(), Timeout}. parse_proxy_opts(Opts) -> %% Target host and port From 2783f2a060b5fc1443cdc768278c9f79f3ce8f55 Mon Sep 17 00:00:00 2001 From: zmstone Date: Fri, 25 Sep 2026 21:26:00 +0200 Subject: [PATCH 2/4] fix(zombie): cap the zombie-detection threshold Zombie detection fired at request timeout + max_inactive after the newest send, so it scaled with the caller's timeout. With a 600 s request timeout a dead connection was reconnected only after 610 s, and lowering the timeout would break the long requests it was chosen for. The threshold is now max(MaxInactive, min(MaxInactiveCap, MaxTimeout + MaxInactive)) MaxTimeout is the largest request timeout among the in-flight requests. MaxInactiveCap defaults to 60 s and is a new pool option, max_inactive_cap. For timeouts up to 50 s with the default max_inactive the result is unchanged. A raised max_inactive stays authoritative through the outer max. The reference timestamp is now the latest send time (max_sent_at), not the latest deadline. The timeout is added by the threshold, so a deadline reference would count it twice. The send time is taken with now_() in put_sent_req/3, when the request goes to gun, rather than the caller's call time. The two differ by the time spent in the pending queue, and the send time is what "silent since the last send" means. It also needs no special case for infinity. max_expire/2 skipped infinity deadlines, so a pool whose requests all used an infinity timeout never recorded one and zombie detection never fired. infinity now wins the timeout maximum, and the threshold for it is the cap. max_sent_expire is renamed to max_sent_at, and max_sent_timeout is tracked next to it. Both reset to 0 when no request is in flight. The force_reconnecting_zombie_http_connection log reports the computed threshold, the latest send time and the largest request timeout. Incompatible case: with a request timeout above the cap minus max_inactive, a response that takes longer than the cap now gets its connection killed at the cap. Received data does not count as activity. Raise max_inactive to the slowest expected response. Refs #70 --- changelog.md | 12 ++++++ src/ehttpc.app.src | 2 +- src/ehttpc.erl | 99 +++++++++++++++++++++++++++++++------------ test/ehttpc_tests.erl | 35 ++++++++++++++- 4 files changed, 118 insertions(+), 30 deletions(-) diff --git a/changelog.md b/changelog.md index 623c64d..1dec2be 100644 --- a/changelog.md +++ b/changelog.md @@ -1,5 +1,17 @@ # ehttpc changes +## 0.7.6 + +- Zombie-connection detection no longer waits for the full request timeout. A connection + with requests in flight is reconnected when the latest send is older than + `request timeout + max_inactive`, capped at 60 seconds but never below `max_inactive`. Previously a 600-second request + timeout delayed detection to 610 seconds, and requests with an `infinity` timeout disabled + detection entirely. +- Behaviour change: with a request timeout above about 50 seconds and the default + `max_inactive`, 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. + ## 0.7.5 - Added the possibility of choosing HTTP2 as the protocol. diff --git a/src/ehttpc.app.src b/src/ehttpc.app.src index 7fda509..6bbce1a 100644 --- a/src/ehttpc.app.src +++ b/src/ehttpc.app.src @@ -1,6 +1,6 @@ {application, ehttpc, [ {description, "HTTP Client for Erlang/OTP"}, - {vsn, "0.7.5"}, + {vsn, "0.7.6"}, {registered, []}, {applications, [ kernel, diff --git a/src/ehttpc.erl b/src/ehttpc.erl index adaedac..ae2d6a5 100644 --- a/src/ehttpc.erl +++ b/src/ehttpc.erl @@ -92,6 +92,7 @@ -define(undef, undefined). -define(IS_POOL(Pool), (not is_tuple(Pool) andalso not is_pid(Pool))). -define(DEFAULT_MAX_INACTIVE, 10_000). +-define(INACTIVE_CAP, 60_000). -define(IS_HEADERS_REQ(REQ), (tuple_size(REQ) =:= 2 andalso is_list(element(2, REQ))) @@ -119,6 +120,7 @@ %% In this case, host and port point to proxy server. origin :: undefined | map(), max_inactive :: pos_integer(), + max_inactive_cap :: pos_integer(), inactive_check_tref :: reference() | ?undef }). @@ -273,6 +275,7 @@ init([Pool, Id, Opts0]) -> PrioLatest = proplists:get_bool(prioritise_latest, Opts0), #{opts := Opts, origin := Origin} = parse_proxy_opts(Opts0), MaxInactive = proplists:get_value(max_inactive, Opts, ?DEFAULT_MAX_INACTIVE), + MaxInactiveCap = proplists:get_value(max_inactive_cap, Opts, ?INACTIVE_CAP), State = #state{ pool = Pool, id = Id, @@ -286,11 +289,13 @@ init([Pool, Id, Opts0]) -> pending => queue:new(), pending_count => 0, sent => #{}, - max_sent_expire => 0, + max_sent_at => 0, + max_sent_timeout => 0, prioritise_latest => PrioLatest }, origin = Origin, - max_inactive = MaxInactive + max_inactive = MaxInactive, + max_inactive_cap = MaxInactiveCap }, true = gproc_pool:connect_worker(ehttpc:name(Pool), {Pool, Id}), @@ -639,25 +644,30 @@ now_() -> %% sent requests %% ================================================================================= +%% max_sent_at is the latest time a request was sent to gun. +%% max_sent_timeout is the largest request timeout among the sent requests. +%% Both reset to 0 when no request is in flight. put_sent_req( StreamRef, Req, #{ sent := Sent, - max_sent_expire := T + max_sent_at := SentAt, + max_sent_timeout := MaxTimeout } = Requests ) -> - ?SENT_REQ(_, ExpireAt, _) = Req, + ?SENT_REQ(_, {_CalledAt, Timeout}, _) = Req, Requests#{ sent := maps:put(StreamRef, Req, Sent), - max_sent_expire := max_expire(T, deadline(ExpireAt)) + max_sent_at := max(SentAt, now_()), + max_sent_timeout := max_timeout(MaxTimeout, Timeout) }. -%% if a request has infinity timeout, ignore it -max_expire(T, infinity) -> T; -max_expire(T1, T2) when is_integer(T2) -> max(T1, T2). +max_timeout(_, infinity) -> infinity; +max_timeout(infinity, _) -> infinity; +max_timeout(T1, T2) -> max(T1, T2). -take_sent_req(StreamRef, #{sent := Sent, max_sent_expire := T} = Requests) -> +take_sent_req(StreamRef, #{sent := Sent} = Requests) -> case maps:take(StreamRef, Sent) of error -> error; @@ -666,18 +676,18 @@ take_sent_req(StreamRef, #{sent := Sent, max_sent_expire := T} = Requests) -> %% so there is no need to scan the map to find a new max %% or even if calls may use different timeout %% the impact of a wrong max is minimal: delayed detection of zombie connection - NewT = + NewRequests = case map_size(NewSent) of 0 -> - 0; + reset_sent(Requests); _ -> - T + Requests#{sent := NewSent} end, case is_sent_req_expired(Req, now_()) of true -> - {expired, Requests#{sent := NewSent, max_sent_expire := NewT}}; + {expired, NewRequests}; false -> - {Req, Requests#{sent := NewSent, max_sent_expire := NewT}} + {Req, NewRequests} end end. @@ -709,7 +719,10 @@ reply_error_for_sent_reqs(#{sent := Sent} = R, Reason) -> end, maps:to_list(Sent) ), - R#{sent => #{}, max_sent_expire => 0}. + reset_sent(R). + +reset_sent(Requests) -> + Requests#{sent => #{}, max_sent_at => 0, max_sent_timeout => 0}. %% Continue droping expired requests, to avoid the state RAM usage %% explosion if http client can not keep up. @@ -764,16 +777,19 @@ maybe_shoot( requests = #{ sent := Sent, - max_sent_expire := MaxExpire + max_sent_at := MaxSentAt, + max_sent_timeout := MaxTimeout } = Requests0, client = Client, max_inactive = MaxInactive, + max_inactive_cap = MaxInactiveCap, enable_pipelining = PipelineLimit } = State0 ) -> State = State0#state{requests = drop_expired(Requests0)}, SentCount = map_size(Sent), - case check_gun(Client, PipelineLimit, SentCount, MaxExpire, MaxInactive) of + Threshold = inactive_threshold(MaxTimeout, MaxInactive, MaxInactiveCap), + case check_gun(Client, PipelineLimit, SentCount, MaxSentAt, Threshold) of continue -> do_shoot(State); pause -> @@ -783,16 +799,17 @@ maybe_shoot( State; reconnect -> %% assert - true = (MaxExpire > 0), + true = (MaxSentAt > 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, [ + last_request_sent_at => calendar:system_time_to_rfc3339(MaxSentAt, [ {unit, millisecond} ]), - inactive_duration_threshold => MaxInactive, + inactive_duration_threshold => Threshold, + max_request_timeout => MaxTimeout, inflight_requests => SentCount, connection_pid => Client, pool => State#state.pool, @@ -805,10 +822,10 @@ maybe_shoot( State end. -check_gun(ClientPid, PipelineLimit, SentCount, MaxExpireTs, MaxInactiveDuration) -> +check_gun(ClientPid, PipelineLimit, SentCount, MaxSentAt, Threshold) -> maybe ok ?= check_gun_pid(ClientPid), - ok ?= check_gun_jammed(SentCount, MaxExpireTs, MaxInactiveDuration), + ok ?= check_gun_jammed(SentCount, MaxSentAt, Threshold), check_gun_limit(PipelineLimit, SentCount) end. @@ -827,19 +844,28 @@ check_gun_pid(Pid) -> pause end. -%% if there are sent requests, and the last reply is older than max_inactive, +%% if there are sent requests, and the latest send is older than the threshold, %% the connection is considered in zomebie state hence require a reconnect. -check_gun_jammed(_SentCount, 0, _MaxInactiveDuration) -> - %% there was no expire time recorded before +check_gun_jammed(_SentCount, 0, _Threshold) -> + %% no request in flight ok; -check_gun_jammed(_SentCount, MaxExpireTs, MaxInactiveDuration) -> - case (now_() - MaxExpireTs) > MaxInactiveDuration of +check_gun_jammed(_SentCount, MaxSentAt, Threshold) -> + case (now_() - MaxSentAt) > Threshold of true -> reconnect; false -> ok end. +%% How long after the latest send a connection with requests in flight +%% may stay silent before it is considered a zombie. +%% It is MaxTimeout + MaxInactive, capped at MaxInactiveCap, +%% but never less than MaxInactive. +inactive_threshold(infinity, MaxInactive, MaxInactiveCap) -> + max(MaxInactive, MaxInactiveCap); +inactive_threshold(MaxTimeout, MaxInactive, MaxInactiveCap) -> + max(MaxInactive, min(MaxInactiveCap, MaxTimeout + MaxInactive)). + %% allow 100 async requests maximum when enable_pipelining is 'true' %% allow only 1 async request when enable_pipelining is 'false' %% otherwise stop shooting at the number limited by enable_pipelining @@ -1223,6 +1249,25 @@ log(Level, Data, #state{host = Host, port = Port}) -> -ifdef(TEST). +inactive_threshold_test_() -> + Cap = ?INACTIVE_CAP, + Cases = [ + %% {MaxTimeout, MaxInactive, Expected} + {5_000, 10_000, 15_000}, + {45_000, 10_000, 55_000}, + {50_000, 10_000, 60_000}, + {600_000, 10_000, 60_000}, + {infinity, 10_000, 60_000}, + {600_000, 180_000, 180_000}, + {infinity, 180_000, 180_000}, + {0, 10_000, 10_000}, + {0, 60_000, 60_000} + ], + [ + ?_assertEqual(Expected, inactive_threshold(MaxTimeout, MaxInactive, Cap)) + || {MaxTimeout, MaxInactive, Expected} <- Cases + ]. + prioritise_latest_test() -> Opts = #{prioritise_latest => true}, Seq = [1, 2, 3, 4], diff --git a/test/ehttpc_tests.erl b/test/ehttpc_tests.erl index c309d4b..4cf7755 100644 --- a/test/ehttpc_tests.erl +++ b/test/ehttpc_tests.erl @@ -651,7 +651,7 @@ zombie_detect_inflight_not_full_test() -> %% let the EXIT signal get to ehttpc process {ok, _} = ?block_until(#{?snk_kind := handle_client_down}, 1000, infinity), #{requests := Requests} = ehttpc:get_state(Pid, normal), - #{sent := Sent, pending_count := PendingCount, max_sent_expire := MaxTs} = Requests, + #{sent := Sent, pending_count := PendingCount, max_sent_at := MaxTs} = Requests, ?assertEqual(0, maps:size(Sent)), ?assertEqual(0, PendingCount), ?assertEqual(0, MaxTs), @@ -699,7 +699,7 @@ zombie_detect_inflight_full_test() -> error(timeout) end, #{requests := Requests} = ehttpc:get_state(Pid, normal), - #{sent := Sent, pending_count := PendingCount, max_sent_expire := MaxTs} = Requests, + #{sent := Sent, pending_count := PendingCount, max_sent_at := MaxTs} = Requests, ?assertEqual(0, maps:size(Sent)), ?assertEqual(0, PendingCount), ?assertEqual(0, MaxTs), @@ -707,6 +707,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 max_inactive_cap = 2s, 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}, {max_inactive_cap, 2_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", From 8279ae8d753a5ea3daf7def90a700afa7fa9246b Mon Sep 17 00:00:00 2001 From: zmstone Date: Fri, 25 Sep 2026 22:43:22 +0200 Subject: [PATCH 3/4] refactor: update existing keys with := in reset_sent/1 --- src/ehttpc.erl | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/ehttpc.erl b/src/ehttpc.erl index ae2d6a5..6ea35c9 100644 --- a/src/ehttpc.erl +++ b/src/ehttpc.erl @@ -722,7 +722,7 @@ reply_error_for_sent_reqs(#{sent := Sent} = R, Reason) -> reset_sent(R). reset_sent(Requests) -> - Requests#{sent => #{}, max_sent_at => 0, max_sent_timeout => 0}. + Requests#{sent := #{}, max_sent_at := 0, max_sent_timeout := 0}. %% Continue droping expired requests, to avoid the state RAM usage %% explosion if http client can not keep up. From 1d29c98af3302d8964aacf040a886118e39e4fa2 Mon Sep 17 00:00:00 2001 From: zmstone Date: Fri, 25 Sep 2026 22:43:22 +0200 Subject: [PATCH 4/4] docs: name max_inactive_cap in the 0.7.6 changelog The cap is a pool option, so the changelog names it and lists it as a way to raise the bound, next to max_inactive. --- changelog.md | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/changelog.md b/changelog.md index 1dec2be..3acbc13 100644 --- a/changelog.md +++ b/changelog.md @@ -4,13 +4,14 @@ - Zombie-connection detection no longer waits for the full request timeout. A connection with requests in flight is reconnected when the latest send is older than - `request timeout + max_inactive`, capped at 60 seconds but never below `max_inactive`. Previously a 600-second request - timeout delayed detection to 610 seconds, and requests with an `infinity` timeout disabled - detection entirely. -- Behaviour change: with a request timeout above about 50 seconds and the default - `max_inactive`, 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. + `request timeout + max_inactive`, capped at `max_inactive_cap` but never below + `max_inactive`. `max_inactive_cap` is a new pool option and defaults to 60 seconds. + Previously a 600-second request timeout delayed detection to 610 seconds, and requests + with an `infinity` timeout disabled detection entirely. +- Behaviour change: with a request timeout above about 50 seconds and the default pool + options, a connection whose response takes longer than 60 seconds is now killed at + 60 seconds. Set `max_inactive_cap` or `max_inactive` to the slowest expected response time + to raise the bound. ## 0.7.5