From 4eeb554921b97832f3dc948c9828d6e6634342ac Mon Sep 17 00:00:00 2001 From: Eric Blankenhorn Date: Wed, 15 Jul 2026 14:51:24 -0500 Subject: [PATCH 1/6] Add auto-ping feature to client --- ChangeLog.md | 19 ++ examples/mqttclient/mqttclient.c | 9 +- examples/multithread/multithread.c | 6 + examples/nbclient/nbclient.c | 10 +- src/mqtt_client.c | 128 ++++++++++ src/mqtt_packet.c | 11 +- tests/test_mqtt_client.c | 364 +++++++++++++++++++++++++++++ wolfmqtt/mqtt_client.h | 41 ++++ 8 files changed, 585 insertions(+), 3 deletions(-) diff --git a/ChangeLog.md b/ChangeLog.md index 2397ece21..6e5560203 100644 --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,24 @@ ## Release Notes +### v2.1.1 (In Development) + +* New Features + - Automatic client keep-alive: the core client now sends a `PINGREQ` from + `MqttClient_WaitMessage` / `MqttClient_WaitMessage_ex` once the outbound + link has been idle for about three quarters of the negotiated keep-alive + interval, so the application no longer has to schedule pings itself and + the ping reaches the broker before the deadline. Active whenever a + non-zero keep-alive is set in `CONNECT`, and honors a v5 CONNACK Server + Keep Alive override. A keep-alive ping left unanswered surfaces from + `MqttClient_WaitMessage` as `MQTT_CODE_ERROR_NETWORK` rather than a + timeout, so a dead link is distinguishable from an idle one. The time source is the + compile-time macro `WOLFMQTT_GET_TIME_S()` (defaults to `time(NULL)`, + overridable in `user_settings.h`); define `WOLFMQTT_NO_TIME` to compile + the scheduler out on clock-less targets, where the explicit + `MqttClient_Ping` APIs still work. The `mqttclient` example now relies on + the automatic ping, keeping its previous manual keep-alive loop under + `WOLFMQTT_NO_TIME` (#501) + ### v2.1.0 (07/02/2026) Release 2.1.0 has been developed according to wolfSSL's development and QA process (see link below) and successfully passed the quality criteria. diff --git a/examples/mqttclient/mqttclient.c b/examples/mqttclient/mqttclient.c index 215379a32..0037d82ae 100644 --- a/examples/mqttclient/mqttclient.c +++ b/examples/mqttclient/mqttclient.c @@ -598,7 +598,9 @@ int mqttclient_test(MQTTCtx *mqttCtx) } #endif else if (rc == MQTT_CODE_ERROR_TIMEOUT) { - /* Keep Alive */ + #ifdef WOLFMQTT_NO_TIME + /* Automatic keep-alive is compiled out (no time source), so the + * application schedules the PINGREQ itself on an idle timeout. */ PRINTF("Keep-alive timeout, sending ping"); rc = MqttClient_Ping_ex(&mqttCtx->client, &mqttCtx->ping); @@ -607,6 +609,11 @@ int mqttclient_test(MQTTCtx *mqttCtx) MqttClient_ReturnCodeToString(rc), rc); break; } + #else + /* The core client sends keep-alive PINGREQ automatically, so an + * idle timeout just means no message arrived; keep waiting. */ + rc = MQTT_CODE_SUCCESS; + #endif } else if (rc != MQTT_CODE_SUCCESS) { /* There was an error */ diff --git a/examples/multithread/multithread.c b/examples/multithread/multithread.c index 83df6bdcc..b2717c6d8 100644 --- a/examples/multithread/multithread.c +++ b/examples/multithread/multithread.c @@ -681,6 +681,12 @@ static void *publish_task(void *param) THREAD_EXIT(0); } +/* Dedicated keep-alive ping thread, retained to demonstrate explicit + * cross-thread ping coordination. When the core auto keep-alive is compiled in + * (the default, WOLFMQTT_NO_TIME not defined) the reader thread's + * MqttClient_WaitMessage_ex already schedules PINGREQ on its own; the two stay + * reconciled because each PINGREQ refreshes last_tx_time, which holds off the + * other path. Define WOLFMQTT_NO_TIME to rely solely on this thread. */ #ifdef USE_WINDOWS_API static DWORD WINAPI ping_task( LPVOID param ) #else diff --git a/examples/nbclient/nbclient.c b/examples/nbclient/nbclient.c index dbd1ef9e8..1d9f75565 100644 --- a/examples/nbclient/nbclient.c +++ b/examples/nbclient/nbclient.c @@ -524,11 +524,19 @@ int mqttclient_test(MQTTCtx *mqttCtx) else #endif if (rc == MQTT_CODE_ERROR_TIMEOUT) { - /* Need to send keep-alive ping */ + #ifdef WOLFMQTT_NO_TIME + /* Automatic keep-alive is compiled out, so the application + * schedules the PINGREQ itself on an idle timeout. */ PRINTF("Keep-alive timeout, sending ping"); rc = MQTT_CODE_CONTINUE; mqttCtx->stat = WMQ_PING; return rc; + #else + /* The core client schedules keep-alive PINGREQ itself, so + * an idle timeout just means no message arrived. */ + rc = MQTT_CODE_CONTINUE; + return rc; + #endif } else if (rc != MQTT_CODE_SUCCESS) { /* There was an error */ diff --git a/src/mqtt_client.c b/src/mqtt_client.c index ce4484db7..500875ca9 100644 --- a/src/mqtt_client.c +++ b/src/mqtt_client.c @@ -600,6 +600,16 @@ static void Handle_ConnectAck_Props(MqttClient* client, MqttProp* props) } } } + #ifndef WOLFMQTT_NO_TIME + else if (prop->type == MQTT_PROP_SERVER_KEEP_ALIVE) { + /* MQTT v5 [3.1.2.11.2]: when the broker returns a Server Keep + * Alive, the client MUST use it in place of the value it sent, and + * a value of 0 disables keep-alive. The flag lets the arming logic + * tell a server-provided 0 from an absent property. */ + client->keep_alive_sec = prop->data_short; + client->keep_alive_from_server = 1; + } + #endif } } @@ -1796,6 +1806,20 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect) return rc; } + #ifndef WOLFMQTT_NO_TIME + /* Disarm auto keep-alive during the handshake and clear any stale + * value from a previous connection on this client. It is armed only + * after CONNACK is accepted, at the end of this function. Also reset + * the ping state machine so a ping left mid-exchange by a prior + * connection (e.g. a WOLFMQTT_NONBLOCK ping, or an abnormal teardown + * with no MqttClient_Disconnect) cannot be resumed against the new + * connection and stall waiting for a PINGRESP that never arrives. */ + client->keep_alive_sec = 0; + client->keep_alive_from_server = 0; + client->keep_alive_ping.stat.write = MQTT_MSG_BEGIN; + client->keep_alive_ping.stat.read = MQTT_MSG_BEGIN; + #endif + #ifdef WOLFMQTT_V5 /* Use specified protocol version if set */ mc_connect->protocol_level = client->protocol_level; @@ -2006,6 +2030,26 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect) rc = MQTT_TRACE_ERROR(MQTT_CODE_ERROR_CONNECT_REFUSED); } +#ifndef WOLFMQTT_NO_TIME + if (rc == MQTT_CODE_SUCCESS) { + /* Connection accepted: arm auto keep-alive. A v5 Server Keep Alive + * (applied while processing CONNACK) takes precedence, including a + * value of 0 which disables keep-alive per [MQTT-3.1.2.11.2]; + * otherwise use the client-requested value. */ + if (!client->keep_alive_from_server) { + client->keep_alive_sec = mc_connect->keep_alive_sec; + } + /* Baseline the idle timer from the completed handshake so the first + * ping is scheduled a full interval out, not immediately. */ + client->last_tx_time = WOLFMQTT_GET_TIME_S(); + } + else { + /* Connect failed or was refused: leave the scheduler disarmed. */ + client->keep_alive_sec = 0; + } + client->keep_alive_from_server = 0; +#endif + return rc; } @@ -2858,6 +2902,15 @@ int MqttClient_Disconnect_ex(MqttClient *client, MqttDisconnect *p_disconnect) } if (disconnect->stat.write == MQTT_MSG_BEGIN) { + #ifndef WOLFMQTT_NO_TIME + /* Stop auto keep-alive scheduling for a client being torn down so a + * stale interval cannot arm a ping after disconnect. Also reset the + * ping state machine: a WOLFMQTT_NONBLOCK ping left mid-exchange would + * otherwise make a reconnect wait for a PINGRESP that was never sent. */ + client->keep_alive_sec = 0; + client->keep_alive_ping.stat.write = MQTT_MSG_BEGIN; + client->keep_alive_ping.stat.read = MQTT_MSG_BEGIN; + #endif #ifdef WOLFMQTT_V5 /* Use specified protocol version if set */ disconnect->protocol_level = client->protocol_level; @@ -3072,9 +3125,84 @@ int MqttClient_PropsFree(MqttProp *head) #endif /* WOLFMQTT_V5 */ +#ifndef WOLFMQTT_NO_TIME +/* Send a keep-alive PINGREQ when the outbound link has been idle for about + * three quarters of the negotiated keep-alive interval, so the application + * does not have to schedule pings itself. Called from the wait path. + * [MQTT-3.1.2-23] */ +static int MqttClient_KeepAlive(MqttClient *client) +{ + int rc = MQTT_CODE_SUCCESS; + word32 now, elapsed, threshold; + + if (client == NULL) { + return MQTT_TRACE_ERROR(MQTT_CODE_ERROR_BAD_ARG); + } + + /* Disabled until a non-zero keep-alive has been negotiated in CONNECT. */ + if (client->keep_alive_sec == 0) { + return MQTT_CODE_SUCCESS; + } + + /* Resume an in-progress ping exchange before evaluating the threshold. In + * WOLFMQTT_NONBLOCK mode MqttClient_Ping_ex can return MQTT_CODE_CONTINUE + * before the PINGRESP arrives, leaving the ping state machine mid-exchange. + * Drive it to completion here so a later ping is not entered with stale + * state, which would skip the PINGREQ and stretch the real ping interval. */ + if (client->keep_alive_ping.stat.write != MQTT_MSG_BEGIN) { + rc = MqttClient_Ping_ex(client, &client->keep_alive_ping); + if (rc == MQTT_CODE_ERROR_TIMEOUT) { + rc = MQTT_TRACE_ERROR(MQTT_CODE_ERROR_NETWORK); + } + return rc; + } + + /* Do not inject a ping into a partially read packet. */ + if (client->packet.stat != MQTT_PK_BEGIN) { + return MQTT_CODE_SUCCESS; + } + + now = WOLFMQTT_GET_TIME_S(); + if (now < client->last_tx_time) { + /* Clock stepped backward: re-baseline instead of pinging early. */ + client->last_tx_time = now; + return MQTT_CODE_SUCCESS; + } + + /* Ping at ~3/4 of the interval so the PINGREQ reaches the broker before + * the hard deadline, leaving headroom for network latency and the + * one-second clock granularity. Floor at one second so a small keep-alive + * still schedules a single ping instead of firing on every poll. */ + threshold = (word32)client->keep_alive_sec * 3 / 4; + if (threshold == 0) { + threshold = 1; + } + + elapsed = now - client->last_tx_time; + if (elapsed >= threshold) { + /* MqttPacket_Write refreshes last_tx_time as the PINGREQ is sent. */ + rc = MqttClient_Ping_ex(client, &client->keep_alive_ping); + if (rc == MQTT_CODE_ERROR_TIMEOUT) { + /* No PINGRESP within cmd_timeout_ms: the link is unresponsive, not + * merely idle. Surface a distinct error so a caller does not treat + * a failed keep-alive as an ordinary read timeout and keep looping + * on a dead connection. */ + rc = MQTT_TRACE_ERROR(MQTT_CODE_ERROR_NETWORK); + } + } + return rc; +} +#endif /* !WOLFMQTT_NO_TIME */ + int MqttClient_WaitMessage_ex(MqttClient *client, MqttObject* msg, int timeout_ms) { +#ifndef WOLFMQTT_NO_TIME + int rc = MqttClient_KeepAlive(client); + if (rc != MQTT_CODE_SUCCESS) { + return rc; + } +#endif return MqttClient_WaitType(client, msg, MQTT_PACKET_TYPE_ANY, 0, timeout_ms); } diff --git a/src/mqtt_packet.c b/src/mqtt_packet.c index b5f37a017..916031a69 100644 --- a/src/mqtt_packet.c +++ b/src/mqtt_packet.c @@ -3925,7 +3925,16 @@ int MqttPacket_Write(MqttClient *client, byte* tx_buf, int tx_buf_len) client->cmd_timeout_ms); } - return MqttPacket_HandleNetError(client, rc); + rc = MqttPacket_HandleNetError(client, rc); +#ifndef WOLFMQTT_NO_TIME + /* Record when a full control packet leaves the wire so the keep-alive + * scheduler can measure how long the outbound link has been idle. Only + * needed while auto-ping is armed; skip the clock read otherwise. */ + if (rc > 0 && rc == tx_buf_len && client->keep_alive_sec != 0) { + client->last_tx_time = WOLFMQTT_GET_TIME_S(); + } +#endif + return rc; } /* Read return code is length when > 0 */ diff --git a/tests/test_mqtt_client.c b/tests/test_mqtt_client.c index 7dfa620c9..110e8f666 100644 --- a/tests/test_mqtt_client.c +++ b/tests/test_mqtt_client.c @@ -2016,6 +2016,354 @@ TEST(wait_message_pubcomp_emits_no_ack) ASSERT_EQ(0, g_frames_written); } +/* ============================================================================ + * Automatic keep-alive (PINGREQ) scheduling Tests + * ============================================================================ */ + +#ifndef WOLFMQTT_NO_TIME +static int g_pingreq_writes; + +/* Counts PINGREQ frames (fixed header C0 00) written to the wire. */ +static int mock_net_write_count_ping(void *context, const byte* buf, + int buf_len, int timeout_ms) +{ + (void)context; (void)timeout_ms; + if (buf != NULL && buf_len >= 2 && buf[0] == 0xC0 && buf[1] == 0x00) { + g_pingreq_writes++; + } + return buf_len; +} + +/* Read side never delivers a packet, so the wait path stays idle. */ +static int mock_net_read_timeout(void *context, byte* buf, int buf_len, + int timeout_ms) +{ + (void)context; (void)buf; (void)buf_len; (void)timeout_ms; + return MQTT_CODE_ERROR_TIMEOUT; +} + +TEST(wait_message_auto_pings_on_keepalive_deadline) +{ + int rc; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + + g_pingreq_writes = 0; + test_net.write = mock_net_write_count_ping; + test_net.read = mock_net_read_timeout; + + /* Outbound link idle well past the negotiated keep-alive: the wait path + * must inject a PINGREQ on its own. last_tx_time=0 with a real clock makes + * the elapsed interval far exceed keep_alive_sec. */ + test_client.keep_alive_sec = 1; + test_client.last_tx_time = 0; + + rc = MqttClient_WaitMessage(&test_client, 0); + + /* The core scheduled the keep-alive PINGREQ itself. The mock never answers + * with a PINGRESP, so the unanswered ping is surfaced as a network error + * (not a plain timeout) so callers can tell a dead link from an idle one. */ + ASSERT_EQ(MQTT_CODE_ERROR_NETWORK, rc); + ASSERT_TRUE(g_pingreq_writes >= 1); + + /* With the timer now ahead of the clock (simulated fresh activity), a + * second call must not ping again; with no ping due the wait reports an + * ordinary idle timeout. This also covers the backward-clock re-baseline + * guard. */ + g_pingreq_writes = 0; + test_client.last_tx_time = 0xFFFFFFFFUL; + rc = MqttClient_WaitMessage(&test_client, 0); + ASSERT_EQ(MQTT_CODE_ERROR_TIMEOUT, rc); + ASSERT_EQ(0, g_pingreq_writes); +} + +TEST(wait_message_no_autoping_when_keepalive_zero) +{ + int rc; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + + g_pingreq_writes = 0; + test_net.write = mock_net_write_count_ping; + test_net.read = mock_net_read_timeout; + + /* Keep-alive of zero disables the mechanism entirely. */ + test_client.keep_alive_sec = 0; + test_client.last_tx_time = 0; + + rc = MqttClient_WaitMessage(&test_client, 0); + ASSERT_EQ(MQTT_CODE_ERROR_TIMEOUT, rc); + ASSERT_EQ(0, g_pingreq_writes); +} + +TEST(wait_message_auto_pings_before_full_interval_with_margin) +{ + int rc; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + + g_pingreq_writes = 0; + test_net.write = mock_net_write_count_ping; + test_net.read = mock_net_read_timeout; + + /* Idle for 90s of a 100s keep-alive: past the ~3/4 (75s) margin but short + * of the full interval. The ping must fire early, not wait for the hard + * deadline. This fails if the trigger is elapsed >= keep_alive_sec. The + * unanswered ping surfaces as a network error. */ + test_client.keep_alive_sec = 100; + test_client.last_tx_time = WOLFMQTT_GET_TIME_S() - 90; + + rc = MqttClient_WaitMessage(&test_client, 0); + ASSERT_EQ(MQTT_CODE_ERROR_NETWORK, rc); + ASSERT_TRUE(g_pingreq_writes >= 1); +} + +TEST(wait_message_no_autoping_when_link_recently_active) +{ + int rc; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + + g_pingreq_writes = 0; + test_net.write = mock_net_write_count_ping; + test_net.read = mock_net_read_timeout; + + /* Outbound link active within the interval: elapsed (now - last_tx_time) + * is small and stays below keep_alive_sec, so the elapsed >= keep_alive + * branch must not fire. Exercises the common steady-state no-ping path. */ + test_client.keep_alive_sec = 3600; + test_client.last_tx_time = WOLFMQTT_GET_TIME_S(); + + rc = MqttClient_WaitMessage(&test_client, 0); + ASSERT_EQ(MQTT_CODE_ERROR_TIMEOUT, rc); + ASSERT_EQ(0, g_pingreq_writes); +} + +TEST(wait_message_no_autoping_during_partial_read) +{ + int rc; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + + g_pingreq_writes = 0; + test_net.write = mock_net_write_count_ping; + test_net.read = mock_net_read_timeout; + + /* Deadline is well past, but a packet read is in progress: the ping must + * not be injected mid-packet. Guards the partial-read check. */ + test_client.keep_alive_sec = 1; + test_client.last_tx_time = 0; + test_client.packet.stat = MQTT_PK_READ_HEAD; + + rc = MqttClient_WaitMessage(&test_client, 0); + ASSERT_EQ(0, g_pingreq_writes); + /* The guard must not swallow the wait result: it still reports the idle + * read timeout rather than an unexpected success or wrong error. */ + ASSERT_EQ(MQTT_CODE_ERROR_TIMEOUT, rc); + + /* Restore so the state machine is clean for later assertions. */ + test_client.packet.stat = MQTT_PK_BEGIN; +} + +TEST(wait_message_resumes_inflight_ping) +{ + int rc; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + + g_pingreq_writes = 0; + test_net.write = mock_net_write_count_ping; + test_net.read = mock_net_read_timeout; + + /* Simulate a ping exchange left in progress (a WOLFMQTT_NONBLOCK ping whose + * PINGRESP had not yet arrived), with the interval not yet re-elapsed. The + * wait path must resume and complete that exchange instead of leaving the + * ping state machine stuck at MQTT_MSG_WAIT and later re-entering it with + * stale state. */ + test_client.keep_alive_sec = 3600; + test_client.last_tx_time = WOLFMQTT_GET_TIME_S(); + test_client.keep_alive_ping.stat.write = MQTT_MSG_WAIT; + + rc = MqttClient_WaitMessage(&test_client, 0); + + /* Resume drove Ping_ex, which reset the state machine. No new PINGREQ was + * encoded (the original was already sent), and the unanswered wait surfaced + * as a network error. */ + ASSERT_EQ(MQTT_MSG_BEGIN, test_client.keep_alive_ping.stat.write); + ASSERT_EQ(0, g_pingreq_writes); + ASSERT_EQ(MQTT_CODE_ERROR_NETWORK, rc); +} + +TEST(connect_arms_keepalive_and_disconnect_disarms) +{ + int rc; + int i; + MqttConnect connect; + /* CONNACK v3.1.1: accepted. */ + static const byte connack[] = { 0x20, 0x02, 0x00, 0x00 }; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); +#ifdef WOLFMQTT_V5 + test_client.protocol_level = MQTT_CONNECT_PROTOCOL_LEVEL_4; +#endif + + test_net.write = mock_net_write_accept; + test_net.read = mock_net_read_canned; + XMEMCPY(g_canned_buf, connack, sizeof(connack)); + g_canned_len = (int)sizeof(connack); + g_canned_pos = 0; + + XMEMSET(&connect, 0, sizeof(connect)); + connect.keep_alive_sec = 45; + connect.clean_session = 1; + connect.client_id = "test_client"; + + /* Simulate a ping left mid-exchange by a prior connection on this same + * client object (e.g. an abnormal teardown with no MqttClient_Disconnect). + * A successful connect must reset it so the first wait does not resume a + * phantom ping. */ + test_client.keep_alive_ping.stat.write = MQTT_MSG_WAIT; + + rc = MQTT_CODE_CONTINUE; + for (i = 0; i < 10 && rc == MQTT_CODE_CONTINUE; i++) { + rc = MqttClient_Connect(&test_client, &connect); + } + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + + /* Accepted connection arms the scheduler with the requested interval, + * baselines the idle timer, and clears the stale ping state. */ + ASSERT_EQ(45, test_client.keep_alive_sec); + ASSERT_TRUE(test_client.last_tx_time != 0); + ASSERT_EQ(MQTT_MSG_BEGIN, test_client.keep_alive_ping.stat.write); + + /* Disconnect disarms the scheduler and resets the ping state machine. */ + rc = MQTT_CODE_CONTINUE; + for (i = 0; i < 10 && rc == MQTT_CODE_CONTINUE; i++) { + rc = MqttClient_Disconnect(&test_client); + } + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + ASSERT_EQ(0, test_client.keep_alive_sec); + ASSERT_EQ(MQTT_MSG_BEGIN, test_client.keep_alive_ping.stat.write); +} + +TEST(connect_refused_leaves_keepalive_disarmed) +{ + int rc; + int i; + MqttConnect connect; + /* CONNACK v3.1.1: refused (0x05 = not authorized). */ + static const byte connack[] = { 0x20, 0x02, 0x00, 0x05 }; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); +#ifdef WOLFMQTT_V5 + test_client.protocol_level = MQTT_CONNECT_PROTOCOL_LEVEL_4; +#endif + + test_net.write = mock_net_write_accept; + test_net.read = mock_net_read_canned; + XMEMCPY(g_canned_buf, connack, sizeof(connack)); + g_canned_len = (int)sizeof(connack); + g_canned_pos = 0; + + XMEMSET(&connect, 0, sizeof(connect)); + connect.keep_alive_sec = 45; + connect.clean_session = 1; + connect.client_id = "test_client"; + + rc = MQTT_CODE_CONTINUE; + for (i = 0; i < 10 && rc == MQTT_CODE_CONTINUE; i++) { + rc = MqttClient_Connect(&test_client, &connect); + } + ASSERT_EQ(MQTT_CODE_ERROR_CONNECT_REFUSED, rc); + + /* A refused connection must not arm the scheduler. */ + ASSERT_EQ(0, test_client.keep_alive_sec); +} + +#if defined(WOLFMQTT_V5) +TEST(connect_v5_server_keep_alive_overrides_requested) +{ + int rc; + int i; + MqttConnect connect; + /* CONNACK v5 accepted, prop_len=3, [0x13 00 1E] = Server Keep Alive 30. */ + static const byte connack[] = { + 0x20, 0x06, 0x00, 0x00, 0x03, + 0x13, 0x00, 0x1E + }; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + test_client.protocol_level = MQTT_CONNECT_PROTOCOL_LEVEL_5; + + test_net.write = mock_net_write_accept; + test_net.read = mock_net_read_canned; + XMEMCPY(g_canned_buf, connack, sizeof(connack)); + g_canned_len = (int)sizeof(connack); + g_canned_pos = 0; + + XMEMSET(&connect, 0, sizeof(connect)); + connect.keep_alive_sec = 60; + connect.clean_session = 1; + connect.client_id = "test_client"; + + rc = MQTT_CODE_CONTINUE; + for (i = 0; i < 10 && rc == MQTT_CODE_CONTINUE; i++) { + rc = MqttClient_Connect(&test_client, &connect); + } + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + + /* The broker's Server Keep Alive (30) must replace the requested 60. */ + ASSERT_EQ(30, test_client.keep_alive_sec); +} + +TEST(connect_v5_server_keep_alive_zero_disables) +{ + int rc; + int i; + MqttConnect connect; + /* CONNACK v5 accepted, prop_len=3, [0x13 00 00] = Server Keep Alive 0. */ + static const byte connack[] = { + 0x20, 0x06, 0x00, 0x00, 0x03, + 0x13, 0x00, 0x00 + }; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + test_client.protocol_level = MQTT_CONNECT_PROTOCOL_LEVEL_5; + + test_net.write = mock_net_write_accept; + test_net.read = mock_net_read_canned; + XMEMCPY(g_canned_buf, connack, sizeof(connack)); + g_canned_len = (int)sizeof(connack); + g_canned_pos = 0; + + XMEMSET(&connect, 0, sizeof(connect)); + connect.keep_alive_sec = 60; + connect.clean_session = 1; + connect.client_id = "test_client"; + + rc = MQTT_CODE_CONTINUE; + for (i = 0; i < 10 && rc == MQTT_CODE_CONTINUE; i++) { + rc = MqttClient_Connect(&test_client, &connect); + } + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + + /* A Server Keep Alive of 0 disables keep-alive per [MQTT-3.1.2.11.2]; the + * client must not fall back to its requested 60. */ + ASSERT_EQ(0, test_client.keep_alive_sec); +} +#endif /* WOLFMQTT_V5 */ +#endif /* !WOLFMQTT_NO_TIME */ + /* ============================================================================ * MqttClient_ReturnCodeToString Tests * ============================================================================ */ @@ -2176,6 +2524,22 @@ void run_mqtt_client_tests(void) RUN_TEST(wait_message_puback_emits_no_ack); RUN_TEST(wait_message_pubcomp_emits_no_ack); +#ifndef WOLFMQTT_NO_TIME + /* Automatic keep-alive (PINGREQ) scheduling tests */ + RUN_TEST(wait_message_auto_pings_on_keepalive_deadline); + RUN_TEST(wait_message_auto_pings_before_full_interval_with_margin); + RUN_TEST(wait_message_no_autoping_when_keepalive_zero); + RUN_TEST(wait_message_no_autoping_when_link_recently_active); + RUN_TEST(wait_message_no_autoping_during_partial_read); + RUN_TEST(wait_message_resumes_inflight_ping); + RUN_TEST(connect_arms_keepalive_and_disconnect_disarms); + RUN_TEST(connect_refused_leaves_keepalive_disarmed); +#if defined(WOLFMQTT_V5) + RUN_TEST(connect_v5_server_keep_alive_overrides_requested); + RUN_TEST(connect_v5_server_keep_alive_zero_disables); +#endif +#endif + #ifndef WOLFMQTT_NO_ERROR_STRINGS /* MqttClient_ReturnCodeToString tests */ RUN_TEST(return_code_to_string_success); diff --git a/wolfmqtt/mqtt_client.h b/wolfmqtt/mqtt_client.h index fe0808783..d2ccad8ec 100644 --- a/wolfmqtt/mqtt_client.h +++ b/wolfmqtt/mqtt_client.h @@ -46,6 +46,41 @@ #include "wolfmqtt/mqtt_sn_packet.h" #endif +/* Automatic keep-alive scheduling. When a non-zero keep-alive is negotiated in + * CONNECT, the core client sends a PINGREQ once the outbound link has been idle + * for about three quarters of that interval, so the application does not have + * to schedule pings itself and the ping reaches the broker before the deadline. + * This mirrors the broker keep-alive handling. + * + * The scheduler is evaluated at the start of each MqttClient_WaitMessage call, + * not during the blocking read, so the application must call it with a + * timeout_ms (poll cadence) shorter than the keep-alive interval. A single + * wait longer than the interval can let the broker deadline pass unchecked. + * + * When a ping is due, MqttClient_WaitMessage performs the PINGREQ/PINGRESP + * round-trip inline, so the call may take up to cmd_timeout_ms longer to + * return. A ping left unanswered returns MQTT_CODE_ERROR_NETWORK rather than + * MQTT_CODE_ERROR_TIMEOUT, so a caller can tell a dead link from an idle one. + * + * WOLFMQTT_GET_TIME_S() returns a clock in seconds; only elapsed differences + * are used, so a monotonic source is fine. The default is time(NULL), except + * on Zephyr where its minimal libc has no linkable time() and the kernel + * uptime is used instead. Override it in user_settings.h on targets without a + * suitable clock, or define WOLFMQTT_NO_TIME to compile out automatic + * scheduling entirely; the explicit MqttClient_Ping / MqttClient_Ping_ex APIs + * still work in that case. */ +#ifndef WOLFMQTT_NO_TIME + #ifndef WOLFMQTT_GET_TIME_S + #ifdef __ZEPHYR__ + #include + #define WOLFMQTT_GET_TIME_S() ((word32)(k_uptime_get() / 1000)) + #else + #include + #define WOLFMQTT_GET_TIME_S() ((word32)time(NULL)) + #endif + #endif +#endif + /* This macro allows the disconnect callback to be triggered when * MqttClient_Disconnect_ex is called. Normally the CB is only used to handle @@ -219,6 +254,12 @@ typedef struct _MqttClient { #if defined(WOLFMQTT_NONBLOCK) && defined(WOLFMQTT_DEBUG_CLIENT) int lastRc; #endif +#ifndef WOLFMQTT_NO_TIME + MqttPing keep_alive_ping; /* dedicated object for auto keep-alive PINGREQ */ + word32 last_tx_time; /* WOLFMQTT_GET_TIME_S() at last control-pkt TX */ + word16 keep_alive_sec; /* negotiated keep-alive; 0 disables auto-ping */ + byte keep_alive_from_server; /* v5 Server Keep Alive applied (0 valid) */ +#endif } MqttClient; #ifdef WOLFMQTT_SN From dbca08fffd6eb3c924805b2385ad301fc581572f Mon Sep 17 00:00:00 2001 From: Eric Blankenhorn Date: Thu, 16 Jul 2026 10:17:50 -0500 Subject: [PATCH 2/6] Fixes from review --- ChangeLog.md | 8 ++-- examples/multithread/multithread.c | 10 ++--- src/mqtt_client.c | 69 +++++++++++++++++++----------- src/mqtt_packet.c | 2 +- tests/test_mqtt_client.c | 46 ++++++++++++++++++-- wolfmqtt/mqtt_client.h | 17 ++++++-- 6 files changed, 111 insertions(+), 41 deletions(-) diff --git a/ChangeLog.md b/ChangeLog.md index 6e5560203..06b1d3efe 100644 --- a/ChangeLog.md +++ b/ChangeLog.md @@ -9,9 +9,11 @@ interval, so the application no longer has to schedule pings itself and the ping reaches the broker before the deadline. Active whenever a non-zero keep-alive is set in `CONNECT`, and honors a v5 CONNACK Server - Keep Alive override. A keep-alive ping left unanswered surfaces from - `MqttClient_WaitMessage` as `MQTT_CODE_ERROR_NETWORK` rather than a - timeout, so a dead link is distinguishable from an idle one. The time source is the + Keep Alive override. In a blocking build a keep-alive ping left + unanswered surfaces from `MqttClient_WaitMessage` as + `MQTT_CODE_ERROR_NETWORK` rather than a timeout, so a dead link is + distinguishable from an idle one; under `WOLFMQTT_NONBLOCK` the + application remains responsible for its own liveness deadline. The time source is the compile-time macro `WOLFMQTT_GET_TIME_S()` (defaults to `time(NULL)`, overridable in `user_settings.h`); define `WOLFMQTT_NO_TIME` to compile the scheduler out on clock-less targets, where the explicit diff --git a/examples/multithread/multithread.c b/examples/multithread/multithread.c index b2717c6d8..ed3991e02 100644 --- a/examples/multithread/multithread.c +++ b/examples/multithread/multithread.c @@ -682,11 +682,11 @@ static void *publish_task(void *param) } /* Dedicated keep-alive ping thread, retained to demonstrate explicit - * cross-thread ping coordination. When the core auto keep-alive is compiled in - * (the default, WOLFMQTT_NO_TIME not defined) the reader thread's - * MqttClient_WaitMessage_ex already schedules PINGREQ on its own; the two stay - * reconciled because each PINGREQ refreshes last_tx_time, which holds off the - * other path. Define WOLFMQTT_NO_TIME to rely solely on this thread. */ + * cross-thread ping coordination. With the core auto keep-alive compiled in + * (the default) the reader thread's MqttClient_WaitMessage_ex also sends + * PINGREQ. last_tx_time only holds off the reader's own next ping, not this + * thread, so both paths can ping and the broker sees redundant PINGREQs. + * Define WOLFMQTT_NO_TIME to rely solely on this thread. */ #ifdef USE_WINDOWS_API static DWORD WINAPI ping_task( LPVOID param ) #else diff --git a/src/mqtt_client.c b/src/mqtt_client.c index 500875ca9..e7e080fae 100644 --- a/src/mqtt_client.c +++ b/src/mqtt_client.c @@ -1801,25 +1801,24 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect) } #endif - /* Flag write active / lock mutex */ - if ((rc = MqttWriteStart(client, &mc_connect->stat)) != 0) { - return rc; - } - #ifndef WOLFMQTT_NO_TIME - /* Disarm auto keep-alive during the handshake and clear any stale - * value from a previous connection on this client. It is armed only - * after CONNACK is accepted, at the end of this function. Also reset - * the ping state machine so a ping left mid-exchange by a prior - * connection (e.g. a WOLFMQTT_NONBLOCK ping, or an abnormal teardown - * with no MqttClient_Disconnect) cannot be resumed against the new - * connection and stall waiting for a PINGRESP that never arrives. */ + /* Disarm auto keep-alive for the handshake; it is armed only after + * CONNACK is accepted at the end of this function. Fully cancel any + * ping left mid-exchange by a prior connection so its held locks and + * pending response are released before the new write starts, rather + * than leaving the ping state machine to stall a later wait. Must run + * before MqttWriteStart so the send lock is free when it is taken. */ client->keep_alive_sec = 0; client->keep_alive_from_server = 0; - client->keep_alive_ping.stat.write = MQTT_MSG_BEGIN; - client->keep_alive_ping.stat.read = MQTT_MSG_BEGIN; + MqttClient_CancelMessage(client, + (MqttObject*)&client->keep_alive_ping); #endif + /* Flag write active / lock mutex */ + if ((rc = MqttWriteStart(client, &mc_connect->stat)) != 0) { + return rc; + } + #ifdef WOLFMQTT_V5 /* Use specified protocol version if set */ mc_connect->protocol_level = client->protocol_level; @@ -2903,13 +2902,13 @@ int MqttClient_Disconnect_ex(MqttClient *client, MqttDisconnect *p_disconnect) if (disconnect->stat.write == MQTT_MSG_BEGIN) { #ifndef WOLFMQTT_NO_TIME - /* Stop auto keep-alive scheduling for a client being torn down so a - * stale interval cannot arm a ping after disconnect. Also reset the - * ping state machine: a WOLFMQTT_NONBLOCK ping left mid-exchange would - * otherwise make a reconnect wait for a PINGRESP that was never sent. */ + /* Stop auto keep-alive for a client being torn down, and fully cancel + * any ping left mid-exchange so its held locks and pending response are + * released before the disconnect write starts. A bare stat reset would + * strand client->write.isActive and livelock MqttClient_Disconnect. */ client->keep_alive_sec = 0; - client->keep_alive_ping.stat.write = MQTT_MSG_BEGIN; - client->keep_alive_ping.stat.read = MQTT_MSG_BEGIN; + MqttClient_CancelMessage(client, + (MqttObject*)&client->keep_alive_ping); #endif #ifdef WOLFMQTT_V5 /* Use specified protocol version if set */ @@ -3130,7 +3129,7 @@ int MqttClient_PropsFree(MqttProp *head) * three quarters of the negotiated keep-alive interval, so the application * does not have to schedule pings itself. Called from the wait path. * [MQTT-3.1.2-23] */ -static int MqttClient_KeepAlive(MqttClient *client) +static int MqttClient_KeepAlive(MqttClient *client, MqttObject* msg) { int rc = MQTT_CODE_SUCCESS; word32 now, elapsed, threshold; @@ -3157,8 +3156,12 @@ static int MqttClient_KeepAlive(MqttClient *client) return rc; } - /* Do not inject a ping into a partially read packet. */ - if (client->packet.stat != MQTT_PK_BEGIN) { + /* Do not inject a ping while a message is mid-transfer: a partially read + * fixed header (packet.stat), or a payload read still in progress on the + * wait object (msg->read), which packet.stat alone does not catch once the + * header has been consumed but the read lock is still held. */ + if (client->packet.stat != MQTT_PK_BEGIN || + (msg != NULL && ((MqttMsgStat*)msg)->read != MQTT_MSG_BEGIN)) { return MQTT_CODE_SUCCESS; } @@ -3198,8 +3201,24 @@ int MqttClient_WaitMessage_ex(MqttClient *client, MqttObject* msg, int timeout_ms) { #ifndef WOLFMQTT_NO_TIME - int rc = MqttClient_KeepAlive(client); - if (rc != MQTT_CODE_SUCCESS) { + int rc; +#endif + + if (client == NULL || msg == NULL) { + return MQTT_TRACE_ERROR(MQTT_CODE_ERROR_BAD_ARG); + } + +#ifndef WOLFMQTT_NO_TIME + /* Send an automatic keep-alive PINGREQ if the outbound link has been idle + * long enough. A deferred ping (another thread holds the write lock) comes + * back as MQTT_CODE_CONTINUE, which is not an error, so in a blocking build + * fall through to the normal wait rather than returning it to the caller. */ + rc = MqttClient_KeepAlive(client, msg); + if (rc != MQTT_CODE_SUCCESS + #ifndef WOLFMQTT_NONBLOCK + && rc != MQTT_CODE_CONTINUE + #endif + ) { return rc; } #endif diff --git a/src/mqtt_packet.c b/src/mqtt_packet.c index 916031a69..22ee41000 100644 --- a/src/mqtt_packet.c +++ b/src/mqtt_packet.c @@ -3930,7 +3930,7 @@ int MqttPacket_Write(MqttClient *client, byte* tx_buf, int tx_buf_len) /* Record when a full control packet leaves the wire so the keep-alive * scheduler can measure how long the outbound link has been idle. Only * needed while auto-ping is armed; skip the clock read otherwise. */ - if (rc > 0 && rc == tx_buf_len && client->keep_alive_sec != 0) { + if (rc == tx_buf_len && client->keep_alive_sec != 0) { client->last_tx_time = WOLFMQTT_GET_TIME_S(); } #endif diff --git a/tests/test_mqtt_client.c b/tests/test_mqtt_client.c index 110e8f666..04aa4b175 100644 --- a/tests/test_mqtt_client.c +++ b/tests/test_mqtt_client.c @@ -2101,6 +2101,7 @@ TEST(wait_message_no_autoping_when_keepalive_zero) TEST(wait_message_auto_pings_before_full_interval_with_margin) { int rc; + word32 now; rc = test_init_client(); ASSERT_EQ(MQTT_CODE_SUCCESS, rc); @@ -2111,10 +2112,12 @@ TEST(wait_message_auto_pings_before_full_interval_with_margin) /* Idle for 90s of a 100s keep-alive: past the ~3/4 (75s) margin but short * of the full interval. The ping must fire early, not wait for the hard - * deadline. This fails if the trigger is elapsed >= keep_alive_sec. The - * unanswered ping surfaces as a network error. */ + * deadline. This fails if the trigger is elapsed >= keep_alive_sec. Clamp + * the subtraction so a monotonic clock near zero does not underflow + * last_tx_time and silently exercise the backward-clock guard instead. */ + now = WOLFMQTT_GET_TIME_S(); test_client.keep_alive_sec = 100; - test_client.last_tx_time = WOLFMQTT_GET_TIME_S() - 90; + test_client.last_tx_time = (now > 90) ? (now - 90) : 0; rc = MqttClient_WaitMessage(&test_client, 0); ASSERT_EQ(MQTT_CODE_ERROR_NETWORK, rc); @@ -2170,6 +2173,35 @@ TEST(wait_message_no_autoping_during_partial_read) test_client.packet.stat = MQTT_PK_BEGIN; } +TEST(wait_message_no_autoping_during_partial_payload) +{ + int rc; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + + g_pingreq_writes = 0; + test_net.write = mock_net_write_count_ping; + test_net.read = mock_net_read_timeout; + + /* Deadline is well past and the fixed header has been fully read + * (packet.stat == MQTT_PK_BEGIN), but a payload read is still in progress + * on the wait object. The ping must not be injected mid-payload; the + * MQTT_PK_READ_HEAD guard alone does not cover this case. */ + test_client.keep_alive_sec = 1; + test_client.last_tx_time = 0; + test_client.packet.stat = MQTT_PK_BEGIN; + test_client.msg.publish.stat.read = MQTT_MSG_PAYLOAD2; + + rc = MqttClient_WaitMessage(&test_client, 0); + /* The guard suppressed the ping regardless of how the resumed read ends. */ + ASSERT_EQ(0, g_pingreq_writes); + (void)rc; + + /* Restore so the state machine is clean for later assertions. */ + test_client.msg.publish.stat.read = MQTT_MSG_BEGIN; +} + TEST(wait_message_resumes_inflight_ping) { int rc; @@ -2231,6 +2263,11 @@ TEST(connect_arms_keepalive_and_disconnect_disarms) * phantom ping. */ test_client.keep_alive_ping.stat.write = MQTT_MSG_WAIT; + /* Seed a sentinel so the arm assertion can check the idle timer was + * baselined without assuming the clock is ever non-zero (a monotonic + * source can legitimately return 0 in the first second of uptime). */ + test_client.last_tx_time = 0x5A5A5A5AUL; + rc = MQTT_CODE_CONTINUE; for (i = 0; i < 10 && rc == MQTT_CODE_CONTINUE; i++) { rc = MqttClient_Connect(&test_client, &connect); @@ -2240,7 +2277,7 @@ TEST(connect_arms_keepalive_and_disconnect_disarms) /* Accepted connection arms the scheduler with the requested interval, * baselines the idle timer, and clears the stale ping state. */ ASSERT_EQ(45, test_client.keep_alive_sec); - ASSERT_TRUE(test_client.last_tx_time != 0); + ASSERT_TRUE(test_client.last_tx_time != 0x5A5A5A5AUL); ASSERT_EQ(MQTT_MSG_BEGIN, test_client.keep_alive_ping.stat.write); /* Disconnect disarms the scheduler and resets the ping state machine. */ @@ -2531,6 +2568,7 @@ void run_mqtt_client_tests(void) RUN_TEST(wait_message_no_autoping_when_keepalive_zero); RUN_TEST(wait_message_no_autoping_when_link_recently_active); RUN_TEST(wait_message_no_autoping_during_partial_read); + RUN_TEST(wait_message_no_autoping_during_partial_payload); RUN_TEST(wait_message_resumes_inflight_ping); RUN_TEST(connect_arms_keepalive_and_disconnect_disarms); RUN_TEST(connect_refused_leaves_keepalive_disarmed); diff --git a/wolfmqtt/mqtt_client.h b/wolfmqtt/mqtt_client.h index d2ccad8ec..5a4bfc8a5 100644 --- a/wolfmqtt/mqtt_client.h +++ b/wolfmqtt/mqtt_client.h @@ -58,9 +58,20 @@ * wait longer than the interval can let the broker deadline pass unchecked. * * When a ping is due, MqttClient_WaitMessage performs the PINGREQ/PINGRESP - * round-trip inline, so the call may take up to cmd_timeout_ms longer to - * return. A ping left unanswered returns MQTT_CODE_ERROR_NETWORK rather than - * MQTT_CODE_ERROR_TIMEOUT, so a caller can tell a dead link from an idle one. + * round-trip inline using cmd_timeout_ms, so the call may take up to + * cmd_timeout_ms longer to return; if other packets keep arriving the round + * trip can extend past a single cmd_timeout_ms before the PINGRESP is seen. A + * packet that arrives while waiting for the PINGRESP is delivered through the + * message callback, not the MqttObject passed to MqttClient_WaitMessage_ex. + * + * In a blocking build a ping left unanswered returns MQTT_CODE_ERROR_NETWORK + * rather than MQTT_CODE_ERROR_TIMEOUT, so a caller can tell a dead link from an + * idle one. Under WOLFMQTT_NONBLOCK the library cannot time the ping out, so + * the application stays responsible for its own liveness deadline. + * + * The keep-alive bookkeeping fields (last_tx_time, keep_alive_sec) are updated + * without a lock under WOLFMQTT_MULTITHREAD; a mis-scheduled or skipped ping + * self-corrects on the next MqttClient_WaitMessage. * * WOLFMQTT_GET_TIME_S() returns a clock in seconds; only elapsed differences * are used, so a monotonic source is fine. The default is time(NULL), except From 31b55ebda7e841a6a0b743029a11e684b99f9c3b Mon Sep 17 00:00:00 2001 From: Eric Blankenhorn Date: Thu, 16 Jul 2026 16:05:33 -0500 Subject: [PATCH 3/6] Fixes from review --- ChangeLog.md | 15 +++- examples/multithread/multithread.c | 40 +++++++-- src/mqtt_client.c | 20 ++++- tests/test_mqtt_client.c | 137 +++++++++++++++++++++++++++++ wolfmqtt/mqtt_client.h | 15 +++- 5 files changed, 208 insertions(+), 19 deletions(-) diff --git a/ChangeLog.md b/ChangeLog.md index 06b1d3efe..ffcf227aa 100644 --- a/ChangeLog.md +++ b/ChangeLog.md @@ -13,13 +13,20 @@ unanswered surfaces from `MqttClient_WaitMessage` as `MQTT_CODE_ERROR_NETWORK` rather than a timeout, so a dead link is distinguishable from an idle one; under `WOLFMQTT_NONBLOCK` the - application remains responsible for its own liveness deadline. The time source is the + application remains responsible for its own liveness deadline. Because the + ping round-trip runs inline using `cmd_timeout_ms`, a + `MqttClient_WaitMessage` call that starts a keep-alive can take up to + `cmd_timeout_ms` longer to return, so poll with a `timeout_ms` shorter + than the keep-alive interval. The time source is the compile-time macro `WOLFMQTT_GET_TIME_S()` (defaults to `time(NULL)`, overridable in `user_settings.h`); define `WOLFMQTT_NO_TIME` to compile the scheduler out on clock-less targets, where the explicit - `MqttClient_Ping` APIs still work. The `mqttclient` example now relies on - the automatic ping, keeping its previous manual keep-alive loop under - `WOLFMQTT_NO_TIME` (#501) + `MqttClient_Ping` APIs still work. An application that already sends its + own `PINGREQ` can disable the core scheduler at runtime - without changing + the negotiated keep-alive - by setting `MQTT_CLIENT_FLAG_NO_AUTO_KEEPALIVE` + with `MqttClient_Flags` before connecting, avoiding duplicate pings. The + `mqttclient` example now relies on the automatic ping, keeping its + previous manual keep-alive loop under `WOLFMQTT_NO_TIME` (#501) ### v2.1.0 (07/02/2026) Release 2.1.0 has been developed according to wolfSSL's development and QA diff --git a/examples/multithread/multithread.c b/examples/multithread/multithread.c index ed3991e02..d869f6745 100644 --- a/examples/multithread/multithread.c +++ b/examples/multithread/multithread.c @@ -79,7 +79,12 @@ static int mNumMsgsDone; #endif static wm_Sem mtLock; /* Protect "packetId" and "stop" */ +#ifdef WOLFMQTT_NO_TIME +/* Only needed when the core auto keep-alive is compiled out; otherwise + * MqttClient_WaitMessage_ex schedules the PINGREQ and this dedicated ping + * thread would be a second keep-alive owner on the same connection. */ static wm_Sem pingSignal; +#endif static MQTTCtx gMqttCtx; @@ -269,11 +274,13 @@ static int multithread_test_init(MQTTCtx *mqttCtx) if (rc != 0) { client_exit(mqttCtx); } +#ifdef WOLFMQTT_NO_TIME rc = wm_SemInit(&pingSignal); if (rc != 0) { wm_SemFree(&mtLock); client_exit(mqttCtx); } +#endif PRINTF("MQTT Client: QoS %d, Use TLS %d", mqttCtx->qos, mqttCtx->use_tls); @@ -393,7 +400,9 @@ static int multithread_test_finish(MQTTCtx *mqttCtx) { client_disconnect(mqttCtx); +#ifdef WOLFMQTT_NO_TIME wm_SemFree(&pingSignal); +#endif wm_SemFree(&mtLock); PRINTF("MQTT Client Done: %d", mqttCtx->return_code); @@ -518,6 +527,7 @@ static void *waitMessage_task(void *param) MQTTCtx *mqttCtx = (MQTTCtx*)param; word32 startSec; word32 cmd_timeout_ms = mqttCtx->cmd_timeout_ms; +#ifdef WOLFMQTT_NO_TIME int needsUnlock = 0; if (wm_SemLock(&pingSignal) != 0) { /* default to locked */ @@ -525,6 +535,7 @@ static void *waitMessage_task(void *param) } needsUnlock = 1; +#endif /* Read Loop */ PRINTF("MQTT Waiting for message..."); @@ -602,10 +613,16 @@ static void *waitMessage_task(void *param) break; } - /* Keep Alive handled in ping thread */ - /* Signal keep alive thread */ + #ifdef WOLFMQTT_NO_TIME + /* Core auto keep-alive is compiled out: wake the dedicated ping + * thread to send the PINGREQ. */ wm_SemUnlock(&pingSignal); needsUnlock = 0; + #else + /* Core auto keep-alive already scheduled the PINGREQ inside + * MqttClient_WaitMessage_ex; an idle timeout just means no message + * arrived, so keep waiting. */ + #endif } else if (rc != MQTT_CODE_SUCCESS) { /* There was an error */ @@ -617,9 +634,11 @@ static void *waitMessage_task(void *param) } while (!mqtt_stop_get()); mqttCtx->return_code = rc; +#ifdef WOLFMQTT_NO_TIME if (needsUnlock) { wm_SemUnlock(&pingSignal); /* wake ping thread */ } +#endif THREAD_EXIT(0); } @@ -681,12 +700,12 @@ static void *publish_task(void *param) THREAD_EXIT(0); } -/* Dedicated keep-alive ping thread, retained to demonstrate explicit - * cross-thread ping coordination. With the core auto keep-alive compiled in - * (the default) the reader thread's MqttClient_WaitMessage_ex also sends - * PINGREQ. last_tx_time only holds off the reader's own next ping, not this - * thread, so both paths can ping and the broker sees redundant PINGREQs. - * Define WOLFMQTT_NO_TIME to rely solely on this thread. */ +#ifdef WOLFMQTT_NO_TIME +/* Dedicated keep-alive ping thread. Only compiled when the core auto + * keep-alive is disabled (WOLFMQTT_NO_TIME); otherwise + * MqttClient_WaitMessage_ex sends the PINGREQ and a second owner here would + * put redundant PINGREQs and PINGRESP waiters on the same connection. The + * reader thread signals this thread on an idle timeout. */ #ifdef USE_WINDOWS_API static DWORD WINAPI ping_task( LPVOID param ) #else @@ -732,6 +751,7 @@ static void *ping_task(void *param) THREAD_EXIT(0); } +#endif /* WOLFMQTT_NO_TIME */ static int unsubscribe_do(MQTTCtx *mqttCtx) { @@ -792,11 +812,13 @@ int multithread_test(MQTTCtx *mqttCtx) PRINTF("THREAD_CREATE failed: %d", errno); return -1; } - /* Ping */ +#ifdef WOLFMQTT_NO_TIME + /* Ping (only when core auto keep-alive is compiled out) */ if (THREAD_CREATE(&threadList[threadCount++], ping_task, mqttCtx)) { PRINTF("THREAD_CREATE failed: %d", errno); return -1; } +#endif /* Create threads that publish unique messages */ for (i = 0; i < NUM_PUB_TASKS; i++) { diff --git a/src/mqtt_client.c b/src/mqtt_client.c index e7e080fae..caa996d4b 100644 --- a/src/mqtt_client.c +++ b/src/mqtt_client.c @@ -1807,10 +1807,13 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect) * ping left mid-exchange by a prior connection so its held locks and * pending response are released before the new write starts, rather * than leaving the ping state machine to stall a later wait. Must run - * before MqttWriteStart so the send lock is free when it is taken. */ + * before MqttWriteStart so the send lock is free when it is taken. The + * cancel is best-effort: under WOLFMQTT_MULTITHREAD a failed client + * lock leaves the stale pending response to be reclaimed on the next + * acquisition, so its return is intentionally not checked. */ client->keep_alive_sec = 0; client->keep_alive_from_server = 0; - MqttClient_CancelMessage(client, + (void)MqttClient_CancelMessage(client, (MqttObject*)&client->keep_alive_ping); #endif @@ -2905,9 +2908,12 @@ int MqttClient_Disconnect_ex(MqttClient *client, MqttDisconnect *p_disconnect) /* Stop auto keep-alive for a client being torn down, and fully cancel * any ping left mid-exchange so its held locks and pending response are * released before the disconnect write starts. A bare stat reset would - * strand client->write.isActive and livelock MqttClient_Disconnect. */ + * strand client->write.isActive and livelock MqttClient_Disconnect. The + * cancel is best-effort: under WOLFMQTT_MULTITHREAD a failed client lock + * leaves the stale pending response to be reclaimed on the next + * acquisition, so its return is intentionally not checked. */ client->keep_alive_sec = 0; - MqttClient_CancelMessage(client, + (void)MqttClient_CancelMessage(client, (MqttObject*)&client->keep_alive_ping); #endif #ifdef WOLFMQTT_V5 @@ -3138,6 +3144,12 @@ static int MqttClient_KeepAlive(MqttClient *client, MqttObject* msg) return MQTT_TRACE_ERROR(MQTT_CODE_ERROR_BAD_ARG); } + /* Disabled at runtime by the application (e.g. it schedules its own ping), + * independent of the keep-alive negotiated with the broker. */ + if ((client->flags & MQTT_CLIENT_FLAG_NO_AUTO_KEEPALIVE) != 0) { + return MQTT_CODE_SUCCESS; + } + /* Disabled until a non-zero keep-alive has been negotiated in CONNECT. */ if (client->keep_alive_sec == 0) { return MQTT_CODE_SUCCESS; diff --git a/tests/test_mqtt_client.c b/tests/test_mqtt_client.c index 04aa4b175..a83b822ed 100644 --- a/tests/test_mqtt_client.c +++ b/tests/test_mqtt_client.c @@ -2042,6 +2042,17 @@ static int mock_net_read_timeout(void *context, byte* buf, int buf_len, return MQTT_CODE_ERROR_TIMEOUT; } +#ifdef WOLFMQTT_NONBLOCK +/* Read side reports would-block, so a non-blocking wait defers rather than + * completing or timing out. */ +static int mock_net_read_continue(void *context, byte* buf, int buf_len, + int timeout_ms) +{ + (void)context; (void)buf; (void)buf_len; (void)timeout_ms; + return MQTT_CODE_CONTINUE; +} +#endif + TEST(wait_message_auto_pings_on_keepalive_deadline) { int rc; @@ -2098,6 +2109,38 @@ TEST(wait_message_no_autoping_when_keepalive_zero) ASSERT_EQ(0, g_pingreq_writes); } +TEST(wait_message_no_autoping_when_flag_disables) +{ + int rc; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + + g_pingreq_writes = 0; + test_net.write = mock_net_write_count_ping; + test_net.read = mock_net_read_timeout; + + /* A non-zero keep-alive is negotiated (so the broker still enforces it) and + * the link is idle well past the deadline, but the application disabled the + * core scheduler at runtime because it sends its own PINGREQ. No auto-ping + * must be emitted, and the idle wait reports a plain timeout rather than a + * keep-alive network error. */ + (void)MqttClient_Flags(&test_client, 0, MQTT_CLIENT_FLAG_NO_AUTO_KEEPALIVE); + test_client.keep_alive_sec = 1; + test_client.last_tx_time = 0; + + rc = MqttClient_WaitMessage(&test_client, 0); + ASSERT_EQ(MQTT_CODE_ERROR_TIMEOUT, rc); + ASSERT_EQ(0, g_pingreq_writes); + + /* Clearing the flag re-enables the scheduler on the same client. */ + (void)MqttClient_Flags(&test_client, MQTT_CLIENT_FLAG_NO_AUTO_KEEPALIVE, 0); + g_pingreq_writes = 0; + rc = MqttClient_WaitMessage(&test_client, 0); + ASSERT_EQ(MQTT_CODE_ERROR_NETWORK, rc); + ASSERT_TRUE(g_pingreq_writes >= 1); +} + TEST(wait_message_auto_pings_before_full_interval_with_margin) { int rc; @@ -2146,6 +2189,59 @@ TEST(wait_message_no_autoping_when_link_recently_active) ASSERT_EQ(0, g_pingreq_writes); } +TEST(packet_write_refreshes_last_tx_time) +{ + int rc; + word32 before; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + + g_pingreq_writes = 0; + test_net.write = mock_net_write_count_ping; + test_net.read = mock_net_read_timeout; + + /* Exercise the real MqttPacket_Write idle-timer stamp instead of poking + * last_tx_time directly. With auto-ping armed, a full control-packet write + * must record the send time so later outbound traffic defers the ping. The + * PINGRESP never arrives, but the PINGREQ still writes through + * MqttPacket_Write, which stamps last_tx_time on completion. */ + test_client.keep_alive_sec = 60; + test_client.last_tx_time = 0; /* stale baseline the write must overwrite */ + + before = WOLFMQTT_GET_TIME_S(); + (void)MqttClient_Ping(&test_client); + + /* The packet really went out, and the write path refreshed the timer to + * roughly now (far above the seeded zero). */ + ASSERT_TRUE(g_pingreq_writes >= 1); + ASSERT_TRUE(test_client.last_tx_time >= before); +} + +TEST(packet_write_skips_last_tx_time_when_keepalive_zero) +{ + int rc; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + + g_pingreq_writes = 0; + test_net.write = mock_net_write_count_ping; + test_net.read = mock_net_read_timeout; + + /* Negative guard: with keep-alive disabled the write path must skip the + * clock read entirely, so a full packet write leaves last_tx_time untouched + * even though the PINGREQ itself is sent. Guards the keep_alive_sec != 0 + * condition on the stamp. */ + test_client.keep_alive_sec = 0; + test_client.last_tx_time = 0; + + (void)MqttClient_Ping(&test_client); + + ASSERT_TRUE(g_pingreq_writes >= 1); + ASSERT_EQ(0, test_client.last_tx_time); +} + TEST(wait_message_no_autoping_during_partial_read) { int rc; @@ -2232,6 +2328,41 @@ TEST(wait_message_resumes_inflight_ping) ASSERT_EQ(MQTT_CODE_ERROR_NETWORK, rc); } +#ifdef WOLFMQTT_NONBLOCK +TEST(wait_message_nonblock_returns_continue_for_deferred_ping) +{ + int rc; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + + g_pingreq_writes = 0; + test_net.write = mock_net_write_count_ping; + test_net.read = mock_net_read_continue; + + /* An auto keep-alive ping is mid-exchange (PINGREQ already sent, PINGRESP + * not yet arrived). The read side reports would-block, so under + * WOLFMQTT_NONBLOCK the resumed ping cannot complete. WaitMessage must + * return MQTT_CODE_CONTINUE to the caller rather than swallowing it (the + * "&& rc != MQTT_CODE_CONTINUE" guard is compiled out under NONBLOCK), and + * leave the ping in-flight so a later call resumes it. This is the caller- + * visible liveness handoff that the blocking build deliberately hides. */ + test_client.keep_alive_sec = 3600; + test_client.last_tx_time = WOLFMQTT_GET_TIME_S(); + test_client.keep_alive_ping.stat.write = MQTT_MSG_WAIT; + + rc = MqttClient_WaitMessage(&test_client, 0); + + ASSERT_EQ(MQTT_CODE_CONTINUE, rc); + /* Ping stays mid-exchange; no new PINGREQ was encoded. */ + ASSERT_EQ(MQTT_MSG_WAIT, test_client.keep_alive_ping.stat.write); + ASSERT_EQ(0, g_pingreq_writes); + + /* Restore so the state machine is clean for later assertions. */ + test_client.keep_alive_ping.stat.write = MQTT_MSG_BEGIN; +} +#endif /* WOLFMQTT_NONBLOCK */ + TEST(connect_arms_keepalive_and_disconnect_disarms) { int rc; @@ -2566,10 +2697,16 @@ void run_mqtt_client_tests(void) RUN_TEST(wait_message_auto_pings_on_keepalive_deadline); RUN_TEST(wait_message_auto_pings_before_full_interval_with_margin); RUN_TEST(wait_message_no_autoping_when_keepalive_zero); + RUN_TEST(wait_message_no_autoping_when_flag_disables); RUN_TEST(wait_message_no_autoping_when_link_recently_active); + RUN_TEST(packet_write_refreshes_last_tx_time); + RUN_TEST(packet_write_skips_last_tx_time_when_keepalive_zero); RUN_TEST(wait_message_no_autoping_during_partial_read); RUN_TEST(wait_message_no_autoping_during_partial_payload); RUN_TEST(wait_message_resumes_inflight_ping); +#ifdef WOLFMQTT_NONBLOCK + RUN_TEST(wait_message_nonblock_returns_continue_for_deferred_ping); +#endif RUN_TEST(connect_arms_keepalive_and_disconnect_disarms); RUN_TEST(connect_refused_leaves_keepalive_disarmed); #if defined(WOLFMQTT_V5) diff --git a/wolfmqtt/mqtt_client.h b/wolfmqtt/mqtt_client.h index 5a4bfc8a5..916dbc2ea 100644 --- a/wolfmqtt/mqtt_client.h +++ b/wolfmqtt/mqtt_client.h @@ -79,7 +79,13 @@ * uptime is used instead. Override it in user_settings.h on targets without a * suitable clock, or define WOLFMQTT_NO_TIME to compile out automatic * scheduling entirely; the explicit MqttClient_Ping / MqttClient_Ping_ex APIs - * still work in that case. */ + * still work in that case. + * + * To keep the feature compiled in but turn it off for a specific client at + * runtime - for example an application that already schedules its own PINGREQ + * and must not double-ping - set MQTT_CLIENT_FLAG_NO_AUTO_KEEPALIVE with + * MqttClient_Flags before connecting; the keep-alive negotiated with the broker + * in CONNECT is unaffected. */ #ifndef WOLFMQTT_NO_TIME #ifndef WOLFMQTT_GET_TIME_S #ifdef __ZEPHYR__ @@ -150,7 +156,12 @@ enum MqttClientFlags { MQTT_CLIENT_FLAG_IS_CONNECTED = 0x01 << 0, MQTT_CLIENT_FLAG_IS_TLS = 0x01 << 1, MQTT_CLIENT_FLAG_IS_DTLS = 0x01 << 2, - MQTT_CLIENT_FLAG_WOLFSSL_INIT = 0x01 << 3 /* this client called wolfSSL_Init */ + MQTT_CLIENT_FLAG_WOLFSSL_INIT = 0x01 << 3, /* this client called wolfSSL_Init */ + /* Disable the core automatic keep-alive PINGREQ at runtime, e.g. for an + * application that schedules its own ping, without changing the keep-alive + * negotiated with the broker. Set via MqttClient_Flags before connecting. + * Has no effect when built with WOLFMQTT_NO_TIME. */ + MQTT_CLIENT_FLAG_NO_AUTO_KEEPALIVE = 0x01 << 4 }; /*! \brief Sets flags in the MqttClient structure. To be used from the application before calling MqttClient_NetConnect. From 5a6191950bda21810d36d8a85bffe6ab1022d6f9 Mon Sep 17 00:00:00 2001 From: Eric Blankenhorn Date: Thu, 16 Jul 2026 16:17:05 -0500 Subject: [PATCH 4/6] Fixes from review --- tests/test_mqtt_client.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_mqtt_client.c b/tests/test_mqtt_client.c index a83b822ed..e64cc3b76 100644 --- a/tests/test_mqtt_client.c +++ b/tests/test_mqtt_client.c @@ -2045,7 +2045,7 @@ static int mock_net_read_timeout(void *context, byte* buf, int buf_len, #ifdef WOLFMQTT_NONBLOCK /* Read side reports would-block, so a non-blocking wait defers rather than * completing or timing out. */ -static int mock_net_read_continue(void *context, byte* buf, int buf_len, +static int mock_net_read_wouldblock(void *context, byte* buf, int buf_len, int timeout_ms) { (void)context; (void)buf; (void)buf_len; (void)timeout_ms; @@ -2338,7 +2338,7 @@ TEST(wait_message_nonblock_returns_continue_for_deferred_ping) g_pingreq_writes = 0; test_net.write = mock_net_write_count_ping; - test_net.read = mock_net_read_continue; + test_net.read = mock_net_read_wouldblock; /* An auto keep-alive ping is mid-exchange (PINGREQ already sent, PINGRESP * not yet arrived). The read side reports would-block, so under From 562c1fa4850c1ae43aa4f22e1c9db528449b992e Mon Sep 17 00:00:00 2001 From: Eric Blankenhorn Date: Fri, 17 Jul 2026 09:15:55 -0500 Subject: [PATCH 5/6] Fix from review --- examples/aws/awsiot.c | 6 ++++++ examples/azure/azureiothub.c | 12 +++++++++++ examples/firmware/fwclient.c | 6 ++++++ examples/mqttsimple/mqttsimple.c | 8 ++++++- examples/pub-sub/mqtt-sub.c | 6 ++++++ examples/wiot/wiot.c | 11 ++++++++++ src/mqtt_client.c | 28 ++++++++++++++++++++++++- tests/test_mqtt_client.c | 36 ++++++++++++++++++++++++++++++++ wolfmqtt/mqtt_client.h | 4 +++- 9 files changed, 114 insertions(+), 3 deletions(-) diff --git a/examples/aws/awsiot.c b/examples/aws/awsiot.c index 7103441b8..3782f1804 100644 --- a/examples/aws/awsiot.c +++ b/examples/aws/awsiot.c @@ -807,6 +807,7 @@ int awsiot_test(MQTTCtx *mqttCtx) else #endif if (rc == MQTT_CODE_ERROR_TIMEOUT) { + #ifdef WOLFMQTT_NO_TIME /* Keep Alive */ PRINTF("Keep-alive timeout, sending ping"); @@ -819,6 +820,11 @@ int awsiot_test(MQTTCtx *mqttCtx) MqttClient_ReturnCodeToString(rc), rc); break; } + #else + /* The core client sends keep-alive PINGREQ automatically, so + * an idle timeout just means no message arrived. */ + rc = MQTT_CODE_SUCCESS; + #endif } else if (rc != MQTT_CODE_SUCCESS) { /* There was an error */ diff --git a/examples/azure/azureiothub.c b/examples/azure/azureiothub.c index 23491cc38..7bfd95ec2 100644 --- a/examples/azure/azureiothub.c +++ b/examples/azure/azureiothub.c @@ -514,10 +514,22 @@ int azureiothub_test(MQTTCtx *mqttCtx) else #endif if (rc == MQTT_CODE_ERROR_TIMEOUT) { + #ifdef WOLFMQTT_NO_TIME /* Keep Alive */ PRINTF("Keep-alive timeout, sending ping"); mqttCtx->stat = WMQ_PING; break; + #else + /* The core client sends keep-alive PINGREQ automatically, so + * an idle timeout just means no message arrived. In test + * mode there is no inbound message and the connection is + * already established, so finish rather than wait forever + * (this example's test relied on the manual ping to end). */ + if (mqttCtx->test_mode) { + mTestDone = 1; + } + rc = MQTT_CODE_SUCCESS; + #endif } else if (rc != MQTT_CODE_SUCCESS) { /* There was an error */ diff --git a/examples/firmware/fwclient.c b/examples/firmware/fwclient.c index 3b9ad37e0..c3c69c1a5 100644 --- a/examples/firmware/fwclient.c +++ b/examples/firmware/fwclient.c @@ -424,6 +424,7 @@ int fwclient_test(MQTTCtx *mqttCtx) PRINTF("Timeout in test mode, exit early!"); mTestDone = 1; } + #ifdef WOLFMQTT_NO_TIME /* Keep Alive */ PRINTF("Keep-alive timeout, sending ping"); @@ -436,6 +437,11 @@ int fwclient_test(MQTTCtx *mqttCtx) MqttClient_ReturnCodeToString(rc), rc); break; } + #else + /* The core client sends keep-alive PINGREQ automatically, so + * an idle timeout just means no message arrived. */ + rc = MQTT_CODE_SUCCESS; + #endif } else if (rc != MQTT_CODE_SUCCESS) { /* There was an error */ diff --git a/examples/mqttsimple/mqttsimple.c b/examples/mqttsimple/mqttsimple.c index fd1bf541a..892331a11 100644 --- a/examples/mqttsimple/mqttsimple.c +++ b/examples/mqttsimple/mqttsimple.c @@ -449,12 +449,18 @@ int mqttsimple_test(void) rc = MqttClient_WaitMessage_ex(&mClient, &mqttObj, MQTT_CMD_TIMEOUT_MS); if (rc == MQTT_CODE_ERROR_TIMEOUT) { - /* send keep-alive ping */ + #ifdef WOLFMQTT_NO_TIME + /* Automatic keep-alive is compiled out, so send the ping here. */ rc = MqttClient_Ping_ex(&mClient, &mqttObj.ping); if (rc != MQTT_CODE_SUCCESS) { break; } PRINTF("MQTT Keep-Alive Ping"); + #else + /* The core client sends keep-alive PINGREQ automatically, so an + * idle timeout just means no message arrived; keep waiting. */ + rc = MQTT_CODE_SUCCESS; + #endif } else if (rc != MQTT_CODE_SUCCESS) { break; diff --git a/examples/pub-sub/mqtt-sub.c b/examples/pub-sub/mqtt-sub.c index dcffc628f..c4e749725 100644 --- a/examples/pub-sub/mqtt-sub.c +++ b/examples/pub-sub/mqtt-sub.c @@ -520,6 +520,7 @@ int sub_client(MQTTCtx *mqttCtx) } #endif else if (rc == MQTT_CODE_ERROR_TIMEOUT) { + #ifdef WOLFMQTT_NO_TIME /* Keep Alive */ if (mqttCtx->debug_on) { PRINTF("Keep-alive timeout, sending ping"); @@ -530,6 +531,11 @@ int sub_client(MQTTCtx *mqttCtx) MqttClient_ReturnCodeToString(rc), rc); break; } + #else + /* The core client sends keep-alive PINGREQ automatically, so an + * idle timeout just means no message arrived. */ + rc = MQTT_CODE_SUCCESS; + #endif } #ifdef WOLFMQTT_NONBLOCK else if (rc == MQTT_CODE_CONTINUE) { diff --git a/examples/wiot/wiot.c b/examples/wiot/wiot.c index e23272aa9..88b9c9f9f 100644 --- a/examples/wiot/wiot.c +++ b/examples/wiot/wiot.c @@ -333,6 +333,7 @@ int wiot_test(MQTTCtx *mqttCtx) else #endif if (rc == MQTT_CODE_ERROR_TIMEOUT) { + #ifdef WOLFMQTT_NO_TIME /* Keep Alive */ PRINTF("Keep-alive timeout, sending ping"); @@ -342,6 +343,16 @@ int wiot_test(MQTTCtx *mqttCtx) MqttClient_ReturnCodeToString(rc), rc); break; } + #else + /* The core client sends keep-alive PINGREQ automatically, so an + * idle timeout just means no message arrived. In test mode there is + * no inbound message and the connection is already established, so + * finish rather than wait forever. */ + if (mqttCtx->test_mode) { + mTestDone = 1; + } + rc = MQTT_CODE_SUCCESS; + #endif } else if (rc != MQTT_CODE_SUCCESS) { /* There was an error */ diff --git a/src/mqtt_client.c b/src/mqtt_client.c index caa996d4b..bfc1b45f5 100644 --- a/src/mqtt_client.c +++ b/src/mqtt_client.c @@ -3138,6 +3138,7 @@ int MqttClient_PropsFree(MqttProp *head) static int MqttClient_KeepAlive(MqttClient *client, MqttObject* msg) { int rc = MQTT_CODE_SUCCESS; + int mid_transfer; word32 now, elapsed, threshold; if (client == NULL) { @@ -3171,9 +3172,34 @@ static int MqttClient_KeepAlive(MqttClient *client, MqttObject* msg) /* Do not inject a ping while a message is mid-transfer: a partially read * fixed header (packet.stat), or a payload read still in progress on the * wait object (msg->read), which packet.stat alone does not catch once the - * header has been consumed but the read lock is still held. */ + * header has been consumed but the read is not finished. Under + * WOLFMQTT_MULTITHREAD client->packet and client->read are protected by the + * read lock, so evaluate this under lockClient (which MqttReadStart and + * MqttReadStop also hold when they set read.isActive and reset + * packet.stat). read.isActive - an in-progress read holding lockRecv - is + * tested first, so packet.stat is only read when no read is active and thus + * cannot be concurrently advanced. If lockClient cannot be taken another + * thread is busy, so skip the ping this round. */ + mid_transfer = 0; +#ifdef WOLFMQTT_MULTITHREAD + if (wm_SemLock(&client->lockClient) == 0) { + if (client->read.isActive || + client->packet.stat != MQTT_PK_BEGIN || + (msg != NULL && ((MqttMsgStat*)msg)->read != MQTT_MSG_BEGIN)) { + mid_transfer = 1; + } + wm_SemUnlock(&client->lockClient); + } + else { + mid_transfer = 1; + } +#else if (client->packet.stat != MQTT_PK_BEGIN || (msg != NULL && ((MqttMsgStat*)msg)->read != MQTT_MSG_BEGIN)) { + mid_transfer = 1; + } +#endif + if (mid_transfer) { return MQTT_CODE_SUCCESS; } diff --git a/tests/test_mqtt_client.c b/tests/test_mqtt_client.c index e64cc3b76..0d804d0d6 100644 --- a/tests/test_mqtt_client.c +++ b/tests/test_mqtt_client.c @@ -2298,6 +2298,39 @@ TEST(wait_message_no_autoping_during_partial_payload) test_client.msg.publish.stat.read = MQTT_MSG_BEGIN; } +#ifdef WOLFMQTT_MULTITHREAD +TEST(wait_message_no_autoping_when_read_active) +{ + int rc; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + + g_pingreq_writes = 0; + test_net.write = mock_net_write_count_ping; + test_net.read = mock_net_read_timeout; + + /* Deadline is well past, but a read is flagged active - under + * WOLFMQTT_MULTITHREAD another thread would hold lockRecv and be advancing + * client->packet. The mid-transfer guard, evaluated under lockClient, must + * see read.isActive and suppress the ping (short-circuiting before it reads + * packet.stat). Exercises the MULTITHREAD-only branch of the guard. */ + test_client.keep_alive_sec = 1; + test_client.last_tx_time = 0; + test_client.packet.stat = MQTT_PK_BEGIN; /* isActive alone must suppress */ + test_client.read.isActive = 1; + + rc = MqttClient_WaitMessage(&test_client, 0); + + /* No auto-ping was injected while a read was active. */ + ASSERT_EQ(0, g_pingreq_writes); + (void)rc; + + /* Restore so the state machine is clean for later assertions. */ + test_client.read.isActive = 0; +} +#endif /* WOLFMQTT_MULTITHREAD */ + TEST(wait_message_resumes_inflight_ping) { int rc; @@ -2703,6 +2736,9 @@ void run_mqtt_client_tests(void) RUN_TEST(packet_write_skips_last_tx_time_when_keepalive_zero); RUN_TEST(wait_message_no_autoping_during_partial_read); RUN_TEST(wait_message_no_autoping_during_partial_payload); +#ifdef WOLFMQTT_MULTITHREAD + RUN_TEST(wait_message_no_autoping_when_read_active); +#endif RUN_TEST(wait_message_resumes_inflight_ping); #ifdef WOLFMQTT_NONBLOCK RUN_TEST(wait_message_nonblock_returns_continue_for_deferred_ping); diff --git a/wolfmqtt/mqtt_client.h b/wolfmqtt/mqtt_client.h index 916dbc2ea..7342fdb38 100644 --- a/wolfmqtt/mqtt_client.h +++ b/wolfmqtt/mqtt_client.h @@ -280,7 +280,9 @@ typedef struct _MqttClient { MqttPing keep_alive_ping; /* dedicated object for auto keep-alive PINGREQ */ word32 last_tx_time; /* WOLFMQTT_GET_TIME_S() at last control-pkt TX */ word16 keep_alive_sec; /* negotiated keep-alive; 0 disables auto-ping */ - byte keep_alive_from_server; /* v5 Server Keep Alive applied (0 valid) */ + /* v5 Server Keep Alive applied (0 is itself a valid keep-alive). Single-bit + * flag per the wolfSSL struct guidance for booleans. */ + unsigned int keep_alive_from_server : 1; #endif } MqttClient; From 01384481e97fb1561581d4ada43c9168fe85aeec Mon Sep 17 00:00:00 2001 From: Eric Blankenhorn Date: Fri, 17 Jul 2026 11:20:49 -0500 Subject: [PATCH 6/6] Fix from review --- examples/azure/azureiothub.c | 11 +++++++---- examples/wiot/wiot.c | 9 ++++++--- src/mqtt_client.c | 33 +++++++++++++++++++++------------ 3 files changed, 34 insertions(+), 19 deletions(-) diff --git a/examples/azure/azureiothub.c b/examples/azure/azureiothub.c index 7bfd95ec2..8c875c03a 100644 --- a/examples/azure/azureiothub.c +++ b/examples/azure/azureiothub.c @@ -521,10 +521,13 @@ int azureiothub_test(MQTTCtx *mqttCtx) break; #else /* The core client sends keep-alive PINGREQ automatically, so - * an idle timeout just means no message arrived. In test - * mode there is no inbound message and the connection is - * already established, so finish rather than wait forever - * (this example's test relied on the manual ping to end). */ + * an idle timeout just means no message arrived; keep + * waiting. In test mode this example has no inbound message + * and previously terminated via the manual ping, so exit + * here instead of looping forever. This mTestDone assignment + * is deliberately not present in awsiot/fwclient/mqttclient: + * those end test mode another way (a received message, a + * pre-existing test-mode flag, or a top-of-loop break). */ if (mqttCtx->test_mode) { mTestDone = 1; } diff --git a/examples/wiot/wiot.c b/examples/wiot/wiot.c index 88b9c9f9f..bf24980ca 100644 --- a/examples/wiot/wiot.c +++ b/examples/wiot/wiot.c @@ -345,9 +345,12 @@ int wiot_test(MQTTCtx *mqttCtx) } #else /* The core client sends keep-alive PINGREQ automatically, so an - * idle timeout just means no message arrived. In test mode there is - * no inbound message and the connection is already established, so - * finish rather than wait forever. */ + * idle timeout just means no message arrived; keep waiting. In test + * mode this example has no inbound message and previously terminated + * via the manual ping, so exit here instead of looping forever. This + * mTestDone assignment is deliberately not present in awsiot/fwclient/ + * mqttclient: those end test mode another way (a received message, a + * pre-existing test-mode flag, or a top-of-loop break). */ if (mqttCtx->test_mode) { mTestDone = 1; } diff --git a/src/mqtt_client.c b/src/mqtt_client.c index bfc1b45f5..08066cd9d 100644 --- a/src/mqtt_client.c +++ b/src/mqtt_client.c @@ -1807,14 +1807,18 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect) * ping left mid-exchange by a prior connection so its held locks and * pending response are released before the new write starts, rather * than leaving the ping state machine to stall a later wait. Must run - * before MqttWriteStart so the send lock is free when it is taken. The - * cancel is best-effort: under WOLFMQTT_MULTITHREAD a failed client - * lock leaves the stale pending response to be reclaimed on the next - * acquisition, so its return is intentionally not checked. */ + * before MqttWriteStart so the send lock is free when it is taken. + * Propagate a cancel failure - only reachable if wm_SemLock itself + * errors (e.g. on ThreadX), since it otherwise blocks until acquired - + * rather than starting the write with locks the abandoned ping may + * still hold. */ client->keep_alive_sec = 0; client->keep_alive_from_server = 0; - (void)MqttClient_CancelMessage(client, + rc = MqttClient_CancelMessage(client, (MqttObject*)&client->keep_alive_ping); + if (rc != MQTT_CODE_SUCCESS) { + return rc; + } #endif /* Flag write active / lock mutex */ @@ -2908,13 +2912,17 @@ int MqttClient_Disconnect_ex(MqttClient *client, MqttDisconnect *p_disconnect) /* Stop auto keep-alive for a client being torn down, and fully cancel * any ping left mid-exchange so its held locks and pending response are * released before the disconnect write starts. A bare stat reset would - * strand client->write.isActive and livelock MqttClient_Disconnect. The - * cancel is best-effort: under WOLFMQTT_MULTITHREAD a failed client lock - * leaves the stale pending response to be reclaimed on the next - * acquisition, so its return is intentionally not checked. */ + * strand client->write.isActive and livelock MqttClient_Disconnect. + * Propagate a cancel failure - only reachable if wm_SemLock itself + * errors (e.g. on ThreadX), since it otherwise blocks until acquired - + * rather than starting the write with locks the abandoned ping may + * still hold. */ client->keep_alive_sec = 0; - (void)MqttClient_CancelMessage(client, + rc = MqttClient_CancelMessage(client, (MqttObject*)&client->keep_alive_ping); + if (rc != MQTT_CODE_SUCCESS) { + return rc; + } #endif #ifdef WOLFMQTT_V5 /* Use specified protocol version if set */ @@ -3178,8 +3186,9 @@ static int MqttClient_KeepAlive(MqttClient *client, MqttObject* msg) * MqttReadStop also hold when they set read.isActive and reset * packet.stat). read.isActive - an in-progress read holding lockRecv - is * tested first, so packet.stat is only read when no read is active and thus - * cannot be concurrently advanced. If lockClient cannot be taken another - * thread is busy, so skip the ping this round. */ + * cannot be concurrently advanced. wm_SemLock blocks until lockClient is + * available and returns an error only on a system fault (e.g. ThreadX), in + * which case the ping is skipped this round. */ mid_transfer = 0; #ifdef WOLFMQTT_MULTITHREAD if (wm_SemLock(&client->lockClient) == 0) {