Fix false pingresp_timeout under publish load

[bettio]

Outbound activity reset ping_timer but not pingresp_timer, so a PUBLISH during a PINGREQ -> PINGRESP window pushed the next PINGREQ past the pingresp deadline and pingresp_timer fired even when the broker was fine.

Replace the dual-timer model with an activity-tracking state machine: any complete inbound packet clears pingresp_timer (PINGRESP no longer has privileged status), and the watchdog is not renewed by subsequent PINGREQs so a dropped PINGRESP surfaces as a real error.

Also exit normal from stop_with_error/2 so linked non-trap_exit owners survive transport errors.

[petermm]

AMP:

PR Review — Fix false pingresp_timeout under publish load

Commit: 1bb30b9 (Davide Bettio, 2026-05-15)
Files touched:

• amqtt_client/src/amqtt_client.erl
• amqtt_client/test/amqtt_client_tests.erl
• amqtt_client/README.md

TL;DR

The new keep-alive / pingresp watchdog logic is sound: clearing the
watchdog on any complete inbound packet (observe_inbound/1) does fix the
class of false timeouts caused by outbound PUBLISH/PUBACK pushing the next
PINGREQ past the pingresp deadline. Stale-timer handling via
erlang:start_timer/3 + ref matching is correct. Three things deserve
attention:

1. stop_with_error/2 now exits normal — supervisor restart semantics changed.
2. Manual ping/1 is not gated on an outstanding pingresp watchdog.
3. The keep_alive_seconds = 0 behavior of manual ping/1 is undocumented.

The new tests are solid; only the KA=1 cadence tests are the ones likely
to flake on slow CI.

---

Findings

1. Major — stop_with_error/2 exit reason change has supervisor implications

{stop, normal, cancel_all_timers(...)}.   %% was {stop, {transport_error, Reason}, ...}

Exiting normal correctly protects a non-trap_exit owner from being
killed by the link signal (covered by the new
owner_without_trap_exit_survives_transport_error_test_). However it
silently changes OTP supervision semantics:

• A transient child will no longer be restarted on transport,
  protocol, buffer-overflow, or pingresp_timeout failures — those are
  now indistinguishable from a clean shutdown.
• A permanent child still restarts.
• The error event is still delivered to the owner, so applications
  using the linked-owner API still observe failures.

This is a real behavior change worth calling out in the README. Suggested
documentation patch:

--- a/amqtt_client/src/amqtt_client.erl
+++ b/amqtt_client/src/amqtt_client.erl
@@
-%%   <li>`{mqtt, Pid, error, #{reason := term()}}' on transport error,
-%%       protocol error, or other fatal condition. The gen_server stops
-%%       immediately afterwards.</li>
+%%   <li>`{mqtt, Pid, error, #{reason := term()}}' on transport error,
+%%       protocol error, or other fatal condition. The gen_server stops
+%%       immediately afterwards with exit reason `normal' so that linked
+%%       non-trap_exit owners survive. Supervised clients that should
+%%       restart on such failures must use a `permanent' restart policy.</li>

2. Minor — Manual ping/1 can issue overlapping PINGREQs

handle_keepalive_check/1 correctly skips sending a new PINGREQ while
pingresp_timer is armed, but handle_cast(ping, State) does not.
A caller can manually emit a PINGREQ while one is already in flight;
arm_pingresp_timer/1 then refuses to renew the watchdog, so the
extra PINGREQ becomes redundant traffic with ambiguous semantics and
weakens the otherwise clean "at most one ping outstanding"
state machine.

--- a/amqtt_client/src/amqtt_client.erl
+++ b/amqtt_client/src/amqtt_client.erl
@@ handle_cast/2
 handle_cast({publish, Topic, Message, 0, Opts}, State) ->
     Packet = amqtt_proto:encode_publish(Opts#{topic => Topic, message => Message, qos => 0}),
     case send_packet(Packet, State) of
         {ok, State1} -> {noreply, State1};
         {error, Reason, State1} -> stop_with_error(Reason, State1)
     end;
+handle_cast(ping, #state{pingresp_timer = Ref} = State) when Ref =/= undefined ->
+    %% A PINGREQ is already outstanding; keep the existing deadline.
+    {noreply, State};
 handle_cast(ping, State) ->
     case send_packet(amqtt_proto:encode_pingreq(), State) of
         {ok, State1} -> {noreply, arm_pingresp_timer(State1)};
         {error, Reason, State1} -> stop_with_error(Reason, State1)
     end.

Worth a small regression test alongside.

3. Minor — ping/1 with keep_alive_seconds = 0 is silently best-effort

arm_pingresp_timer(#state{keep_alive_seconds = 0}) returns the state
unchanged, so a manual ping/1 sends a PINGREQ but arms no watchdog.
Combined with the ping/1 docstring ("probe the link explicitly"), this
is ambiguous: the link is poked but not actually probed.

--- a/amqtt_client/src/amqtt_client.erl
+++ b/amqtt_client/src/amqtt_client.erl
@@
 %% The keep-alive timer already pings automatically; this is for
 %% applications that want to probe the link explicitly. Asynchronous:
-%% the PINGRESP is consumed silently by the gen_server.
+%% the PINGRESP is consumed silently by the gen_server. When
+%% `keep_alive_seconds = 0' no pingresp watchdog is armed, so the call
+%% is best-effort only.

4. Nit — KA=1 cadence tests may be tight on slow CI

no_false_timeout_under_publish_cadence_test_ and
inbound_publish_cancels_pingresp_watchdog_test_ rely on:

• ping at ~750 ms (KA·750)
• pingresp watchdog at ~1000 ms after ping
• a 300 ms sleep + publish/PUBACK round trip + 1200 ms wait window

On loopback OTP this is fine; on slow CI runners it's tight. If they
flake, bump KA to 2 and the wait windows accordingly:

--- a/amqtt_client/test/amqtt_client_tests.erl
+++ b/amqtt_client/test/amqtt_client_tests.erl
@@ no_false_timeout_under_publish_cadence_test_
-        Client = connect_and_handshake(Port, <<"c">>, 1),
+        Client = connect_and_handshake(Port, <<"c">>, 2),
@@
-        after 1200 ->
+        after 2200 ->
             ok
         end,

Don't change pingresp_timeout_fires_when_broker_silent_test_ unless it
flakes — short timing is a feature there.

---

What looks correct (no finding)

• Core fix. process_buffer/1 → observe_inbound/1 → handle_packet/3
  is the right place to clear the watchdog. Clearing on any complete
  inbound MQTT packet (not just PINGRESP) eliminates the false-timeout
  class described in the commit message.
• Stale-timer handling. keepalive_timer and pingresp_timer use
  erlang:start_timer/3, and handle_info matches both the live-ref
  and stale-ref forms of {timeout, Ref, Msg}. request_timeout keeps
  send_after/3 and its plain-message form, which is consistent.
• Initial timer arming. last_outbound_ms is set inside
  send_packet/2, so by the time handle_packet(connack, …) calls
  schedule_keepalive_timer/1 the value is already populated by the
  CONNECT send. The last_outbound_ms = undefined guard in
  schedule_keepalive_timer/1 is therefore defensive only — fine to
  keep.
• schedule_keepalive_timer/1 / handle_keepalive_check/1. Coherent.
  Not rescheduling on every outbound send is intentional and safe: the
  next firing of the timer recalculates against the current
  last_outbound_ms, so a long publish burst quietly keeps pushing the
  next PINGREQ out without timer churn.
• Order of arm_pingresp_timer/1 clauses. keep_alive_seconds = 0
  is matched first, then the "already armed, do not renew" clause, then
  the default — correct precedence.
• README updates. The "stops `normal' immediately afterwards"
  wording correctly mirrors the new exit reason.

---

Recommended path

1. Land the commit as-is.
2. Apply the handle_cast(ping, …) guard (Finding 2).
3. Apply the documentation tweaks for stop_with_error/2 exit reason
   (Finding 1) and ping/1 with KA=0 (Finding 3).
4. Only loosen the KA=1 cadence tests if they actually flake.

[bettio]

1. Major — stop_with_error/2 exit reason change has supervisor implications

I didn't address this. I suggest doing this on a subsequent PR. I'm looking for suggestion non-AI suggestions on this point.

[petermm]

imho it should just be documented - later options for the behaviour can be provided, and start/monitor can be supported etc.

Think it's good to keep it small and simple for now.