From e70350640760d3dba82b26db4622ad092325c025 Mon Sep 17 00:00:00 2001 From: Kareem Date: Thu, 24 Sep 2026 16:08:33 -0700 Subject: [PATCH 01/11] wolfMQTT broker: stop the outbound drain when the Session hand-off takes its queue. Thanks to Gwanhyun Lee for the report. --- src/mqtt_broker.c | 22 +++++ tests/test_broker_connect.c | 157 ++++++++++++++++++++++++++++++++++++ 2 files changed, 179 insertions(+) diff --git a/src/mqtt_broker.c b/src/mqtt_broker.c index c86adc03d..abef5b410 100644 --- a/src/mqtt_broker.c +++ b/src/mqtt_broker.c @@ -2247,6 +2247,22 @@ static BrokerOutPub* BrokerClient_OutQRemove(BrokerClient* bc, return next; } +/* A write can service this same client's close callback before it returns - + * the WebSocket transport runs lws_service inline - and that hands the whole + * outbound queue to the Session carrier. Entries the drain is walking were + * reached from bc->out_q_head, so an empty head while one is still held means + * the carrier owns them now: they must not be unlinked, freed, or re-linked + * into bc. Returns non-zero when the drain has to stop for that reason. */ +static int BrokerClient_OutQueueMoved(BrokerClient* bc, int enc_len) +{ + if (bc->out_q_head != NULL) { + return 0; + } + bc->out_q_pending_len = 0; + BROKER_FORCE_ZERO(bc->tx_buf, enc_len); + return 1; +} + /* Send as many QUEUED entries from out_q as the inflight cap allows. * * Ordering: walks from out_q_head, never reorders. Already-sent entries @@ -2311,6 +2327,9 @@ static int BrokerClient_DrainOutQueue(BrokerClient* bc) return MQTT_CODE_ERROR_SYSTEM; } wr_rc = MqttPacket_Write(&bc->client, bc->tx_buf, rel_rc); + if (BrokerClient_OutQueueMoved(bc, rel_rc)) { + return sent; + } if (wr_rc == MQTT_CODE_CONTINUE) { bc->out_q_pending_len = rel_rc; return wr_rc; @@ -2409,6 +2428,9 @@ static int BrokerClient_DrainOutQueue(BrokerClient* bc) { int wr_rc; wr_rc = MqttPacket_Write(&bc->client, bc->tx_buf, enc_rc); + if (BrokerClient_OutQueueMoved(bc, enc_rc)) { + return sent; + } /* Scrub the forwarded PUBLISH (which may carry a replayed will or * an application payload) once the buffer is idle. Skip only the * MQTT_CODE_CONTINUE case, where a non-blocking or TLS-async send diff --git a/tests/test_broker_connect.c b/tests/test_broker_connect.c index 384d99bce..78ca8a219 100644 --- a/tests/test_broker_connect.c +++ b/tests/test_broker_connect.c @@ -72,6 +72,8 @@ static byte g_capture_freed[64]; static size_t g_capture_freed_len; static int g_capture_free_seen; static size_t g_alloc_fail_size; /* fail the next malloc of exactly this size */ +static void* g_free_watch_ptr; +static int g_free_watch_count; void* wolfmqtt_test_broker_malloc(size_t size) { @@ -107,6 +109,9 @@ void wolfmqtt_test_broker_free(void* ptr) { if (ptr != NULL) { g_alloc_free_count++; + if (ptr == g_free_watch_ptr) { + g_free_watch_count++; + } if (ptr == g_capture_alloc_ptr) { g_capture_freed_len = g_capture_alloc_size; XMEMCPY(g_capture_freed, ptr, g_capture_freed_len); @@ -139,6 +144,21 @@ int wolfmqtt_test_broker_fsync(int fd) } #endif +#ifndef WOLFMQTT_STATIC_MEMORY +/* Count frees of one specific allocation, so a test can assert on the fate of + * a single object rather than on a total that unrelated activity moves. */ +static void broker_test_watch_free(void* ptr) +{ + g_free_watch_ptr = ptr; + g_free_watch_count = 0; +} + +static int broker_test_watched_frees(void) +{ + return g_free_watch_count; +} +#endif + #ifndef WOLFMQTT_STATIC_MEMORY static void broker_test_fail_alloc_after(int successful_allocations) { @@ -2840,6 +2860,142 @@ TEST(fanout_survives_reentrant_sub_free) MqttBroker_Free(&broker); } +/* The outbound drain holds a queue entry across the write that sends it. That + * write can service this client's own close callback before it returns - the + * WebSocket transport runs lws_service inline - and the callback hands the + * whole queue to the carrier that keeps the Session alive while the client is + * gone. The entry the drain is holding belongs to that carrier from then on. + * + * BrokerSubs_OrphanClient performs the hand-off through file-local helpers, so + * the hook below stages the same ownership transfer directly: the queue moves + * to a carrier and the client's own queue fields are cleared, which is what + * stops its teardown from freeing the entries a second time. */ +static MqttBroker* g_handoff_broker; +static BrokerClient* g_handoff_client; +static void handoff_out_queue_mid_write(void) +{ + MqttBroker* broker = g_handoff_broker; + BrokerClient* bc = g_handoff_client; + BrokerOrphanSession* o; + size_t id_len; + + if (broker == NULL || bc == NULL || bc->client_id == NULL || + bc->out_q_head == NULL) { + return; + } + o = (BrokerOrphanSession*)WOLFMQTT_MALLOC(sizeof(*o)); + if (o == NULL) { + return; + } + XMEMSET(o, 0, sizeof(*o)); + id_len = XSTRLEN(bc->client_id); + o->client_id = (char*)WOLFMQTT_MALLOC(id_len + 1); + if (o->client_id == NULL) { + WOLFMQTT_FREE(o); + return; + } + XMEMCPY(o->client_id, bc->client_id, id_len + 1); + o->protocol_level = bc->protocol_level; + o->session_expiry_sec = bc->session_expiry_sec; + o->orphan_since = wolfmqtt_test_broker_time_s(); + + /* Watch the entry the drain is holding, so the test can tell whether the + * drain went on to free memory the carrier owns. */ + broker_test_watch_free(bc->out_q_head); + + o->out_q_head = bc->out_q_head; + o->out_q_tail = bc->out_q_tail; + o->out_q_count = bc->out_q_count; + o->out_q_inflight = bc->out_q_inflight; + bc->out_q_head = NULL; + bc->out_q_tail = NULL; + bc->out_q_count = 0; + bc->out_q_inflight = 0; + + o->next = broker->orphan_sessions; + broker->orphan_sessions = o; + broker->orphan_session_count++; +} + +TEST(drain_stops_when_session_handoff_takes_queue) +{ + MqttBroker broker; + MqttBrokerNet net; + BrokerClient* sub_bc; + BrokerOrphanSession* o; + int i; + /* Publisher "P". */ + static const byte connect_pub[] = { + 0x10, 0x0D, 0x00, 0x04, 'M', 'Q', 'T', 'T', 0x04, 0x02, 0x00, 0x3C, + 0x00, 0x01, 'P' + }; + /* Subscriber "S" with CleanSession=0, so the Session it owns is the kind + * that survives a close and has a carrier to move the queue into. */ + static const byte connect_sub[] = { + 0x10, 0x0D, 0x00, 0x04, 'M', 'Q', 'T', 'T', 0x04, 0x00, 0x00, 0x3C, + 0x00, 0x01, 'S' + }; + /* SUBSCRIBE packet_id=1, filter "x", granted QoS 0: the delivery the + * drain completes and then unlinks. */ + static const byte subscribe_x[] = { + 0x82, 0x06, 0x00, 0x01, 0x00, 0x01, 'x', 0x00 + }; + /* QoS 0 PUBLISH, topic "x", payload "ABC". */ + static const byte publish_x[] = { + 0x30, 0x06, 0x00, 0x01, 'x', 'A', 'B', 'C' + }; + + install_mock_net(&net); + XMEMSET(&broker, 0, sizeof(broker)); + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttBroker_Init(&broker, &net)); + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttBroker_Start(&broker)); + + reset_mock_clients(2); + mock_client_input_append(0, connect_pub, sizeof(connect_pub)); + mock_client_input_append(1, connect_sub, sizeof(connect_sub)); + mock_client_input_append(1, subscribe_x, sizeof(subscribe_x)); + for (i = 0; i < 16; i++) { + (void)MqttBroker_Step(&broker); + } + sub_bc = find_broker_client(&broker, "S"); + ASSERT_NOT_NULL(sub_bc); + ASSERT_NULL(sub_bc->out_q_head); + + broker_test_watch_free(NULL); + g_handoff_broker = &broker; + g_handoff_client = sub_bc; + g_write_publish_hook = handoff_out_queue_mid_write; + + mock_client_input_append(0, publish_x, sizeof(publish_x)); + (void)MqttBroker_Step(&broker); + + /* The hand-off really did run from inside the write. */ + ASSERT_NULL(g_write_publish_hook); + + /* The carrier holds the queue it took. */ + o = broker.orphan_sessions; + ASSERT_NOT_NULL(o); + ASSERT_EQ(1, broker.orphan_session_count); + ASSERT_EQ(1, o->out_q_count); + ASSERT_NOT_NULL(o->out_q_head); + + /* The drain must not free an entry it no longer owns. */ + ASSERT_EQ(0, broker_test_watched_frees()); + + /* And it must leave no trace of that entry on the client. */ + ASSERT_EQ(0, sub_bc->out_q_count); + ASSERT_EQ(0, sub_bc->out_q_inflight); + ASSERT_NULL(sub_bc->out_q_head); + ASSERT_NULL(sub_bc->out_q_tail); + ASSERT_EQ(0, sub_bc->out_q_pending_len); + + g_handoff_broker = NULL; + g_handoff_client = NULL; + broker_test_watch_free(NULL); + MqttBroker_Stop(&broker); + MqttBroker_Free(&broker); +} + #ifdef WOLFMQTT_BROKER_WILL /* Same hazard on the Will fan-out, which runs its own subscription walk. The * publisher's socket drops, the broker fans its Will out to three subscribers, @@ -16635,6 +16791,7 @@ int main(int argc, char** argv) #ifdef WOLFMQTT_V5 RUN_TEST(publish_fanout_alloc_failure_v5_qos2_pubrec_failure); #endif + RUN_TEST(drain_stops_when_session_handoff_takes_queue); #ifdef WOLFMQTT_NONBLOCK RUN_TEST(outbound_queue_short_write_resumes_on_next_step); #ifdef WOLFMQTT_BROKER_RETAINED From 3524e1947857308b4beaeaf87c56a64e5ac454ed Mon Sep 17 00:00:00 2001 From: Kareem Date: Thu, 24 Sep 2026 16:10:05 -0700 Subject: [PATCH 02/11] wolfMQTT broker: claim the Will before publishing it. Thanks to Gwanhyun Lee for the report. --- src/mqtt_broker.c | 7 +++ tests/test_broker_connect.c | 112 ++++++++++++++++++++++++++++++++++++ 2 files changed, 119 insertions(+) diff --git a/src/mqtt_broker.c b/src/mqtt_broker.c index abef5b410..742559dad 100644 --- a/src/mqtt_broker.c +++ b/src/mqtt_broker.c @@ -6495,6 +6495,13 @@ static void BrokerClient_PublishWill(MqttBroker* broker, BrokerClient* bc) (int)bc->sock, BrokerLog_Sanitize(bc->will_topic), (unsigned)bc->will_payload_len); + /* Claim the Will before fan-out. A fan-out write can service this client's + * close callback inline, which re-enters here; the claim makes that nested + * call return at the has_will check above, so this frame stays the single + * owner of the topic and payload it lends to the encoder below and the + * Will is published once. */ + bc->has_will = 0; + BrokerClient_PublishWillImmediate(broker, bc->will_topic, bc->will_payload, bc->will_payload_len, bc->will_qos, bc->will_retain diff --git a/tests/test_broker_connect.c b/tests/test_broker_connect.c index 78ca8a219..21e17c397 100644 --- a/tests/test_broker_connect.c +++ b/tests/test_broker_connect.c @@ -3070,7 +3070,118 @@ TEST(will_fanout_survives_reentrant_sub_free) MqttBroker_Stop(&broker); MqttBroker_Free(&broker); } + +/* A Will is published by fanning it out to matching subscribers, and only + * cleared once that returns. A write inside the fan-out can service the + * owner's own close callback - the WebSocket transport runs lws_service + * inline - and that callback publishes the Will of the client that just + * closed. It must find nothing left to publish: the fan-out below is still + * lending the topic and payload to the encoder, and a nested publish would + * both deliver the Will a second time and free those buffers underneath it. + * + * The hook samples the ownership flag the nested path tests, from inside the + * first Will delivery. */ +static BrokerClient* g_will_owner; +static int g_will_claimed_during_fanout; +static int g_will_topic_live_during_fanout; +static void sample_will_state_mid_write(void) +{ + if (g_will_owner == NULL) { + return; + } + g_will_claimed_during_fanout = !g_will_owner->has_will; + g_will_topic_live_during_fanout = (g_will_owner->will_topic != NULL); +} + +TEST(takeover_will_claimed_before_fanout) +{ + MqttBroker broker; + MqttBrokerNet net; + int i; + /* Subscribers "A" and "B", CleanSession=1. */ + static const byte connect_a[] = { + 0x10, 0x0D, 0x00, 0x04, 'M', 'Q', 'T', 'T', 0x04, 0x02, 0x00, 0x3C, + 0x00, 0x01, 'A' + }; + static const byte connect_b[] = { + 0x10, 0x0D, 0x00, 0x04, 'M', 'Q', 'T', 'T', 0x04, 0x02, 0x00, 0x3C, + 0x00, 0x01, 'B' + }; + /* SUBSCRIBE packet_id=1, filter "w", QoS 0. */ + static const byte subscribe_w[] = { + 0x82, 0x06, 0x00, 0x01, 0x00, 0x01, 'w', 0x00 + }; + /* v3.1.1 CONNECT for "W" carrying a Will: flags 0x06 = CleanSession + + * Will Flag, Will Topic "w", Will Payload "bye". Remaining Length 21. */ + static const byte connect_w_will[] = { + 0x10, 0x15, 0x00, 0x04, 'M', 'Q', 'T', 'T', 0x04, 0x06, 0x00, 0x3C, + 0x00, 0x01, 'W', + 0x00, 0x01, 'w', + 0x00, 0x03, 'b', 'y', 'e' + }; + /* Second CONNECT with the same ClientId takes the Session over, which + * publishes the old connection's Will. */ + static const byte connect_w_dup[] = { + 0x10, 0x0D, 0x00, 0x04, 'M', 'Q', 'T', 'T', 0x04, 0x02, 0x00, 0x3C, + 0x00, 0x01, 'W' + }; + + install_mock_net(&net); + XMEMSET(&broker, 0, sizeof(broker)); + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttBroker_Init(&broker, &net)); + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttBroker_Start(&broker)); + + reset_mock_clients(3); + mock_client_input_append(0, connect_a, sizeof(connect_a)); + mock_client_input_append(0, subscribe_w, sizeof(subscribe_w)); + mock_client_input_append(1, connect_b, sizeof(connect_b)); + mock_client_input_append(1, subscribe_w, sizeof(subscribe_w)); + mock_client_input_append(2, connect_w_will, sizeof(connect_w_will)); + for (i = 0; i < 24; i++) { + (void)MqttBroker_Step(&broker); + } + + g_will_owner = find_broker_client(&broker, "W"); + ASSERT_NOT_NULL(g_will_owner); + ASSERT_EQ(1, (int)g_will_owner->has_will); + ASSERT_NOT_NULL(g_will_owner->will_topic); + /* Nothing has been published yet, so the hook below cannot be consumed by + * an earlier delivery. */ + ASSERT_EQ(0, count_packets_of_type(g_clients[0].out_buf, + g_clients[0].out_len, MQTT_PACKET_TYPE_PUBLISH)); + + g_will_claimed_during_fanout = -1; + g_will_topic_live_during_fanout = -1; + g_write_publish_hook = sample_will_state_mid_write; + + mock_client_input_append(3, connect_w_dup, sizeof(connect_w_dup)); + g_clients_active = 4; + for (i = 0; i < 24; i++) { + (void)MqttBroker_Step(&broker); + } + + /* The Will really was fanned out, so the sample is meaningful. */ + ASSERT_NULL(g_write_publish_hook); + + /* Claimed before the first write: a close callback delivered during the + * fan-out returns without republishing or freeing anything. */ + ASSERT_EQ(1, g_will_claimed_during_fanout); + /* And the buffers the fan-out is lending out are still owned, so the + * encoder is reading live memory. */ + ASSERT_EQ(1, g_will_topic_live_during_fanout); + + /* Exactly one copy reaches each subscriber [MQTT-3.1.2-8]. */ + ASSERT_EQ(1, count_packets_of_type(g_clients[0].out_buf, + g_clients[0].out_len, MQTT_PACKET_TYPE_PUBLISH)); + ASSERT_EQ(1, count_packets_of_type(g_clients[1].out_buf, + g_clients[1].out_len, MQTT_PACKET_TYPE_PUBLISH)); + + g_will_owner = NULL; + MqttBroker_Stop(&broker); + MqttBroker_Free(&broker); +} #endif /* WOLFMQTT_BROKER_WILL */ + #endif /* !WOLFMQTT_STATIC_MEMORY */ TEST(qos2_duplicate_publish_dedup) @@ -16760,6 +16871,7 @@ int main(int argc, char** argv) RUN_TEST(fanout_survives_reentrant_sub_free); #ifdef WOLFMQTT_BROKER_WILL RUN_TEST(will_fanout_survives_reentrant_sub_free); + RUN_TEST(takeover_will_claimed_before_fanout); #endif #endif RUN_TEST(qos2_duplicate_publish_dedup); From 325b00bfdd0b1cf23a9a2fe00b9ebd72571ed4f2 Mon Sep 17 00:00:00 2001 From: Kareem Date: Thu, 24 Sep 2026 16:26:18 -0700 Subject: [PATCH 03/11] wolfMQTT client: take the client lock over the Session replay pool in MqttClient_Connect. Thanks to Gwanhyun Lee for the report. --- src/mqtt_client.c | 75 +++++++++++++++++++++++++----- tests/include.am | 3 +- tests/test_mqtt_client.c | 98 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 163 insertions(+), 13 deletions(-) diff --git a/src/mqtt_client.c b/src/mqtt_client.c index 711884ad3..646a5688f 100644 --- a/src/mqtt_client.c +++ b/src/mqtt_client.c @@ -890,6 +890,24 @@ static int MqttClient_ConnectIsPersistent(const MqttConnect* mc_connect) return (mc_connect->clean_session == 0) ? 1 : 0; } +/* Unlike the other wrappers this one reports a lock failure. Skipping the + * reset would leave the previous Session's messages retained, and a later + * reconnect would replay them into a Session they do not belong to. */ +static int MqttClient_Replay_ResetSafe(MqttClient* client) +{ +#ifdef WOLFMQTT_MULTITHREAD + int rc = wm_SemLock(&client->lockClient); + if (rc != MQTT_CODE_SUCCESS) { + return rc; + } +#endif + MqttClient_Replay_Reset(client); +#ifdef WOLFMQTT_MULTITHREAD + wm_SemUnlock(&client->lockClient); +#endif + return MQTT_CODE_SUCCESS; +} + static void MqttClient_Replay_RemoveSafe(MqttClient* client, word16 packet_id) { #ifdef WOLFMQTT_MULTITHREAD @@ -955,8 +973,12 @@ static int MqttClient_SendIds_Find(const MqttClient* client, word16 packet_id) * MQTT_CODE_SUCCESS when it was free (or when isRetransmit says this is a * re-send of the same Control Packet, which [MQTT-2.3.1-3] requires to keep * its original identifier), and MQTT_CODE_ERROR_PACKET_ID when it is still - * awaiting its acknowledgement. */ -static int MqttClient_SendIdReserve(MqttClient* client, word16 packet_id, + * awaiting its acknowledgement. + * + * The _Locked form is for a caller that already holds client->lockClient - the + * Session resume walks the replay pool under it - since the semaphore is not + * recursive. */ +static int MqttClient_SendIdReserve_Locked(MqttClient* client, word16 packet_id, void* owner, int isRetransmit, MqttPacketType ack_type) { int rc = MQTT_CODE_SUCCESS; @@ -965,12 +987,6 @@ static int MqttClient_SendIdReserve(MqttClient* client, word16 packet_id, if (packet_id == 0) { return MQTT_CODE_SUCCESS; /* nothing to track */ } -#ifdef WOLFMQTT_MULTITHREAD - rc = wm_SemLock(&client->lockClient); - if (rc != MQTT_CODE_SUCCESS) { - return rc; - } -#endif i = MqttClient_SendIds_Find(client, packet_id); if (i >= 0) { /* Only a re-send of the same Control Packet may keep an identifier @@ -1000,6 +1016,25 @@ static int MqttClient_SendIdReserve(MqttClient* client, word16 packet_id, * packets in flight, so the check is best effort past that point - * raise MQTT_MAX_SEND_INFLIGHT to widen the window. */ } + return rc; +} + +static int MqttClient_SendIdReserve(MqttClient* client, word16 packet_id, + void* owner, int isRetransmit, MqttPacketType ack_type) +{ + int rc; + + if (packet_id == 0) { + return MQTT_CODE_SUCCESS; /* nothing to track */ + } +#ifdef WOLFMQTT_MULTITHREAD + rc = wm_SemLock(&client->lockClient); + if (rc != MQTT_CODE_SUCCESS) { + return rc; + } +#endif + rc = MqttClient_SendIdReserve_Locked(client, packet_id, owner, + isRetransmit, ack_type); #ifdef WOLFMQTT_MULTITHREAD wm_SemUnlock(&client->lockClient); #endif @@ -3987,8 +4022,14 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect) * into it would inject one Session's messages into another. */ if (!(mc_connect->ack.flags & MQTT_CONNECT_ACK_FLAG_SESSION_PRESENT) || !session_id_matched) { - /* A fresh Session starts with no outbound state to re-send. */ - MqttClient_Replay_Reset(client); + /* A fresh Session starts with no outbound state to re-send. + * Under the client lock: a send thread may be replacing a slot's + * topic and payload through MqttClient_Replay_AddSafe at the same + * moment, and both sides free what they find there. */ + rc = MqttClient_Replay_ResetSafe(client); + if (rc != MQTT_CODE_SUCCESS) { + return rc; + } } else { int i; @@ -3996,7 +4037,14 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect) /* The server resumed the Session, so these messages are still in * flight as far as it is concerned: keep their Packet Identifiers * reserved (MqttClient_Connect cleared the table above) and - * re-send them [MQTT-4.4.0-1]. */ + * re-send them [MQTT-4.4.0-1]. Walked under the client lock for + * the same reason as the reset above. */ +#ifdef WOLFMQTT_MULTITHREAD + rc = wm_SemLock(&client->lockClient); + if (rc != MQTT_CODE_SUCCESS) { + return rc; + } +#endif for (i = 0; i < MQTT_MAX_REPLAY_MSGS; i++) { if (client->replay[i].packet_id == 0) { continue; @@ -4010,7 +4058,7 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect) MqttClient_Replay_FreeSlot(&client->replay[i]); continue; } - (void)MqttClient_SendIdReserve(client, + (void)MqttClient_SendIdReserve_Locked(client, client->replay[i].packet_id, &client->replay[i], 1, client->replay[i].pubrelSent ? MQTT_PACKET_TYPE_PUBLISH_COMP : @@ -4018,6 +4066,9 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect) MQTT_PACKET_TYPE_PUBLISH_COMP : MQTT_PACKET_TYPE_PUBLISH_ACK)); } +#ifdef WOLFMQTT_MULTITHREAD + wm_SemUnlock(&client->lockClient); +#endif client->replayIdx = 0; mc_connect->stat.write = MQTT_MSG_PAYLOAD; rc = MqttClient_ReplaySession(client, mc_connect); diff --git a/tests/include.am b/tests/include.am index f2cf26a93..56700dff3 100644 --- a/tests/include.am +++ b/tests/include.am @@ -14,7 +14,8 @@ tests_unit_tests_SOURCES = \ src/mqtt_socket.c \ src/mqtt_sn_client.c \ src/mqtt_sn_packet.c -tests_unit_tests_CPPFLAGS = -I$(top_srcdir) $(AM_CPPFLAGS) +tests_unit_tests_CPPFLAGS = -I$(top_srcdir) \ + -include $(top_srcdir)/tests/test_broker_alloc.h $(AM_CPPFLAGS) tests_unit_tests_LDADD = $(PTHREAD_LIBS) # Exercise peer-name verification with a trusted local certificate and an diff --git a/tests/test_mqtt_client.c b/tests/test_mqtt_client.c index 1aa8af144..aa5b1f71d 100644 --- a/tests/test_mqtt_client.c +++ b/tests/test_mqtt_client.c @@ -48,6 +48,47 @@ static MqttNet test_net; static byte test_tx_buf[TEST_TX_BUF_SIZE]; static byte test_rx_buf[TEST_RX_BUF_SIZE]; +/* The suite routes the library's allocator through here (see + * tests/test_client_alloc.h) so a test can watch one specific allocation and + * record the state the library was in when it released it. Everything else is + * a straight pass-through. */ +static void* g_free_watch_ptr; +static int g_free_watch_count; +static int g_free_watch_lock_count; + +void* wolfmqtt_test_client_malloc(size_t size) +{ + return malloc(size); +} + +/* How many times the client lock is held right now, or -1 where the build + * keeps no count to read. */ +static int test_client_lock_depth(void) +{ +#if defined(WOLFMQTT_MULTITHREAD) && defined(WOLFMQTT_POSIX_SEMAPHORES) && \ + !defined(WOLFMQTT_NO_COND_SIGNAL) + return test_client.lockClient.lockCount; +#else + return -1; +#endif +} + +void wolfmqtt_test_client_free(void* ptr) +{ + if (ptr != NULL && ptr == g_free_watch_ptr) { + g_free_watch_count++; + g_free_watch_lock_count = test_client_lock_depth(); + } + free(ptr); +} + +UT_MAYBE_UNUSED static void test_client_watch_free(void* ptr) +{ + g_free_watch_ptr = ptr; + g_free_watch_count = 0; + g_free_watch_lock_count = -1; +} + /* Mock network callbacks - just return errors since we're not actually * connecting to anything */ static int mock_net_connect(void *context, const char* host, word16 port, @@ -3969,6 +4010,59 @@ TEST(reconnect_without_session_present_replays_nothing) ASSERT_EQ(1, g_frames_written); /* the CONNECT only */ } +#if defined(WOLFMQTT_MULTITHREAD) && defined(WOLFMQTT_POSIX_SEMAPHORES) && \ + !defined(WOLFMQTT_NO_COND_SIGNAL) +/* Discarding the retained copies releases the topic and payload of every + * slot. A send thread reaches those same slots through the replay store's + * lock-taking wrapper while the connection is being re-established - the + * client admits send calls once CONNECT has reached the transport - and both + * sides free what they find there. So the discard has to hold the client lock + * as well, or the two can free the same buffer or leave one pointing at + * memory the other released. + * + * The POSIX lock keeps its own depth, and the allocator hook samples it at the + * moment the slot's topic is released, which is the access that has to be + * covered. */ +TEST(fresh_session_reset_frees_replay_under_client_lock) +{ + int rc; + MqttConnect connect; + MqttPublish publish; + static byte payload[] = "hello"; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); +#ifdef WOLFMQTT_V5 + test_client.protocol_level = MQTT_CONNECT_PROTOCOL_LEVEL_4; +#endif + ASSERT_EQ(MQTT_CODE_SUCCESS, run_initial_connect(&connect)); + + init_qos_publish(&publish, MQTT_QOS_1, 0x1234, "sensor/temp", + payload, (word32)(sizeof(payload) - 1)); + rc = run_publish_unacked(&publish); + ASSERT_EQ(MQTT_CODE_ERROR_NETWORK, rc); + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttClient_NetDisconnect(&test_client)); + + /* The unacknowledged PUBLISH is retained, so the discard has a real + * allocation to release. */ + ASSERT_EQ(0x1234, (int)test_client.replay[0].packet_id); + ASSERT_NOT_NULL(test_client.replay[0].topic); + test_client_watch_free(test_client.replay[0].topic); + + /* Session Present = 0 is a different Session, so nothing is carried. */ + rc = run_reconnect(&connect, 0); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + ASSERT_EQ(0, (int)test_client.replay[0].packet_id); + + /* It really was released on this path, exactly once. */ + ASSERT_EQ(1, g_free_watch_count); + /* And under the client lock. */ + ASSERT_EQ(1, g_free_watch_lock_count); + + test_client_watch_free(NULL); +} +#endif + /* A completed exchange leaves Session state, so an acknowledged PUBLISH is * not replayed [MQTT-4.4.0-1] covers unacknowledged messages only. */ TEST(reconnect_does_not_replay_acked_publish) @@ -9868,6 +9962,10 @@ void run_mqtt_client_tests(void) RUN_TEST(publish_allowed_past_replay_pool_when_connack_ends_session); #endif RUN_TEST(reconnect_without_session_present_replays_nothing); +#if defined(WOLFMQTT_MULTITHREAD) && defined(WOLFMQTT_POSIX_SEMAPHORES) && \ + !defined(WOLFMQTT_NO_COND_SIGNAL) + RUN_TEST(fresh_session_reset_frees_replay_under_client_lock); +#endif RUN_TEST(reconnect_does_not_replay_acked_publish); #if WOLFMQTT_MAX_QOS >= 2 RUN_TEST(reconnect_replays_unacked_pubrel); From c8063860db8c65008af87de85058c3884e04a586 Mon Sep 17 00:00:00 2001 From: Kareem Date: Fri, 2 Oct 2026 12:56:54 -0700 Subject: [PATCH 04/11] Address code review feedback and minimize comments. --- src/mqtt_broker.c | 86 +++++++++++++++++---- src/mqtt_client.c | 37 +++++---- tests/test_broker_connect.c | 144 ++++++++++++++++++++++++++---------- tests/test_mqtt_client.c | 122 ++++++++++++++++++++++++------ 4 files changed, 298 insertions(+), 91 deletions(-) diff --git a/src/mqtt_broker.c b/src/mqtt_broker.c index 742559dad..0d1699da4 100644 --- a/src/mqtt_broker.c +++ b/src/mqtt_broker.c @@ -2247,17 +2247,30 @@ static BrokerOutPub* BrokerClient_OutQRemove(BrokerClient* bc, return next; } -/* A write can service this same client's close callback before it returns - - * the WebSocket transport runs lws_service inline - and that hands the whole - * outbound queue to the Session carrier. Entries the drain is walking were - * reached from bc->out_q_head, so an empty head while one is still held means - * the carrier owns them now: they must not be unlinked, freed, or re-linked - * into bc. Returns non-zero when the drain has to stop for that reason. */ -static int BrokerClient_OutQueueMoved(BrokerClient* bc, int enc_len) +#ifndef WOLFMQTT_STATIC_MEMORY +static void BrokerOrphan_ReconcileSent(MqttBroker* broker, + const char* client_id, BrokerOutPub* sent); +#endif + +/* A write can service this client's close callback before returning - the + * WebSocket transport runs lws_service inline - which hands the queue to the + * Session carrier. The walked entries came from bc->out_q_head, so an empty + * head while one is still held means the carrier owns them now. Returns + * non-zero when the drain must stop and leave them to it. */ +static int BrokerClient_OutQueueMoved(BrokerClient* bc, BrokerOutPub* sent, + int completed, int enc_len) { if (bc->out_q_head != NULL) { return 0; } +#ifndef WOLFMQTT_STATIC_MEMORY + if (completed) { + BrokerOrphan_ReconcileSent(bc->broker, bc->client_id, sent); + } +#else + (void)sent; + (void)completed; +#endif bc->out_q_pending_len = 0; BROKER_FORCE_ZERO(bc->tx_buf, enc_len); return 1; @@ -2327,7 +2340,8 @@ static int BrokerClient_DrainOutQueue(BrokerClient* bc) return MQTT_CODE_ERROR_SYSTEM; } wr_rc = MqttPacket_Write(&bc->client, bc->tx_buf, rel_rc); - if (BrokerClient_OutQueueMoved(bc, rel_rc)) { + if (BrokerClient_OutQueueMoved(bc, cur, + wr_rc == rel_rc, rel_rc)) { return sent; } if (wr_rc == MQTT_CODE_CONTINUE) { @@ -2428,7 +2442,8 @@ static int BrokerClient_DrainOutQueue(BrokerClient* bc) { int wr_rc; wr_rc = MqttPacket_Write(&bc->client, bc->tx_buf, enc_rc); - if (BrokerClient_OutQueueMoved(bc, enc_rc)) { + if (BrokerClient_OutQueueMoved(bc, cur, + wr_rc == enc_rc, enc_rc)) { return sent; } /* Scrub the forwarded PUBLISH (which may carry a replayed will or @@ -3783,6 +3798,52 @@ static int BrokerSession_CountOthers(MqttBroker* broker, return count; } +/* Account for a write that completed after the hand-off moved the queue here. + * A delivered QoS 0 entry is spent - MQTT 3.1.1 section 4.3.1 allows no retry - + * so retire it rather than let the resume send it again. A QoS > 0 entry is + * still unacknowledged and stays, but its resume copy is a re-delivery and + * needs DUP [MQTT-3.3.1-1]. Found by walking the carrier, so an entry the + * hand-off dropped on its way in is simply absent. */ +static void BrokerOrphan_ReconcileSent(MqttBroker* broker, + const char* client_id, BrokerOutPub* sent) +{ + BrokerOrphanSession* o; + BrokerOutPub* prev = NULL; + BrokerOutPub* cur; + + if (sent == NULL || !BROKER_STR_VALID(client_id)) { + return; + } + o = BrokerOrphan_Find(broker, client_id); + if (o == NULL) { + return; + } + for (cur = o->out_q_head; cur != NULL; prev = cur, cur = cur->next) { + if (cur != sent) { + continue; + } + if (cur->qos != MQTT_QOS_0) { + cur->retransmit_dup = 1; + return; + } + if (prev == NULL) { + o->out_q_head = cur->next; + } + else { + prev->next = cur->next; + } + if (o->out_q_tail == cur) { + o->out_q_tail = prev; + } + o->out_q_count--; + WBLOG_DBG(broker, + "broker: retiring delivered qos0 topic=%s client_id=%s", + BrokerLog_Sanitize(cur->topic), BrokerLog_Sanitize(client_id)); + BrokerOutPub_Free(cur); + return; + } +} + /* Free everything an orphan owns (queue entries + client_id) but do * NOT unlink from broker->orphan_sessions; the caller does that. */ static void BrokerOrphan_FreeContents(BrokerOrphanSession* o) @@ -6495,11 +6556,8 @@ static void BrokerClient_PublishWill(MqttBroker* broker, BrokerClient* bc) (int)bc->sock, BrokerLog_Sanitize(bc->will_topic), (unsigned)bc->will_payload_len); - /* Claim the Will before fan-out. A fan-out write can service this client's - * close callback inline, which re-enters here; the claim makes that nested - * call return at the has_will check above, so this frame stays the single - * owner of the topic and payload it lends to the encoder below and the - * Will is published once. */ + /* Claimed before fan-out, whose writes can service this client's close + * callback and re-enter here; that nested call then returns above. */ bc->has_will = 0; BrokerClient_PublishWillImmediate(broker, bc->will_topic, diff --git a/src/mqtt_client.c b/src/mqtt_client.c index 646a5688f..0c13b8655 100644 --- a/src/mqtt_client.c +++ b/src/mqtt_client.c @@ -890,18 +890,31 @@ static int MqttClient_ConnectIsPersistent(const MqttConnect* mc_connect) return (mc_connect->clean_session == 0) ? 1 : 0; } -/* Unlike the other wrappers this one reports a lock failure. Skipping the - * reset would leave the previous Session's messages retained, and a later - * reconnect would replay them into a Session they do not belong to. */ +/* Discard the retained copies of a Session the server did not resume. A send + * is allowed once CONNECT reaches the transport, so an entry may instead belong + * to the Session being established: MqttClient_Connect empties the Packet + * Identifier table before sending CONNECT, so a still-reserved identifier marks + * one of those, and it is kept for a later resume [MQTT-4.4.0-1]. Reports a + * lock failure, unlike the wrappers above: a skipped discard would replay the + * previous Session's messages into this one. */ static int MqttClient_Replay_ResetSafe(MqttClient* client) { + int i; #ifdef WOLFMQTT_MULTITHREAD int rc = wm_SemLock(&client->lockClient); if (rc != MQTT_CODE_SUCCESS) { return rc; } #endif - MqttClient_Replay_Reset(client); + for (i = 0; i < MQTT_MAX_REPLAY_MSGS; i++) { + if (client->replay[i].packet_id != 0 && + MqttClient_SendIds_Find(client, + client->replay[i].packet_id) >= 0) { + continue; /* published on this connection */ + } + MqttClient_Replay_FreeSlot(&client->replay[i]); + } + client->replayIdx = MQTT_MAX_REPLAY_MSGS; #ifdef WOLFMQTT_MULTITHREAD wm_SemUnlock(&client->lockClient); #endif @@ -973,11 +986,8 @@ static int MqttClient_SendIds_Find(const MqttClient* client, word16 packet_id) * MQTT_CODE_SUCCESS when it was free (or when isRetransmit says this is a * re-send of the same Control Packet, which [MQTT-2.3.1-3] requires to keep * its original identifier), and MQTT_CODE_ERROR_PACKET_ID when it is still - * awaiting its acknowledgement. - * - * The _Locked form is for a caller that already holds client->lockClient - the - * Session resume walks the replay pool under it - since the semaphore is not - * recursive. */ + * awaiting its acknowledgement. The _Locked form is for a caller already + * holding client->lockClient, which is not recursive. */ static int MqttClient_SendIdReserve_Locked(MqttClient* client, word16 packet_id, void* owner, int isRetransmit, MqttPacketType ack_type) { @@ -4022,10 +4032,7 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect) * into it would inject one Session's messages into another. */ if (!(mc_connect->ack.flags & MQTT_CONNECT_ACK_FLAG_SESSION_PRESENT) || !session_id_matched) { - /* A fresh Session starts with no outbound state to re-send. - * Under the client lock: a send thread may be replacing a slot's - * topic and payload through MqttClient_Replay_AddSafe at the same - * moment, and both sides free what they find there. */ + /* A fresh Session starts with no outbound state to re-send. */ rc = MqttClient_Replay_ResetSafe(client); if (rc != MQTT_CODE_SUCCESS) { return rc; @@ -4037,8 +4044,8 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect) /* The server resumed the Session, so these messages are still in * flight as far as it is concerned: keep their Packet Identifiers * reserved (MqttClient_Connect cleared the table above) and - * re-send them [MQTT-4.4.0-1]. Walked under the client lock for - * the same reason as the reset above. */ + * re-send them [MQTT-4.4.0-1]. A send thread reaches these same + * slots, so the walk holds the client lock. */ #ifdef WOLFMQTT_MULTITHREAD rc = wm_SemLock(&client->lockClient); if (rc != MQTT_CODE_SUCCESS) { diff --git a/tests/test_broker_connect.c b/tests/test_broker_connect.c index 21e17c397..212b0e8f6 100644 --- a/tests/test_broker_connect.c +++ b/tests/test_broker_connect.c @@ -145,8 +145,8 @@ int wolfmqtt_test_broker_fsync(int fd) #endif #ifndef WOLFMQTT_STATIC_MEMORY -/* Count frees of one specific allocation, so a test can assert on the fate of - * a single object rather than on a total that unrelated activity moves. */ +/* Count frees of one allocation, so a test can assert on a single object + * rather than a total that unrelated activity moves. */ static void broker_test_watch_free(void* ptr) { g_free_watch_ptr = ptr; @@ -2860,16 +2860,10 @@ TEST(fanout_survives_reentrant_sub_free) MqttBroker_Free(&broker); } -/* The outbound drain holds a queue entry across the write that sends it. That - * write can service this client's own close callback before it returns - the - * WebSocket transport runs lws_service inline - and the callback hands the - * whole queue to the carrier that keeps the Session alive while the client is - * gone. The entry the drain is holding belongs to that carrier from then on. - * - * BrokerSubs_OrphanClient performs the hand-off through file-local helpers, so - * the hook below stages the same ownership transfer directly: the queue moves - * to a carrier and the client's own queue fields are cleared, which is what - * stops its teardown from freeing the entries a second time. */ +/* The drain holds a queue entry across the write that sends it, and that write + * can service this client's close callback, which hands the whole queue to the + * carrier holding the Session. BrokerSubs_OrphanClient does that through + * file-local helpers, so the hook stages the same transfer directly. */ static MqttBroker* g_handoff_broker; static BrokerClient* g_handoff_client; static void handoff_out_queue_mid_write(void) @@ -2899,8 +2893,7 @@ static void handoff_out_queue_mid_write(void) o->session_expiry_sec = bc->session_expiry_sec; o->orphan_since = wolfmqtt_test_broker_time_s(); - /* Watch the entry the drain is holding, so the test can tell whether the - * drain went on to free memory the carrier owns. */ + /* Watch the entry the drain holds, to see if it frees the carrier's. */ broker_test_watch_free(bc->out_q_head); o->out_q_head = bc->out_q_head; @@ -2929,14 +2922,12 @@ TEST(drain_stops_when_session_handoff_takes_queue) 0x10, 0x0D, 0x00, 0x04, 'M', 'Q', 'T', 'T', 0x04, 0x02, 0x00, 0x3C, 0x00, 0x01, 'P' }; - /* Subscriber "S" with CleanSession=0, so the Session it owns is the kind - * that survives a close and has a carrier to move the queue into. */ + /* CleanSession=0, so the Session survives and has a carrier. */ static const byte connect_sub[] = { 0x10, 0x0D, 0x00, 0x04, 'M', 'Q', 'T', 'T', 0x04, 0x00, 0x00, 0x3C, 0x00, 0x01, 'S' }; - /* SUBSCRIBE packet_id=1, filter "x", granted QoS 0: the delivery the - * drain completes and then unlinks. */ + /* SUBSCRIBE packet_id=1, filter "x", granted QoS 0. */ static const byte subscribe_x[] = { 0x82, 0x06, 0x00, 0x01, 0x00, 0x01, 'x', 0x00 }; @@ -2969,20 +2960,26 @@ TEST(drain_stops_when_session_handoff_takes_queue) mock_client_input_append(0, publish_x, sizeof(publish_x)); (void)MqttBroker_Step(&broker); - /* The hand-off really did run from inside the write. */ + /* The hand-off ran from inside the write. */ ASSERT_NULL(g_write_publish_hook); + /* The delivery completed, which is what makes the entry spent. */ + ASSERT_EQ(1, count_packets_of_type(g_clients[1].out_buf, + g_clients[1].out_len, MQTT_PACKET_TYPE_PUBLISH)); - /* The carrier holds the queue it took. */ + /* The carrier took the queue, and the delivered QoS 0 entry is retired + * from it: MQTT 3.1.1 section 4.3.1 allows no retry. */ o = broker.orphan_sessions; ASSERT_NOT_NULL(o); ASSERT_EQ(1, broker.orphan_session_count); - ASSERT_EQ(1, o->out_q_count); - ASSERT_NOT_NULL(o->out_q_head); + ASSERT_EQ(0, o->out_q_count); + ASSERT_NULL(o->out_q_head); + ASSERT_NULL(o->out_q_tail); - /* The drain must not free an entry it no longer owns. */ - ASSERT_EQ(0, broker_test_watched_frees()); + /* Released once by the owner that retired it: more would mean the drain + * freed a carrier entry, none would leak it. */ + ASSERT_EQ(1, broker_test_watched_frees()); - /* And it must leave no trace of that entry on the client. */ + /* And no trace of it left on the client. */ ASSERT_EQ(0, sub_bc->out_q_count); ASSERT_EQ(0, sub_bc->out_q_inflight); ASSERT_NULL(sub_bc->out_q_head); @@ -2996,6 +2993,82 @@ TEST(drain_stops_when_session_handoff_takes_queue) MqttBroker_Free(&broker); } +/* The QoS > 0 half. A completed at-least-once delivery is still unacknowledged + * so it stays with the carrier, but the subscriber has seen it, so the resume + * copy must be marked a duplicate [MQTT-3.3.1-1]. */ +TEST(handoff_marks_sent_qos1_for_redelivery) +{ + MqttBroker broker; + MqttBrokerNet net; + BrokerClient* sub_bc; + BrokerOrphanSession* o; + int i; + static const byte connect_pub[] = { + 0x10, 0x0D, 0x00, 0x04, 'M', 'Q', 'T', 'T', 0x04, 0x02, 0x00, 0x3C, + 0x00, 0x01, 'P' + }; + /* CleanSession=0, so the Session survives and has a carrier. */ + static const byte connect_sub[] = { + 0x10, 0x0D, 0x00, 0x04, 'M', 'Q', 'T', 'T', 0x04, 0x00, 0x00, 0x3C, + 0x00, 0x01, 'S' + }; + /* SUBSCRIBE packet_id=1, filter "x", granted QoS 1. */ + static const byte subscribe_x[] = { + 0x82, 0x06, 0x00, 0x01, 0x00, 0x01, 'x', 0x01 + }; + /* QoS 1 PUBLISH, Packet Identifier 7, topic "x", payload "ABC". */ + static const byte publish_x[] = { + 0x32, 0x08, 0x00, 0x01, 'x', 0x00, 0x07, 'A', 'B', 'C' + }; + + install_mock_net(&net); + XMEMSET(&broker, 0, sizeof(broker)); + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttBroker_Init(&broker, &net)); + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttBroker_Start(&broker)); + + reset_mock_clients(2); + mock_client_input_append(0, connect_pub, sizeof(connect_pub)); + mock_client_input_append(1, connect_sub, sizeof(connect_sub)); + mock_client_input_append(1, subscribe_x, sizeof(subscribe_x)); + for (i = 0; i < 16; i++) { + (void)MqttBroker_Step(&broker); + } + sub_bc = find_broker_client(&broker, "S"); + ASSERT_NOT_NULL(sub_bc); + ASSERT_NULL(sub_bc->out_q_head); + + broker_test_watch_free(NULL); + g_handoff_broker = &broker; + g_handoff_client = sub_bc; + g_write_publish_hook = handoff_out_queue_mid_write; + + mock_client_input_append(0, publish_x, sizeof(publish_x)); + (void)MqttBroker_Step(&broker); + + ASSERT_NULL(g_write_publish_hook); + ASSERT_EQ(1, count_packets_of_type(g_clients[1].out_buf, + g_clients[1].out_len, MQTT_PACKET_TYPE_PUBLISH)); + + /* Still the carrier's, awaiting its PUBACK, and nothing freed. */ + o = broker.orphan_sessions; + ASSERT_NOT_NULL(o); + ASSERT_EQ(1, o->out_q_count); + ASSERT_NOT_NULL(o->out_q_head); + ASSERT_EQ(0, broker_test_watched_frees()); + /* Flagged, so the resume sends it with DUP set. */ + ASSERT_EQ(1, (int)o->out_q_head->retransmit_dup); + ASSERT_EQ(MQTT_QOS_1, (int)o->out_q_head->qos); + + ASSERT_EQ(0, sub_bc->out_q_count); + ASSERT_NULL(sub_bc->out_q_head); + + g_handoff_broker = NULL; + g_handoff_client = NULL; + broker_test_watch_free(NULL); + MqttBroker_Stop(&broker); + MqttBroker_Free(&broker); +} + #ifdef WOLFMQTT_BROKER_WILL /* Same hazard on the Will fan-out, which runs its own subscription walk. The * publisher's socket drops, the broker fans its Will out to three subscribers, @@ -3071,16 +3144,10 @@ TEST(will_fanout_survives_reentrant_sub_free) MqttBroker_Free(&broker); } -/* A Will is published by fanning it out to matching subscribers, and only - * cleared once that returns. A write inside the fan-out can service the - * owner's own close callback - the WebSocket transport runs lws_service - * inline - and that callback publishes the Will of the client that just - * closed. It must find nothing left to publish: the fan-out below is still - * lending the topic and payload to the encoder, and a nested publish would - * both deliver the Will a second time and free those buffers underneath it. - * - * The hook samples the ownership flag the nested path tests, from inside the - * first Will delivery. */ +/* The Will is fanned out before it is cleared, and a write inside that fan-out + * can service the owner's own close callback, which publishes the same Will. + * It must find nothing left to publish [MQTT-3.1.2-8]. The hook samples the + * flag that nested path tests, from inside the first delivery. */ static BrokerClient* g_will_owner; static int g_will_claimed_during_fanout; static int g_will_topic_live_during_fanout; @@ -3145,8 +3212,7 @@ TEST(takeover_will_claimed_before_fanout) ASSERT_NOT_NULL(g_will_owner); ASSERT_EQ(1, (int)g_will_owner->has_will); ASSERT_NOT_NULL(g_will_owner->will_topic); - /* Nothing has been published yet, so the hook below cannot be consumed by - * an earlier delivery. */ + /* Nothing published yet, so the hook cannot be consumed early. */ ASSERT_EQ(0, count_packets_of_type(g_clients[0].out_buf, g_clients[0].out_len, MQTT_PACKET_TYPE_PUBLISH)); @@ -3166,8 +3232,7 @@ TEST(takeover_will_claimed_before_fanout) /* Claimed before the first write: a close callback delivered during the * fan-out returns without republishing or freeing anything. */ ASSERT_EQ(1, g_will_claimed_during_fanout); - /* And the buffers the fan-out is lending out are still owned, so the - * encoder is reading live memory. */ + /* And the buffers it lends the encoder are still owned. */ ASSERT_EQ(1, g_will_topic_live_during_fanout); /* Exactly one copy reaches each subscriber [MQTT-3.1.2-8]. */ @@ -16904,6 +16969,7 @@ int main(int argc, char** argv) RUN_TEST(publish_fanout_alloc_failure_v5_qos2_pubrec_failure); #endif RUN_TEST(drain_stops_when_session_handoff_takes_queue); + RUN_TEST(handoff_marks_sent_qos1_for_redelivery); #ifdef WOLFMQTT_NONBLOCK RUN_TEST(outbound_queue_short_write_resumes_on_next_step); #ifdef WOLFMQTT_BROKER_RETAINED diff --git a/tests/test_mqtt_client.c b/tests/test_mqtt_client.c index aa5b1f71d..cdf3e7b9e 100644 --- a/tests/test_mqtt_client.c +++ b/tests/test_mqtt_client.c @@ -48,21 +48,18 @@ static MqttNet test_net; static byte test_tx_buf[TEST_TX_BUF_SIZE]; static byte test_rx_buf[TEST_RX_BUF_SIZE]; -/* The suite routes the library's allocator through here (see - * tests/test_client_alloc.h) so a test can watch one specific allocation and - * record the state the library was in when it released it. Everything else is - * a straight pass-through. */ +/* The library allocator is routed through here (tests/test_broker_alloc.h) so + * a test can watch one allocation and record the state it was released in. */ static void* g_free_watch_ptr; static int g_free_watch_count; static int g_free_watch_lock_count; -void* wolfmqtt_test_client_malloc(size_t size) +void* wolfmqtt_test_broker_malloc(size_t size) { return malloc(size); } -/* How many times the client lock is held right now, or -1 where the build - * keeps no count to read. */ +/* Lock depth, or -1 where the build keeps no count to read. */ static int test_client_lock_depth(void) { #if defined(WOLFMQTT_MULTITHREAD) && defined(WOLFMQTT_POSIX_SEMAPHORES) && \ @@ -73,7 +70,7 @@ static int test_client_lock_depth(void) #endif } -void wolfmqtt_test_client_free(void* ptr) +void wolfmqtt_test_broker_free(void* ptr) { if (ptr != NULL && ptr == g_free_watch_ptr) { g_free_watch_count++; @@ -4010,19 +4007,98 @@ TEST(reconnect_without_session_present_replays_nothing) ASSERT_EQ(1, g_frames_written); /* the CONNECT only */ } +/* A send is allowed once CONNECT reaches the transport, so a publish can be + * retained while this handshake still waits for CONNACK. It belongs to the + * Session being established, so a Session Present = 0 answer must keep it for + * a later resume [MQTT-4.4.0-1]. The write callback seeds what such a publish + * leaves behind, at the only point it is reachable: the Packet Identifier + * table is empty and CONNECT is on the wire. */ +static int g_seed_concurrent_publish; +static byte g_seed_payload[] = "seeded"; + +static int mock_net_write_seed_publish(void *context, const byte* buf, + int buf_len, int timeout_ms) +{ + int rc = mock_net_write_accept(context, buf, buf_len, timeout_ms); + + if (g_seed_concurrent_publish && buf_len > 0 && + (buf[0] >> 4) == MQTT_PACKET_TYPE_CONNECT) { +#ifndef WOLFMQTT_STATIC_MEMORY + char* topic; + byte* payload; +#endif + + g_seed_concurrent_publish = 0; + test_client.replay[1].packet_id = 0x5678; + test_client.replay[1].qos = MQTT_QOS_1; + test_client.replay[1].haveCopy = 1; + test_client.replay[1].payload_len = (word32)(sizeof(g_seed_payload) - 1); +#ifdef WOLFMQTT_STATIC_MEMORY + XMEMCPY(test_client.replay[1].topic, "new/1", 6); + XMEMCPY(test_client.replay[1].payload, g_seed_payload, + sizeof(g_seed_payload) - 1); +#else + topic = (char*)WOLFMQTT_MALLOC(6); + payload = (byte*)WOLFMQTT_MALLOC(sizeof(g_seed_payload) - 1); + if (topic == NULL || payload == NULL) { + WOLFMQTT_FREE(topic); + WOLFMQTT_FREE(payload); + test_client.replay[1].packet_id = 0; + return rc; + } + XMEMCPY(topic, "new/1", 6); + XMEMCPY(payload, g_seed_payload, sizeof(g_seed_payload) - 1); + test_client.replay[1].topic = topic; + test_client.replay[1].payload = payload; +#endif + /* The reservation that marks the entry as this connection's. */ + test_client.send_inflight[0].packet_id = 0x5678; + test_client.send_inflight[0].ack_type = MQTT_PACKET_TYPE_PUBLISH_ACK; + } + return rc; +} + +TEST(fresh_session_reset_keeps_publish_from_this_connection) +{ + int rc; + MqttConnect connect; + MqttPublish publish; + static byte payload[] = "hello"; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); +#ifdef WOLFMQTT_V5 + test_client.protocol_level = MQTT_CONNECT_PROTOCOL_LEVEL_4; +#endif + ASSERT_EQ(MQTT_CODE_SUCCESS, run_initial_connect(&connect)); + + /* An unacknowledged publish from the Session that is about to end. */ + init_qos_publish(&publish, MQTT_QOS_1, 0x1234, "sensor/temp", + payload, (word32)(sizeof(payload) - 1)); + rc = run_publish_unacked(&publish); + ASSERT_EQ(MQTT_CODE_ERROR_NETWORK, rc); + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttClient_NetDisconnect(&test_client)); + ASSERT_EQ(0x1234, (int)test_client.replay[0].packet_id); + + g_seed_concurrent_publish = 1; + rc = run_reconnect_with(&connect, 0, mock_net_write_seed_publish); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + /* The seed really was planted, so the assertions below mean something. */ + ASSERT_EQ(0, g_seed_concurrent_publish); + + /* The ended Session's message is gone. */ + ASSERT_EQ(0, (int)test_client.replay[0].packet_id); + /* The one published on this connection is kept, with its copy intact. */ + ASSERT_EQ(0x5678, (int)test_client.replay[1].packet_id); + ASSERT_STR_EQ("new/1", test_client.replay[1].topic); +} + #if defined(WOLFMQTT_MULTITHREAD) && defined(WOLFMQTT_POSIX_SEMAPHORES) && \ - !defined(WOLFMQTT_NO_COND_SIGNAL) -/* Discarding the retained copies releases the topic and payload of every - * slot. A send thread reaches those same slots through the replay store's - * lock-taking wrapper while the connection is being re-established - the - * client admits send calls once CONNECT has reached the transport - and both - * sides free what they find there. So the discard has to hold the client lock - * as well, or the two can free the same buffer or leave one pointing at - * memory the other released. - * - * The POSIX lock keeps its own depth, and the allocator hook samples it at the - * moment the slot's topic is released, which is the access that has to be - * covered. */ + !defined(WOLFMQTT_NO_COND_SIGNAL) && !defined(WOLFMQTT_STATIC_MEMORY) +/* A send thread reaches the same replay slots under the client lock, and both + * sides release what they find there, so the discard has to hold it too. The + * POSIX lock keeps its own depth; the allocator hook samples it as the slot's + * topic is released, which is the access that has to be covered. */ TEST(fresh_session_reset_frees_replay_under_client_lock) { int rc; @@ -4043,8 +4119,7 @@ TEST(fresh_session_reset_frees_replay_under_client_lock) ASSERT_EQ(MQTT_CODE_ERROR_NETWORK, rc); ASSERT_EQ(MQTT_CODE_SUCCESS, MqttClient_NetDisconnect(&test_client)); - /* The unacknowledged PUBLISH is retained, so the discard has a real - * allocation to release. */ + /* Retained, so the discard has a real allocation to release. */ ASSERT_EQ(0x1234, (int)test_client.replay[0].packet_id); ASSERT_NOT_NULL(test_client.replay[0].topic); test_client_watch_free(test_client.replay[0].topic); @@ -9962,8 +10037,9 @@ void run_mqtt_client_tests(void) RUN_TEST(publish_allowed_past_replay_pool_when_connack_ends_session); #endif RUN_TEST(reconnect_without_session_present_replays_nothing); + RUN_TEST(fresh_session_reset_keeps_publish_from_this_connection); #if defined(WOLFMQTT_MULTITHREAD) && defined(WOLFMQTT_POSIX_SEMAPHORES) && \ - !defined(WOLFMQTT_NO_COND_SIGNAL) + !defined(WOLFMQTT_NO_COND_SIGNAL) && !defined(WOLFMQTT_STATIC_MEMORY) RUN_TEST(fresh_session_reset_frees_replay_under_client_lock); #endif RUN_TEST(reconnect_does_not_replay_acked_publish); From 0db728f36fd263053d791b2f8213eba2f2eddb5e Mon Sep 17 00:00:00 2001 From: Kareem Date: Tue, 6 Oct 2026 10:10:35 -0700 Subject: [PATCH 05/11] Code review feedback --- src/mqtt_broker.c | 5 + src/mqtt_client.c | 65 ++++++++--- tests/test_broker_connect.c | 145 ++++++++++++++++++++++++ tests/test_mqtt_client.c | 214 +++++++++++++++++++++++++++++------- wolfmqtt/mqtt_client.h | 6 + 5 files changed, 379 insertions(+), 56 deletions(-) diff --git a/src/mqtt_broker.c b/src/mqtt_broker.c index 0d1699da4..f1a1c4372 100644 --- a/src/mqtt_broker.c +++ b/src/mqtt_broker.c @@ -3824,6 +3824,11 @@ static void BrokerOrphan_ReconcileSent(MqttBroker* broker, } if (cur->qos != MQTT_QOS_0) { cur->retransmit_dup = 1; + #ifdef WOLFMQTT_BROKER_PERSIST + /* BrokerOrphan_Take shadow-wrote this entry before the write + * returned, so the stored copy still says it was never sent. */ + (void)BrokerPersist_PutOutPub(broker, client_id, cur); + #endif return; } if (prev == NULL) { diff --git a/src/mqtt_client.c b/src/mqtt_client.c index 0c13b8655..e53ffd153 100644 --- a/src/mqtt_client.c +++ b/src/mqtt_client.c @@ -765,6 +765,7 @@ static void MqttClient_Replay_Add(MqttClient* client, MqttPublish* publish) #endif slot->payload_len = publish->total_len; slot->haveCopy = 1; + slot->onThisConn = 1; } /* QoS 2 advanced past PUBREC, so [MQTT-4.4.0-1] replays the PUBREL rather @@ -797,6 +798,7 @@ static void MqttClient_Replay_PubRelSent(MqttClient* client, word16 packet_id) MqttClient_Replay_FreeSlot(slot); slot->packet_id = packet_id; slot->qos = MQTT_QOS_2; + slot->onThisConn = 1; } else if (slot->qos != MQTT_QOS_2) { return; @@ -890,13 +892,28 @@ static int MqttClient_ConnectIsPersistent(const MqttConnect* mc_connect) return (mc_connect->clean_session == 0) ? 1 : 0; } -/* Discard the retained copies of a Session the server did not resume. A send - * is allowed once CONNECT reaches the transport, so an entry may instead belong - * to the Session being established: MqttClient_Connect empties the Packet - * Identifier table before sending CONNECT, so a still-reserved identifier marks - * one of those, and it is kept for a later resume [MQTT-4.4.0-1]. Reports a - * lock failure, unlike the wrappers above: a skipped discard would replay the - * previous Session's messages into this one. */ +/* Everything retained so far belongs to the Session that is ending; a send + * admitted once CONNECT is on the wire tags itself. */ +static void MqttClient_Replay_NewConnSafe(MqttClient* client) +{ + int i; +#ifdef WOLFMQTT_MULTITHREAD + if (wm_SemLock(&client->lockClient) != MQTT_CODE_SUCCESS) { + return; + } +#endif + for (i = 0; i < MQTT_MAX_REPLAY_MSGS; i++) { + client->replay[i].onThisConn = 0; + } +#ifdef WOLFMQTT_MULTITHREAD + wm_SemUnlock(&client->lockClient); +#endif +} + +/* Discard the retained copies of a Session the server did not resume, keeping + * anything published on this connection: a later resume still owes that + * [MQTT-4.4.0-1]. Reports a lock failure, unlike the wrappers above: a skipped + * discard would replay the old Session's messages into this one. */ static int MqttClient_Replay_ResetSafe(MqttClient* client) { int i; @@ -907,10 +924,8 @@ static int MqttClient_Replay_ResetSafe(MqttClient* client) } #endif for (i = 0; i < MQTT_MAX_REPLAY_MSGS; i++) { - if (client->replay[i].packet_id != 0 && - MqttClient_SendIds_Find(client, - client->replay[i].packet_id) >= 0) { - continue; /* published on this connection */ + if (client->replay[i].onThisConn) { + continue; } MqttClient_Replay_FreeSlot(&client->replay[i]); } @@ -3292,7 +3307,8 @@ static int MqttClient_Replay_EncodeNext(MqttClient* client, word16* dropped_id) * still acknowledge the message it stands for, and that acknowledgement * is what releases both the slot and its Packet Identifier * [MQTT-2.3.1-3]. */ - if (slot->packet_id == 0 || (!slot->pubrelSent && !slot->haveCopy)) { + if (slot->packet_id == 0 || slot->onThisConn || + (!slot->pubrelSent && !slot->haveCopy)) { rc = 0; keep = 1; } @@ -3605,6 +3621,9 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect) * and outside the WOLFMQTT_V5 block below, since the table exists in * every build. */ MqttClient_SendIdsReset(client); +#ifndef WOLFMQTT_NO_SESSION_REPLAY + MqttClient_Replay_NewConnSafe(client); +#endif #ifdef WOLFMQTT_V5 #ifdef WOLFMQTT_MULTITHREAD @@ -4053,7 +4072,9 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect) } #endif for (i = 0; i < MQTT_MAX_REPLAY_MSGS; i++) { - if (client->replay[i].packet_id == 0) { + if (client->replay[i].packet_id == 0 || + client->replay[i].onThisConn) { + /* Not the resumed Session's: already in flight here. */ continue; } if (!client->replay[i].pubrelSent && @@ -4065,6 +4086,14 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect) MqttClient_Replay_FreeSlot(&client->replay[i]); continue; } + if (MqttClient_SendIds_Find(client, + client->replay[i].packet_id) >= 0) { + /* A live request holds the identifier and + * [MQTT-2.3.1-2] allows one use at a time, so drop this + * rather than take its reservation over. */ + MqttClient_Replay_FreeSlot(&client->replay[i]); + continue; + } (void)MqttClient_SendIdReserve_Locked(client, client->replay[i].packet_id, &client->replay[i], 1, client->replay[i].pubrelSent ? @@ -4671,7 +4700,7 @@ static int MqttPublishMsg(MqttClient *client, MqttPublish *publish, * message's ownership of it either way. */ MqttClient_RestoreRecvQuota(client, publish); #endif - MqttClient_CancelMessage(client, (MqttObject*)publish); + (void)MqttClient_CancelMessage(client, (MqttObject*)publish); if (wrote > 0) { /* Part of the PUBLISH reached the server, so it may have * seen the Packet Identifier. Reclaim the reservation the @@ -4718,7 +4747,7 @@ static int MqttPublishMsg(MqttClient *client, MqttPublish *publish, * above. */ MqttClient_RestoreRecvQuota(client, publish); #endif - MqttClient_CancelMessage(client, (MqttObject*)publish); + (void)MqttClient_CancelMessage(client, (MqttObject*)publish); /* Reaching the payload means the fixed header - and with it * the Packet Identifier - is already on the wire, so keep the * reservation the cancel dropped [MQTT-2.3.1-3]. */ @@ -4971,7 +5000,7 @@ int MqttClient_Subscribe(MqttClient *client, MqttSubscribe *subscribe) wrote = (rc == xfer) ? xfer : client->write.pos; MqttWriteStop(client, &subscribe->stat); if (rc != xfer) { - MqttClient_CancelMessage(client, (MqttObject*)subscribe); + (void)MqttClient_CancelMessage(client, (MqttObject*)subscribe); if (wrote > 0) { /* Part of the SUBSCRIBE reached the server, so it may have * seen the Packet Identifier. Reclaim the reservation the @@ -5132,7 +5161,7 @@ int MqttClient_Unsubscribe(MqttClient *client, MqttUnsubscribe *unsubscribe) wrote = (rc == xfer) ? xfer : client->write.pos; MqttWriteStop(client, &unsubscribe->stat); if (rc != xfer) { - MqttClient_CancelMessage(client, (MqttObject*)unsubscribe); + (void)MqttClient_CancelMessage(client, (MqttObject*)unsubscribe); if (wrote > 0) { /* Part of the UNSUBSCRIBE reached the server, so it may have * seen the Packet Identifier. Reclaim the reservation the @@ -5315,7 +5344,7 @@ int MqttClient_Ping_ex(MqttClient *client, MqttPing* ping) #endif MqttWriteStop(client, &ping->stat); if (rc != xfer) { - MqttClient_CancelMessage(client, (MqttObject*)ping); + (void)MqttClient_CancelMessage(client, (MqttObject*)ping); return rc; } diff --git a/tests/test_broker_connect.c b/tests/test_broker_connect.c index 212b0e8f6..25a2ca61a 100644 --- a/tests/test_broker_connect.c +++ b/tests/test_broker_connect.c @@ -2908,6 +2908,24 @@ static void handoff_out_queue_mid_write(void) o->next = broker->orphan_sessions; broker->orphan_sessions = o; broker->orphan_session_count++; + +#ifdef WOLFMQTT_BROKER_PERSIST + /* BrokerOrphan_Take shadow-writes here, before the write it interrupted + * has returned, so the stored copy records an unsent entry. */ + (void)BrokerPersist_PutOrphanSession(broker, o->client_id, + o->protocol_level, o->session_expiry_sec, o->orphan_since); + { + BrokerOutPub* e; + word64 seq = 1; + + for (e = o->out_q_head; e != NULL; e = e->next) { + e->enqueue_seq = seq++; + if (e->qos > MQTT_QOS_0) { + (void)BrokerPersist_PutOutPub(broker, o->client_id, e); + } + } + } +#endif } TEST(drain_stops_when_session_handoff_takes_queue) @@ -13449,6 +13467,31 @@ static int persist_order_put(void* ctx, byte ns, const byte* key, return MQTT_CODE_SUCCESS; } +/* Keyed variant: a record written again under the same key replaces it, as a + * real backend does. persist_order_put appends, which would turn a re-write + * into a second queue entry. */ +static int persist_keyed_put(void* ctx, byte ns, const byte* key, + word16 key_len, const byte* blob, word32 blob_len) +{ + PersistOrderStore* store = (PersistOrderStore*)ctx; + int i; + + if (ns == BROKER_PERSIST_NS_OUTQ) { + for (i = 0; i < store->outq_count; i++) { + if (store->outq[i].key_len == key_len && + XMEMCMP(store->outq[i].key, key, key_len) == 0) { + if (blob_len > sizeof(store->outq[i].blob)) { + return MQTT_CODE_ERROR_OUT_OF_BUFFER; + } + XMEMCPY(store->outq[i].blob, blob, blob_len); + store->outq[i].blob_len = blob_len; + return MQTT_CODE_SUCCESS; + } + } + } + return persist_order_put(ctx, ns, key, key_len, blob, blob_len); +} + static int persist_order_get(void* ctx, byte ns, const byte* key, word16 key_len, byte* out, word32* inout_len) { @@ -13629,6 +13672,107 @@ TEST(persist_partial_publish_restart_keeps_dup) MqttBroker_Stop(&restored); MqttBroker_Free(&restored); } + +/* The same requirement when the Session hand-off takes the queue mid-write. + * BrokerOrphan_Take shadow-writes each entry before that write returns, so the + * stored copy still says it was never sent; the completed delivery has to be + * written through, or a restart before reconnect re-sends it without DUP + * [MQTT-3.3.1-1]. */ +TEST(persist_handoff_sent_qos1_restart_keeps_dup) +{ + MqttBroker source; + MqttBroker restored; + MqttBrokerNet net; + MqttBrokerPersistHooks hooks; + PersistOrderStore store; + BrokerClient* sub_bc; + PublishInfo info; + word16 packet_id; + int i; + static const byte connect_pub[] = { + 0x10, 0x0D, 0x00, 0x04, 'M', 'Q', 'T', 'T', 0x04, 0x02, 0x00, 0x3C, + 0x00, 0x01, 'P' + }; + /* CleanSession=0, so the Session survives and has a carrier. */ + static const byte connect_sub[] = { + 0x10, 0x0D, 0x00, 0x04, 'M', 'Q', 'T', 'T', 0x04, 0x00, 0x00, 0x3C, + 0x00, 0x01, 'S' + }; + static const byte subscribe_x[] = { + 0x82, 0x06, 0x00, 0x01, 0x00, 0x01, 'x', 0x01 + }; + static const byte publish_x[] = { + 0x32, 0x08, 0x00, 0x01, 'x', 0x00, 0x07, 'A', 'B', 'C' + }; + + install_mock_net(&net); + XMEMSET(&source, 0, sizeof(source)); + XMEMSET(&restored, 0, sizeof(restored)); + XMEMSET(&hooks, 0, sizeof(hooks)); + XMEMSET(&store, 0, sizeof(store)); + hooks.kv_put = persist_keyed_put; + hooks.kv_get = persist_order_get; + hooks.kv_iter = persist_order_iter; + hooks.ctx = &store; +#ifdef WOLFMQTT_BROKER_PERSIST_ENCRYPT + hooks.derive_key = persist_order_derive_key; +#endif + + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttBroker_Init(&source, &net)); + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttBroker_SetPersistHooks(&source, &hooks)); + ASSERT_EQ(MQTT_CODE_SUCCESS, BrokerPersist_Restore(&source)); + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttBroker_Start(&source)); + reset_mock_clients(2); + mock_client_input_append(0, connect_pub, sizeof(connect_pub)); + mock_client_input_append(1, connect_sub, sizeof(connect_sub)); + mock_client_input_append(1, subscribe_x, sizeof(subscribe_x)); + for (i = 0; i < 16; i++) { + (void)MqttBroker_Step(&source); + } + sub_bc = find_broker_client(&source, "S"); + ASSERT_NOT_NULL(sub_bc); + + /* The hand-off runs from inside the write, which then completes. */ + broker_test_watch_free(NULL); + g_handoff_broker = &source; + g_handoff_client = sub_bc; + g_write_publish_hook = handoff_out_queue_mid_write; + mock_client_input_append(0, publish_x, sizeof(publish_x)); + (void)MqttBroker_Step(&source); + ASSERT_NULL(g_write_publish_hook); + g_handoff_broker = NULL; + g_handoff_client = NULL; + + ASSERT_NOT_NULL(source.orphan_sessions); + ASSERT_NOT_NULL(source.orphan_sessions->out_q_head); + ASSERT_EQ(1, (int)source.orphan_sessions->out_q_head->retransmit_dup); + packet_id = source.orphan_sessions->out_q_head->packet_id; + ASSERT_EQ(1, store.outq_count); + + MqttBroker_Stop(&source); + MqttBroker_Free(&source); + + /* Restart before the subscriber reconnects. */ + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttBroker_Init(&restored, &net)); + ASSERT_EQ(MQTT_CODE_SUCCESS, + MqttBroker_SetPersistHooks(&restored, &hooks)); + ASSERT_EQ(MQTT_CODE_SUCCESS, BrokerPersist_Restore(&restored)); + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttBroker_Start(&restored)); + reset_mock_clients(1); + mock_client_input_append(0, connect_sub, sizeof(connect_sub)); + for (i = 0; i < 16; i++) { + (void)MqttBroker_Step(&restored); + } + + info = first_publish_info(g_clients[0].out_buf, g_clients[0].out_len); + ASSERT_TRUE(info.found); + /* 0x3A = PUBLISH | DUP | QoS 1. */ + ASSERT_EQ(0x3A, info.first_byte); + ASSERT_EQ(packet_id, info.packet_id); + + MqttBroker_Stop(&restored); + MqttBroker_Free(&restored); +} #endif #if defined(WOLFMQTT_V5) && defined(WOLFMQTT_BROKER_WILL) && \ @@ -17304,6 +17448,7 @@ int main(int argc, char** argv) #ifdef WOLFMQTT_NONBLOCK RUN_TEST(persist_subs_written_when_ack_deferred); RUN_TEST(persist_partial_publish_restart_keeps_dup); + RUN_TEST(persist_handoff_sent_qos1_restart_keeps_dup); #endif RUN_TEST(persist_mixed_qos_queue_preserves_fifo); #if defined(WOLFMQTT_V5) && defined(WOLFMQTT_BROKER_WILL) && \ diff --git a/tests/test_mqtt_client.c b/tests/test_mqtt_client.c index cdf3e7b9e..ec3e7494f 100644 --- a/tests/test_mqtt_client.c +++ b/tests/test_mqtt_client.c @@ -4009,51 +4009,74 @@ TEST(reconnect_without_session_present_replays_nothing) /* A send is allowed once CONNECT reaches the transport, so a publish can be * retained while this handshake still waits for CONNACK. It belongs to the - * Session being established, so a Session Present = 0 answer must keep it for - * a later resume [MQTT-4.4.0-1]. The write callback seeds what such a publish - * leaves behind, at the only point it is reachable: the Packet Identifier - * table is empty and CONNECT is on the wire. */ -static int g_seed_concurrent_publish; + * Session being established, which the entry's onThisConn tag records. The + * write callback seeds what such a send leaves behind, at the only point it is + * reachable: the tags have just been cleared and CONNECT is on the wire. */ +static int g_seed_slot = -1; /* slot to seed, -1 seeds nothing */ +static word16 g_seed_packet_id; +static byte g_seed_on_this_conn; +static byte g_seed_reserve_ack_type; /* 0 reserves no identifier */ static byte g_seed_payload[] = "seeded"; +static void seed_replay_slot(int idx, word16 packet_id, byte on_this_conn) +{ + MqttReplayMsg* slot = &test_client.replay[idx]; + word32 len = (word32)(sizeof(g_seed_payload) - 1); + + slot->packet_id = packet_id; + slot->qos = MQTT_QOS_1; + slot->haveCopy = 1; + slot->onThisConn = on_this_conn; + slot->payload_len = len; +#ifdef WOLFMQTT_STATIC_MEMORY + XMEMCPY(slot->topic, "new/1", 6); + XMEMCPY(slot->payload, g_seed_payload, len); +#else + slot->topic = (char*)WOLFMQTT_MALLOC(6); + slot->payload = (byte*)WOLFMQTT_MALLOC(len); + if (slot->topic == NULL || slot->payload == NULL) { + WOLFMQTT_FREE(slot->topic); + WOLFMQTT_FREE(slot->payload); + slot->topic = NULL; + slot->payload = NULL; + slot->packet_id = 0; + return; + } + XMEMCPY(slot->topic, "new/1", 6); + XMEMCPY(slot->payload, g_seed_payload, len); +#endif +} + +/* A SUBSCRIBE admitted once CONNECT is on the wire, reduced to the state it + * leaves behind: its Packet Identifier reservation. */ +static word16 g_reserve_sub_id = 0x1234; +static int mock_net_write_reserve_subscribe_id(void *context, const byte* buf, + int buf_len, int timeout_ms) +{ + int rc = mock_net_write_accept(context, buf, buf_len, timeout_ms); + + if (buf_len > 0 && (buf[0] >> 4) == MQTT_PACKET_TYPE_CONNECT) { + test_client.send_inflight[0].packet_id = g_reserve_sub_id; + test_client.send_inflight[0].ack_type = MQTT_PACKET_TYPE_SUBSCRIBE_ACK; + test_client.send_inflight[0].owner = &test_client.msg; + } + return rc; +} + static int mock_net_write_seed_publish(void *context, const byte* buf, int buf_len, int timeout_ms) { int rc = mock_net_write_accept(context, buf, buf_len, timeout_ms); - if (g_seed_concurrent_publish && buf_len > 0 && + if (g_seed_slot >= 0 && buf_len > 0 && (buf[0] >> 4) == MQTT_PACKET_TYPE_CONNECT) { -#ifndef WOLFMQTT_STATIC_MEMORY - char* topic; - byte* payload; -#endif - - g_seed_concurrent_publish = 0; - test_client.replay[1].packet_id = 0x5678; - test_client.replay[1].qos = MQTT_QOS_1; - test_client.replay[1].haveCopy = 1; - test_client.replay[1].payload_len = (word32)(sizeof(g_seed_payload) - 1); -#ifdef WOLFMQTT_STATIC_MEMORY - XMEMCPY(test_client.replay[1].topic, "new/1", 6); - XMEMCPY(test_client.replay[1].payload, g_seed_payload, - sizeof(g_seed_payload) - 1); -#else - topic = (char*)WOLFMQTT_MALLOC(6); - payload = (byte*)WOLFMQTT_MALLOC(sizeof(g_seed_payload) - 1); - if (topic == NULL || payload == NULL) { - WOLFMQTT_FREE(topic); - WOLFMQTT_FREE(payload); - test_client.replay[1].packet_id = 0; - return rc; + seed_replay_slot(g_seed_slot, g_seed_packet_id, g_seed_on_this_conn); + if (g_seed_reserve_ack_type != 0) { + test_client.send_inflight[0].packet_id = g_seed_packet_id; + test_client.send_inflight[0].ack_type = g_seed_reserve_ack_type; + test_client.send_inflight[0].owner = &test_client.msg; } - XMEMCPY(topic, "new/1", 6); - XMEMCPY(payload, g_seed_payload, sizeof(g_seed_payload) - 1); - test_client.replay[1].topic = topic; - test_client.replay[1].payload = payload; -#endif - /* The reservation that marks the entry as this connection's. */ - test_client.send_inflight[0].packet_id = 0x5678; - test_client.send_inflight[0].ack_type = MQTT_PACKET_TYPE_PUBLISH_ACK; + g_seed_slot = -1; } return rc; } @@ -4080,11 +4103,14 @@ TEST(fresh_session_reset_keeps_publish_from_this_connection) ASSERT_EQ(MQTT_CODE_SUCCESS, MqttClient_NetDisconnect(&test_client)); ASSERT_EQ(0x1234, (int)test_client.replay[0].packet_id); - g_seed_concurrent_publish = 1; + g_seed_slot = 1; + g_seed_packet_id = 0x5678; + g_seed_on_this_conn = 1; + g_seed_reserve_ack_type = MQTT_PACKET_TYPE_PUBLISH_ACK; rc = run_reconnect_with(&connect, 0, mock_net_write_seed_publish); ASSERT_EQ(MQTT_CODE_SUCCESS, rc); /* The seed really was planted, so the assertions below mean something. */ - ASSERT_EQ(0, g_seed_concurrent_publish); + ASSERT_EQ(-1, g_seed_slot); /* The ended Session's message is gone. */ ASSERT_EQ(0, (int)test_client.replay[0].packet_id); @@ -4093,6 +4119,115 @@ TEST(fresh_session_reset_keeps_publish_from_this_connection) ASSERT_STR_EQ("new/1", test_client.replay[1].topic); } +/* A reserved Packet Identifier does not prove where a replay entry came from: + * the table also holds SUBSCRIBE and UNSUBSCRIBE reservations. A request sent + * after CONNECT that happens to reuse an ended Session's identifier must not + * make a fresh CONNACK keep that Session's PUBLISH, which could otherwise be + * replayed into a later Session. */ +TEST(fresh_session_reset_drops_old_publish_behind_subscribe_id) +{ + int rc; + MqttConnect connect; + MqttPublish publish; + static byte payload[] = "hello"; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); +#ifdef WOLFMQTT_V5 + test_client.protocol_level = MQTT_CONNECT_PROTOCOL_LEVEL_4; +#endif + ASSERT_EQ(MQTT_CODE_SUCCESS, run_initial_connect(&connect)); + + init_qos_publish(&publish, MQTT_QOS_1, 0x1234, "sensor/temp", + payload, (word32)(sizeof(payload) - 1)); + rc = run_publish_unacked(&publish); + ASSERT_EQ(MQTT_CODE_ERROR_NETWORK, rc); + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttClient_NetDisconnect(&test_client)); + ASSERT_EQ(0x1234, (int)test_client.replay[0].packet_id); + + /* A SUBSCRIBE on the new connection takes the same identifier. Nothing is + * seeded into the pool: slot 0 is already the ended Session's entry. */ + g_seed_slot = -1; + test_net.write = mock_net_write_accept; + rc = run_reconnect_with(&connect, 0, mock_net_write_reserve_subscribe_id); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + ASSERT_EQ(0x1234, (int)test_client.send_inflight[0].packet_id); + + /* Discarded: it was never this connection's, whoever holds the id now. */ + ASSERT_EQ(0, (int)test_client.replay[0].packet_id); +} + +/* The resumed-Session walk must re-send only what the previous connection left + * unacknowledged. An entry published after CONNECT is already in flight here, + * so replaying it would deliver it twice. */ +TEST(resume_does_not_replay_publish_from_this_connection) +{ + int rc; + MqttConnect connect; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); +#ifdef WOLFMQTT_V5 + test_client.protocol_level = MQTT_CONNECT_PROTOCOL_LEVEL_4; +#endif + ASSERT_EQ(MQTT_CODE_SUCCESS, run_initial_connect(&connect)); + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttClient_NetDisconnect(&test_client)); + + g_seed_slot = 0; + g_seed_packet_id = 0x5678; + g_seed_on_this_conn = 1; + g_seed_reserve_ack_type = MQTT_PACKET_TYPE_PUBLISH_ACK; + g_frames_written = 0; + rc = run_reconnect_with(&connect, 1, mock_net_write_seed_publish); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + ASSERT_EQ(-1, g_seed_slot); + + /* The CONNECT only: nothing from this connection is re-sent. */ + ASSERT_EQ(1, g_frames_written); + /* And the entry stays retained, awaiting its own acknowledgement. */ + ASSERT_EQ(0x5678, (int)test_client.replay[0].packet_id); +} + +/* [MQTT-2.3.1-2] allows one use of an identifier at a time. If a request on + * the new connection already holds the one an ended Session's entry needs, the + * replay cannot go out, and it must not take that live request's reservation + * over either. */ +TEST(resume_drops_replay_whose_id_is_taken_on_this_connection) +{ + int rc; + MqttConnect connect; + MqttPublish publish; + static byte payload[] = "hello"; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); +#ifdef WOLFMQTT_V5 + test_client.protocol_level = MQTT_CONNECT_PROTOCOL_LEVEL_4; +#endif + ASSERT_EQ(MQTT_CODE_SUCCESS, run_initial_connect(&connect)); + + init_qos_publish(&publish, MQTT_QOS_1, 0x1234, "sensor/temp", + payload, (word32)(sizeof(payload) - 1)); + rc = run_publish_unacked(&publish); + ASSERT_EQ(MQTT_CODE_ERROR_NETWORK, rc); + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttClient_NetDisconnect(&test_client)); + ASSERT_EQ(0x1234, (int)test_client.replay[0].packet_id); + + g_seed_slot = -1; + g_frames_written = 0; + rc = run_reconnect_with(&connect, 1, mock_net_write_reserve_subscribe_id); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + + /* The CONNECT only: the entry could not be re-sent. */ + ASSERT_EQ(1, g_frames_written); + /* Dropped rather than replayed behind the live request. */ + ASSERT_EQ(0, (int)test_client.replay[0].packet_id); + /* And the SUBSCRIBE still owns its reservation, unchanged. */ + ASSERT_EQ(0x1234, (int)test_client.send_inflight[0].packet_id); + ASSERT_EQ(MQTT_PACKET_TYPE_SUBSCRIBE_ACK, + (int)test_client.send_inflight[0].ack_type); +} + #if defined(WOLFMQTT_MULTITHREAD) && defined(WOLFMQTT_POSIX_SEMAPHORES) && \ !defined(WOLFMQTT_NO_COND_SIGNAL) && !defined(WOLFMQTT_STATIC_MEMORY) /* A send thread reaches the same replay slots under the client lock, and both @@ -10038,6 +10173,9 @@ void run_mqtt_client_tests(void) #endif RUN_TEST(reconnect_without_session_present_replays_nothing); RUN_TEST(fresh_session_reset_keeps_publish_from_this_connection); + RUN_TEST(fresh_session_reset_drops_old_publish_behind_subscribe_id); + RUN_TEST(resume_does_not_replay_publish_from_this_connection); + RUN_TEST(resume_drops_replay_whose_id_is_taken_on_this_connection); #if defined(WOLFMQTT_MULTITHREAD) && defined(WOLFMQTT_POSIX_SEMAPHORES) && \ !defined(WOLFMQTT_NO_COND_SIGNAL) && !defined(WOLFMQTT_STATIC_MEMORY) RUN_TEST(fresh_session_reset_frees_replay_under_client_lock); diff --git a/wolfmqtt/mqtt_client.h b/wolfmqtt/mqtt_client.h index c6be1d77c..c6c8fb4bb 100644 --- a/wolfmqtt/mqtt_client.h +++ b/wolfmqtt/mqtt_client.h @@ -351,6 +351,12 @@ typedef struct _MqttReplayMsg { * for the pool or came from a payload callback, which has nothing to * copy; such an entry can still replay a PUBREL but not a PUBLISH. */ byte haveCopy; + /* Published on the Network Connection being established rather than + * carried from the Session that ended. Sends are admitted once CONNECT is + * on the wire, so both kinds can be in the pool when CONNACK arrives, and + * only the latter are the old Session's to discard or re-send. Cleared on + * every entry when a handshake starts. */ + byte onThisConn; } MqttReplayMsg; #endif /* !WOLFMQTT_NO_SESSION_REPLAY */ From 5d1779a2bdfa9efdbc28049c88cb99887745655c Mon Sep 17 00:00:00 2001 From: Kareem Date: Tue, 6 Oct 2026 11:59:50 -0700 Subject: [PATCH 06/11] Fix compilation error --- tests/test_broker_connect.c | 55 +++++++++++++++++++------------------ 1 file changed, 28 insertions(+), 27 deletions(-) diff --git a/tests/test_broker_connect.c b/tests/test_broker_connect.c index 25a2ca61a..22d95199b 100644 --- a/tests/test_broker_connect.c +++ b/tests/test_broker_connect.c @@ -13467,31 +13467,6 @@ static int persist_order_put(void* ctx, byte ns, const byte* key, return MQTT_CODE_SUCCESS; } -/* Keyed variant: a record written again under the same key replaces it, as a - * real backend does. persist_order_put appends, which would turn a re-write - * into a second queue entry. */ -static int persist_keyed_put(void* ctx, byte ns, const byte* key, - word16 key_len, const byte* blob, word32 blob_len) -{ - PersistOrderStore* store = (PersistOrderStore*)ctx; - int i; - - if (ns == BROKER_PERSIST_NS_OUTQ) { - for (i = 0; i < store->outq_count; i++) { - if (store->outq[i].key_len == key_len && - XMEMCMP(store->outq[i].key, key, key_len) == 0) { - if (blob_len > sizeof(store->outq[i].blob)) { - return MQTT_CODE_ERROR_OUT_OF_BUFFER; - } - XMEMCPY(store->outq[i].blob, blob, blob_len); - store->outq[i].blob_len = blob_len; - return MQTT_CODE_SUCCESS; - } - } - } - return persist_order_put(ctx, ns, key, key_len, blob, blob_len); -} - static int persist_order_get(void* ctx, byte ns, const byte* key, word16 key_len, byte* out, word32* inout_len) { @@ -13673,6 +13648,33 @@ TEST(persist_partial_publish_restart_keeps_dup) MqttBroker_Free(&restored); } +#endif + +/* Keyed variant: a record written again under the same key replaces it, as a + * real backend does. persist_order_put appends, which would turn a re-write + * into a second queue entry. */ +static int persist_keyed_put(void* ctx, byte ns, const byte* key, + word16 key_len, const byte* blob, word32 blob_len) +{ + PersistOrderStore* store = (PersistOrderStore*)ctx; + int i; + + if (ns == BROKER_PERSIST_NS_OUTQ) { + for (i = 0; i < store->outq_count; i++) { + if (store->outq[i].key_len == key_len && + XMEMCMP(store->outq[i].key, key, key_len) == 0) { + if (blob_len > sizeof(store->outq[i].blob)) { + return MQTT_CODE_ERROR_OUT_OF_BUFFER; + } + XMEMCPY(store->outq[i].blob, blob, blob_len); + store->outq[i].blob_len = blob_len; + return MQTT_CODE_SUCCESS; + } + } + } + return persist_order_put(ctx, ns, key, key_len, blob, blob_len); +} + /* The same requirement when the Session hand-off takes the queue mid-write. * BrokerOrphan_Take shadow-writes each entry before that write returns, so the * stored copy still says it was never sent; the completed delivery has to be @@ -13773,7 +13775,6 @@ TEST(persist_handoff_sent_qos1_restart_keeps_dup) MqttBroker_Stop(&restored); MqttBroker_Free(&restored); } -#endif #if defined(WOLFMQTT_V5) && defined(WOLFMQTT_BROKER_WILL) && \ WOLFMQTT_MAX_QOS >= 1 @@ -17448,8 +17449,8 @@ int main(int argc, char** argv) #ifdef WOLFMQTT_NONBLOCK RUN_TEST(persist_subs_written_when_ack_deferred); RUN_TEST(persist_partial_publish_restart_keeps_dup); - RUN_TEST(persist_handoff_sent_qos1_restart_keeps_dup); #endif + RUN_TEST(persist_handoff_sent_qos1_restart_keeps_dup); RUN_TEST(persist_mixed_qos_queue_preserves_fifo); #if defined(WOLFMQTT_V5) && defined(WOLFMQTT_BROKER_WILL) && \ WOLFMQTT_MAX_QOS >= 1 From 9dc2b2d50a366bba0b07f574c6ecaac64f5ad278 Mon Sep 17 00:00:00 2001 From: Kareem Date: Tue, 6 Oct 2026 14:56:56 -0700 Subject: [PATCH 07/11] Code review feedback --- src/mqtt_client.c | 20 ++++++++++---- tests/test_broker_connect.c | 8 ++++++ tests/test_mqtt_client.c | 55 +++++++++++++++++++++++++++++++++++++ 3 files changed, 77 insertions(+), 6 deletions(-) diff --git a/src/mqtt_client.c b/src/mqtt_client.c index e53ffd153..11d781a33 100644 --- a/src/mqtt_client.c +++ b/src/mqtt_client.c @@ -712,6 +712,9 @@ static void MqttClient_Replay_Add(MqttClient* client, MqttPublish* publish) slot->packet_id = publish->packet_id; slot->qos = (byte)publish->qos; slot->retain = publish->retain; + /* Before the returns below: they leave the slot in use, and a PUBREL can + * still be recorded against it. */ + slot->onThisConn = 1; /* A streamed publish delivers its payload through a callback, so there is * nothing here to copy; the entry still tracks the QoS 2 PUBREL stage. @@ -765,7 +768,6 @@ static void MqttClient_Replay_Add(MqttClient* client, MqttPublish* publish) #endif slot->payload_len = publish->total_len; slot->haveCopy = 1; - slot->onThisConn = 1; } /* QoS 2 advanced past PUBREC, so [MQTT-4.4.0-1] replays the PUBREL rather @@ -893,13 +895,15 @@ static int MqttClient_ConnectIsPersistent(const MqttConnect* mc_connect) } /* Everything retained so far belongs to the Session that is ending; a send - * admitted once CONNECT is on the wire tags itself. */ -static void MqttClient_Replay_NewConnSafe(MqttClient* client) + * admitted once CONNECT is on the wire tags itself. Reports a lock failure: + * stale tags would have the handshake skip or keep the wrong entries. */ +static int MqttClient_Replay_NewConnSafe(MqttClient* client) { int i; #ifdef WOLFMQTT_MULTITHREAD - if (wm_SemLock(&client->lockClient) != MQTT_CODE_SUCCESS) { - return; + int rc = wm_SemLock(&client->lockClient); + if (rc != MQTT_CODE_SUCCESS) { + return rc; } #endif for (i = 0; i < MQTT_MAX_REPLAY_MSGS; i++) { @@ -908,6 +912,7 @@ static void MqttClient_Replay_NewConnSafe(MqttClient* client) #ifdef WOLFMQTT_MULTITHREAD wm_SemUnlock(&client->lockClient); #endif + return MQTT_CODE_SUCCESS; } /* Discard the retained copies of a Session the server did not resume, keeping @@ -3622,7 +3627,10 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect) * every build. */ MqttClient_SendIdsReset(client); #ifndef WOLFMQTT_NO_SESSION_REPLAY - MqttClient_Replay_NewConnSafe(client); + rc = MqttClient_Replay_NewConnSafe(client); + if (rc != MQTT_CODE_SUCCESS) { + return rc; + } #endif #ifdef WOLFMQTT_V5 diff --git a/tests/test_broker_connect.c b/tests/test_broker_connect.c index 22d95199b..32d0141ed 100644 --- a/tests/test_broker_connect.c +++ b/tests/test_broker_connect.c @@ -3011,6 +3011,7 @@ TEST(drain_stops_when_session_handoff_takes_queue) MqttBroker_Free(&broker); } +#if WOLFMQTT_MAX_QOS >= 1 /* The QoS > 0 half. A completed at-least-once delivery is still unacknowledged * so it stays with the carrier, but the subscriber has seen it, so the resume * copy must be marked a duplicate [MQTT-3.3.1-1]. */ @@ -3086,6 +3087,7 @@ TEST(handoff_marks_sent_qos1_for_redelivery) MqttBroker_Stop(&broker); MqttBroker_Free(&broker); } +#endif /* WOLFMQTT_MAX_QOS >= 1 */ #ifdef WOLFMQTT_BROKER_WILL /* Same hazard on the Will fan-out, which runs its own subscription walk. The @@ -13675,6 +13677,7 @@ static int persist_keyed_put(void* ctx, byte ns, const byte* key, return persist_order_put(ctx, ns, key, key_len, blob, blob_len); } +#if WOLFMQTT_MAX_QOS >= 1 /* The same requirement when the Session hand-off takes the queue mid-write. * BrokerOrphan_Take shadow-writes each entry before that write returns, so the * stored copy still says it was never sent; the completed delivery has to be @@ -13775,6 +13778,7 @@ TEST(persist_handoff_sent_qos1_restart_keeps_dup) MqttBroker_Stop(&restored); MqttBroker_Free(&restored); } +#endif /* WOLFMQTT_MAX_QOS >= 1 */ #if defined(WOLFMQTT_V5) && defined(WOLFMQTT_BROKER_WILL) && \ WOLFMQTT_MAX_QOS >= 1 @@ -17114,7 +17118,9 @@ int main(int argc, char** argv) RUN_TEST(publish_fanout_alloc_failure_v5_qos2_pubrec_failure); #endif RUN_TEST(drain_stops_when_session_handoff_takes_queue); +#if WOLFMQTT_MAX_QOS >= 1 RUN_TEST(handoff_marks_sent_qos1_for_redelivery); +#endif #ifdef WOLFMQTT_NONBLOCK RUN_TEST(outbound_queue_short_write_resumes_on_next_step); #ifdef WOLFMQTT_BROKER_RETAINED @@ -17450,7 +17456,9 @@ int main(int argc, char** argv) RUN_TEST(persist_subs_written_when_ack_deferred); RUN_TEST(persist_partial_publish_restart_keeps_dup); #endif +#if WOLFMQTT_MAX_QOS >= 1 RUN_TEST(persist_handoff_sent_qos1_restart_keeps_dup); +#endif RUN_TEST(persist_mixed_qos_queue_preserves_fifo); #if defined(WOLFMQTT_V5) && defined(WOLFMQTT_BROKER_WILL) && \ WOLFMQTT_MAX_QOS >= 1 diff --git a/tests/test_mqtt_client.c b/tests/test_mqtt_client.c index ec3e7494f..50326802d 100644 --- a/tests/test_mqtt_client.c +++ b/tests/test_mqtt_client.c @@ -4592,6 +4592,60 @@ TEST(reconnect_does_not_replay_streamed_publish) ASSERT_EQ(MQTT_PACKET_TYPE_CONNECT, g_replay_types[0]); } +/* MqttClient_Replay_Add gives up on the copy for a streamed publish, a v5 + * publish carrying properties, and an oversized one, but the slot it already + * claimed stays in use and a PUBREL can still be recorded against it. The tag + * that says which connection the entry belongs to therefore has to be set + * before those returns, or a send admitted during a handshake is mistaken for + * the previous Session's and re-sent. */ +TEST(replay_uncopied_slot_still_tagged_to_this_connection) +{ + int rc; + int i; + int slot = -1; + MqttConnect connect; + MqttPublish publish; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); +#ifdef WOLFMQTT_V5 + test_client.protocol_level = MQTT_CONNECT_PROTOCOL_LEVEL_4; +#endif + ASSERT_EQ(MQTT_CODE_SUCCESS, run_initial_connect(&connect)); + + XMEMSET(g_replay_stream_buf, 0, sizeof(g_replay_stream_buf)); + test_net.write = mock_net_write_accept; + test_net.read = mock_net_read; /* network error: no PUBACK */ + + /* Streamed: buffer_len is one chunk, total_len the whole message, so the + * store has nothing to copy. */ + XMEMSET(&publish, 0, sizeof(publish)); + publish.qos = MQTT_QOS_1; + publish.packet_id = 0x4321; + publish.topic_name = "sensor/temp"; + publish.buffer = g_replay_stream_buf; + publish.buffer_len = (word32)REPLAY_STREAM_CHUNK; + publish.total_len = (word32)REPLAY_BIG_PAYLOAD_LEN; + + rc = MQTT_CODE_CONTINUE; + for (i = 0; i < 20 && rc == MQTT_CODE_CONTINUE; i++) { + rc = MqttClient_Publish_ex(&test_client, &publish, replay_stream_cb); + } + ASSERT_EQ(MQTT_CODE_ERROR_NETWORK, rc); + + for (i = 0; i < MQTT_MAX_REPLAY_MSGS; i++) { + if (test_client.replay[i].packet_id == 0x4321) { + slot = i; + break; + } + } + /* The slot is in use even though the copy was skipped. */ + ASSERT_TRUE(slot >= 0); + ASSERT_EQ(0, (int)test_client.replay[slot].haveCopy); + /* And it is tagged to this connection. */ + ASSERT_EQ(1, (int)test_client.replay[slot].onThisConn); +} + /* An entry the replay cannot rebuild is dropped from the Session, so nothing * will ever acknowledge it and the Packet Identifier the reconnect reserved for * it must be given back [MQTT-2.3.1-3]. Left reserved, the next legitimate @@ -10187,6 +10241,7 @@ void run_mqtt_client_tests(void) RUN_TEST(reconnect_drops_replay_larger_than_tx_buf); RUN_TEST(reconnect_replays_at_most_pool_size); RUN_TEST(reconnect_does_not_replay_streamed_publish); + RUN_TEST(replay_uncopied_slot_still_tagged_to_this_connection); RUN_TEST(reconnect_dropped_replay_frees_packet_id); RUN_TEST(reconnect_with_new_client_id_does_not_replay); RUN_TEST(reconnect_with_hash_colliding_client_id_does_not_replay); From 49362a85759f907ab3e0f2b6d73ced084eb91b88 Mon Sep 17 00:00:00 2001 From: Kareem Date: Thu, 22 Oct 2026 17:45:22 -0700 Subject: [PATCH 08/11] Keep retained replay entries' Packet Identifiers reserved for the whole CONNECT handshake instead of only from CONNACK, releasing them if the server answers with a fresh Session. --- src/mqtt_client.c | 76 +++++++++++++------- tests/test_mqtt_client.c | 145 +++++++++++++++++++++------------------ 2 files changed, 129 insertions(+), 92 deletions(-) diff --git a/src/mqtt_client.c b/src/mqtt_client.c index 11d781a33..6b84fe39c 100644 --- a/src/mqtt_client.c +++ b/src/mqtt_client.c @@ -90,6 +90,12 @@ static int MqttClient_AuthEx(MqttClient *client, MqttAuth* auth, #if !defined(WOLFMQTT_MULTITHREAD) && !defined(WOLFMQTT_NONBLOCK) static int MqttClient_CancelMessage(MqttClient *client, MqttObject* msg); #endif +#ifndef WOLFMQTT_NO_SESSION_REPLAY +static void MqttClient_SendIds_Drop_Locked(MqttClient* client, + word16 packet_id); +static int MqttClient_SendIdReserve_Locked(MqttClient* client, word16 packet_id, + void* owner, int isRetransmit, MqttPacketType ack_type); +#endif #ifdef WOLFMQTT_MULTITHREAD #ifdef WOLFMQTT_USER_THREADING @@ -894,6 +900,14 @@ static int MqttClient_ConnectIsPersistent(const MqttConnect* mc_connect) return (mc_connect->clean_session == 0) ? 1 : 0; } +static MqttPacketType MqttClient_Replay_AckType(const MqttReplayMsg* slot) +{ + if (slot->pubrelSent || slot->qos == MQTT_QOS_2) { + return MQTT_PACKET_TYPE_PUBLISH_COMP; + } + return MQTT_PACKET_TYPE_PUBLISH_ACK; +} + /* Everything retained so far belongs to the Session that is ending; a send * admitted once CONNECT is on the wire tags itself. Reports a lock failure: * stale tags would have the handshake skip or keep the wrong entries. */ @@ -908,6 +922,14 @@ static int MqttClient_Replay_NewConnSafe(MqttClient* client) #endif for (i = 0; i < MQTT_MAX_REPLAY_MSGS; i++) { client->replay[i].onThisConn = 0; + /* Hold for the handshake: [MQTT-2.3.1-2] allows one use at a time + * and [MQTT-4.4.0-1] still owes these a re-send. Released below on a + * fresh Session. */ + if (client->replay[i].packet_id != 0) { + (void)MqttClient_SendIdReserve_Locked(client, + client->replay[i].packet_id, &client->replay[i], 1, + MqttClient_Replay_AckType(&client->replay[i])); + } } #ifdef WOLFMQTT_MULTITHREAD wm_SemUnlock(&client->lockClient); @@ -932,6 +954,8 @@ static int MqttClient_Replay_ResetSafe(MqttClient* client) if (client->replay[i].onThisConn) { continue; } + /* Release the handshake hold. */ + MqttClient_SendIds_Drop_Locked(client, client->replay[i].packet_id); MqttClient_Replay_FreeSlot(&client->replay[i]); } client->replayIdx = MQTT_MAX_REPLAY_MSGS; @@ -1081,6 +1105,25 @@ static int MqttClient_SendIdReserve(MqttClient* client, word16 packet_id, * lock: releasing first and removing afterwards would let a publisher reuse * the identifier in between, and this ack would then delete the new * exchange's replay state. */ +#ifndef WOLFMQTT_NO_SESSION_REPLAY +/* Caller holds client->lockClient. Unlike MqttClient_SendIdRelease, drops only + * the reservation; the caller is already disposing of the replay entry. */ +static void MqttClient_SendIds_Drop_Locked(MqttClient* client, word16 packet_id) +{ + int i; + + if (packet_id == 0) { + return; + } + i = MqttClient_SendIds_Find(client, packet_id); + if (i >= 0) { + client->send_inflight[i].packet_id = 0; + client->send_inflight[i].owner = NULL; + client->send_inflight[i].ack_type = (byte)MQTT_PACKET_TYPE_RESERVED; + } +} +#endif + static void MqttClient_SendIdRelease(MqttClient* client, word16 packet_id, MqttPacketType ack_type) { @@ -4068,11 +4111,9 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect) else { int i; - /* The server resumed the Session, so these messages are still in - * flight as far as it is concerned: keep their Packet Identifiers - * reserved (MqttClient_Connect cleared the table above) and - * re-send them [MQTT-4.4.0-1]. A send thread reaches these same - * slots, so the walk holds the client lock. */ + /* Resumed Session: still in flight to the server and go out + * again [MQTT-4.4.0-1], identifiers held since the handshake. A + * send thread reaches these slots, so the walk holds the lock. */ #ifdef WOLFMQTT_MULTITHREAD rc = wm_SemLock(&client->lockClient); if (rc != MQTT_CODE_SUCCESS) { @@ -4087,28 +4128,13 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect) } if (!client->replay[i].pubrelSent && !client->replay[i].haveCopy) { - /* Nothing to re-send: the payload was never retained, so - * [MQTT-4.4.0-1] cannot be honoured for this one. Drop it - * instead of reserving an identifier that no - * acknowledgement will ever release [MQTT-2.3.1-3]. */ - MqttClient_Replay_FreeSlot(&client->replay[i]); - continue; - } - if (MqttClient_SendIds_Find(client, - client->replay[i].packet_id) >= 0) { - /* A live request holds the identifier and - * [MQTT-2.3.1-2] allows one use at a time, so drop this - * rather than take its reservation over. */ + /* Payload never retained, so [MQTT-4.4.0-1] cannot be + * honoured: drop it and release the identifier, which no + * acknowledgement would free [MQTT-2.3.1-3]. */ + MqttClient_SendIds_Drop_Locked(client, + client->replay[i].packet_id); MqttClient_Replay_FreeSlot(&client->replay[i]); - continue; } - (void)MqttClient_SendIdReserve_Locked(client, - client->replay[i].packet_id, &client->replay[i], 1, - client->replay[i].pubrelSent ? - MQTT_PACKET_TYPE_PUBLISH_COMP : - ((client->replay[i].qos == MQTT_QOS_2) ? - MQTT_PACKET_TYPE_PUBLISH_COMP : - MQTT_PACKET_TYPE_PUBLISH_ACK)); } #ifdef WOLFMQTT_MULTITHREAD wm_SemUnlock(&client->lockClient); diff --git a/tests/test_mqtt_client.c b/tests/test_mqtt_client.c index 50326802d..a363779b8 100644 --- a/tests/test_mqtt_client.c +++ b/tests/test_mqtt_client.c @@ -674,12 +674,24 @@ TEST(connect_rejects_nonzero_inbound_topic_alias_max) static byte g_canned_buf[64]; static int g_canned_len; static int g_canned_pos; +/* -1 samples the outbound reservations on the next read: for a reconnect, + * after CONNECT is on the wire and before CONNACK. 0 disables. */ +static int g_hold_probe_id; static int mock_net_read_canned(void *context, byte* buf, int buf_len, int timeout_ms) { int n; (void)context; (void)timeout_ms; + if (g_hold_probe_id < 0) { + g_hold_probe_id = 0; + for (n = 0; n < MQTT_MAX_SEND_INFLIGHT; n++) { + if (test_client.send_inflight[n].packet_id != 0) { + g_hold_probe_id = (int)test_client.send_inflight[n].packet_id; + break; + } + } + } n = g_canned_len - g_canned_pos; if (n <= 0) { return 0; /* exhausted -> MQTT_CODE_CONTINUE under nonblock */ @@ -4007,6 +4019,7 @@ TEST(reconnect_without_session_present_replays_nothing) ASSERT_EQ(1, g_frames_written); /* the CONNECT only */ } + /* A send is allowed once CONNECT reaches the transport, so a publish can be * retained while this handshake still waits for CONNACK. It belongs to the * Session being established, which the entry's onThisConn tag records. The @@ -4047,21 +4060,6 @@ static void seed_replay_slot(int idx, word16 packet_id, byte on_this_conn) #endif } -/* A SUBSCRIBE admitted once CONNECT is on the wire, reduced to the state it - * leaves behind: its Packet Identifier reservation. */ -static word16 g_reserve_sub_id = 0x1234; -static int mock_net_write_reserve_subscribe_id(void *context, const byte* buf, - int buf_len, int timeout_ms) -{ - int rc = mock_net_write_accept(context, buf, buf_len, timeout_ms); - - if (buf_len > 0 && (buf[0] >> 4) == MQTT_PACKET_TYPE_CONNECT) { - test_client.send_inflight[0].packet_id = g_reserve_sub_id; - test_client.send_inflight[0].ack_type = MQTT_PACKET_TYPE_SUBSCRIBE_ACK; - test_client.send_inflight[0].owner = &test_client.msg; - } - return rc; -} static int mock_net_write_seed_publish(void *context, const byte* buf, int buf_len, int timeout_ms) @@ -4119,14 +4117,14 @@ TEST(fresh_session_reset_keeps_publish_from_this_connection) ASSERT_STR_EQ("new/1", test_client.replay[1].topic); } -/* A reserved Packet Identifier does not prove where a replay entry came from: - * the table also holds SUBSCRIBE and UNSUBSCRIBE reservations. A request sent - * after CONNECT that happens to reuse an ended Session's identifier must not - * make a fresh CONNACK keep that Session's PUBLISH, which could otherwise be - * replayed into a later Session. */ -TEST(fresh_session_reset_drops_old_publish_behind_subscribe_id) +/* A retained entry's Packet Identifier stays reserved across the handshake: + * [MQTT-2.3.1-2] allows one use at a time and [MQTT-4.4.0-1] still owes it a + * re-send. A fresh Session gives it back. */ +TEST(handshake_holds_retained_packet_id) { int rc; + int i; + int held = 0; MqttConnect connect; MqttPublish publish; static byte payload[] = "hello"; @@ -4144,26 +4142,43 @@ TEST(fresh_session_reset_drops_old_publish_behind_subscribe_id) ASSERT_EQ(MQTT_CODE_ERROR_NETWORK, rc); ASSERT_EQ(MQTT_CODE_SUCCESS, MqttClient_NetDisconnect(&test_client)); ASSERT_EQ(0x1234, (int)test_client.replay[0].packet_id); + /* No reservations survive the disconnect, so the hold below is the + * handshake's own. */ + for (i = 0; i < MQTT_MAX_SEND_INFLIGHT; i++) { + if (test_client.send_inflight[i].packet_id != 0) { + held++; + } + } + ASSERT_EQ(0, held); - /* A SUBSCRIBE on the new connection takes the same identifier. Nothing is - * seeded into the pool: slot 0 is already the ended Session's entry. */ - g_seed_slot = -1; - test_net.write = mock_net_write_accept; - rc = run_reconnect_with(&connect, 0, mock_net_write_reserve_subscribe_id); + g_hold_probe_id = -1; /* arm the sample */ + rc = run_reconnect_with(&connect, 0, mock_net_write_accept); ASSERT_EQ(MQTT_CODE_SUCCESS, rc); - ASSERT_EQ(0x1234, (int)test_client.send_inflight[0].packet_id); - /* Discarded: it was never this connection's, whoever holds the id now. */ + /* Held while the handshake was in flight. */ + ASSERT_EQ(0x1234, g_hold_probe_id); + + /* The Session came back fresh, so the entry and its identifier both go. */ ASSERT_EQ(0, (int)test_client.replay[0].packet_id); + held = 0; + for (i = 0; i < MQTT_MAX_SEND_INFLIGHT; i++) { + if (test_client.send_inflight[i].packet_id == 0x1234) { + held++; + } + } + ASSERT_EQ(0, held); } -/* The resumed-Session walk must re-send only what the previous connection left - * unacknowledged. An entry published after CONNECT is already in flight here, - * so replaying it would deliver it twice. */ -TEST(resume_does_not_replay_publish_from_this_connection) +/* Resumed instead: the entry is owed a re-send, so the hold stays and the + * identifier is still reserved once the handshake completes. */ +TEST(resume_keeps_retained_packet_id_reserved) { int rc; + int i; + int held = 0; MqttConnect connect; + MqttPublish publish; + static byte payload[] = "hello"; rc = test_init_client(); ASSERT_EQ(MQTT_CODE_SUCCESS, rc); @@ -4171,33 +4186,34 @@ TEST(resume_does_not_replay_publish_from_this_connection) test_client.protocol_level = MQTT_CONNECT_PROTOCOL_LEVEL_4; #endif ASSERT_EQ(MQTT_CODE_SUCCESS, run_initial_connect(&connect)); + + init_qos_publish(&publish, MQTT_QOS_1, 0x1234, "sensor/temp", + payload, (word32)(sizeof(payload) - 1)); + rc = run_publish_unacked(&publish); + ASSERT_EQ(MQTT_CODE_ERROR_NETWORK, rc); ASSERT_EQ(MQTT_CODE_SUCCESS, MqttClient_NetDisconnect(&test_client)); - g_seed_slot = 0; - g_seed_packet_id = 0x5678; - g_seed_on_this_conn = 1; - g_seed_reserve_ack_type = MQTT_PACKET_TYPE_PUBLISH_ACK; - g_frames_written = 0; - rc = run_reconnect_with(&connect, 1, mock_net_write_seed_publish); + g_hold_probe_id = -1; + rc = run_reconnect_with(&connect, 1, mock_net_write_accept); ASSERT_EQ(MQTT_CODE_SUCCESS, rc); - ASSERT_EQ(-1, g_seed_slot); - /* The CONNECT only: nothing from this connection is re-sent. */ - ASSERT_EQ(1, g_frames_written); - /* And the entry stays retained, awaiting its own acknowledgement. */ - ASSERT_EQ(0x5678, (int)test_client.replay[0].packet_id); + ASSERT_EQ(0x1234, g_hold_probe_id); + /* Still retained and still reserved, awaiting its acknowledgement. */ + ASSERT_EQ(0x1234, (int)test_client.replay[0].packet_id); + for (i = 0; i < MQTT_MAX_SEND_INFLIGHT; i++) { + if (test_client.send_inflight[i].packet_id == 0x1234) { + held++; + } + } + ASSERT_EQ(1, held); } -/* [MQTT-2.3.1-2] allows one use of an identifier at a time. If a request on - * the new connection already holds the one an ended Session's entry needs, the - * replay cannot go out, and it must not take that live request's reservation - * over either. */ -TEST(resume_drops_replay_whose_id_is_taken_on_this_connection) +/* The resumed walk re-sends only what the previous connection left + * unacknowledged; an entry published after CONNECT is already in flight. */ +TEST(resume_does_not_replay_publish_from_this_connection) { int rc; MqttConnect connect; - MqttPublish publish; - static byte payload[] = "hello"; rc = test_init_client(); ASSERT_EQ(MQTT_CODE_SUCCESS, rc); @@ -4205,29 +4221,24 @@ TEST(resume_drops_replay_whose_id_is_taken_on_this_connection) test_client.protocol_level = MQTT_CONNECT_PROTOCOL_LEVEL_4; #endif ASSERT_EQ(MQTT_CODE_SUCCESS, run_initial_connect(&connect)); - - init_qos_publish(&publish, MQTT_QOS_1, 0x1234, "sensor/temp", - payload, (word32)(sizeof(payload) - 1)); - rc = run_publish_unacked(&publish); - ASSERT_EQ(MQTT_CODE_ERROR_NETWORK, rc); ASSERT_EQ(MQTT_CODE_SUCCESS, MqttClient_NetDisconnect(&test_client)); - ASSERT_EQ(0x1234, (int)test_client.replay[0].packet_id); - g_seed_slot = -1; + g_seed_slot = 0; + g_seed_packet_id = 0x5678; + g_seed_on_this_conn = 1; + g_seed_reserve_ack_type = MQTT_PACKET_TYPE_PUBLISH_ACK; g_frames_written = 0; - rc = run_reconnect_with(&connect, 1, mock_net_write_reserve_subscribe_id); + rc = run_reconnect_with(&connect, 1, mock_net_write_seed_publish); ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + ASSERT_EQ(-1, g_seed_slot); - /* The CONNECT only: the entry could not be re-sent. */ + /* The CONNECT only: nothing from this connection is re-sent. */ ASSERT_EQ(1, g_frames_written); - /* Dropped rather than replayed behind the live request. */ - ASSERT_EQ(0, (int)test_client.replay[0].packet_id); - /* And the SUBSCRIBE still owns its reservation, unchanged. */ - ASSERT_EQ(0x1234, (int)test_client.send_inflight[0].packet_id); - ASSERT_EQ(MQTT_PACKET_TYPE_SUBSCRIBE_ACK, - (int)test_client.send_inflight[0].ack_type); + /* And the entry stays retained, awaiting its own acknowledgement. */ + ASSERT_EQ(0x5678, (int)test_client.replay[0].packet_id); } + #if defined(WOLFMQTT_MULTITHREAD) && defined(WOLFMQTT_POSIX_SEMAPHORES) && \ !defined(WOLFMQTT_NO_COND_SIGNAL) && !defined(WOLFMQTT_STATIC_MEMORY) /* A send thread reaches the same replay slots under the client lock, and both @@ -10227,9 +10238,9 @@ void run_mqtt_client_tests(void) #endif RUN_TEST(reconnect_without_session_present_replays_nothing); RUN_TEST(fresh_session_reset_keeps_publish_from_this_connection); - RUN_TEST(fresh_session_reset_drops_old_publish_behind_subscribe_id); + RUN_TEST(handshake_holds_retained_packet_id); RUN_TEST(resume_does_not_replay_publish_from_this_connection); - RUN_TEST(resume_drops_replay_whose_id_is_taken_on_this_connection); + RUN_TEST(resume_keeps_retained_packet_id_reserved); #if defined(WOLFMQTT_MULTITHREAD) && defined(WOLFMQTT_POSIX_SEMAPHORES) && \ !defined(WOLFMQTT_NO_COND_SIGNAL) && !defined(WOLFMQTT_STATIC_MEMORY) RUN_TEST(fresh_session_reset_frees_replay_under_client_lock); From 68282c2d3916d69395d0ed7824ebfa9e3873e23a Mon Sep 17 00:00:00 2001 From: Kareem Date: Mon, 26 Oct 2026 16:43:53 -0700 Subject: [PATCH 09/11] Respect persistent messages with the newly added PERSISTENT_SESSION flag in MqttClient_Replay_ResetSafe. Treat fully and partially written packets the same in BrokerClient_OutQueueMoved. --- src/mqtt_broker.c | 29 +++++++++------ src/mqtt_client.c | 15 +++++--- tests/test_broker_connect.c | 73 +++++++++++++++++++++++++++++++++++++ tests/test_mqtt_client.c | 55 +++++++++++++++++++++++++++- 4 files changed, 154 insertions(+), 18 deletions(-) diff --git a/src/mqtt_broker.c b/src/mqtt_broker.c index f1a1c4372..50ea1441e 100644 --- a/src/mqtt_broker.c +++ b/src/mqtt_broker.c @@ -2255,27 +2255,30 @@ static void BrokerOrphan_ReconcileSent(MqttBroker* broker, /* A write can service this client's close callback before returning - the * WebSocket transport runs lws_service inline - which hands the queue to the * Session carrier. The walked entries came from bc->out_q_head, so an empty - * head while one is still held means the carrier owns them now. Returns + * head while one is still held means the carrier owns them now. onWire says + * whether any of this packet reached the subscriber, since the markers the + * drain would have set below are skipped once the entry is gone. Returns * non-zero when the drain must stop and leave them to it. */ static int BrokerClient_OutQueueMoved(BrokerClient* bc, BrokerOutPub* sent, - int completed, int enc_len) + int onWire, int enc_len) { if (bc->out_q_head != NULL) { return 0; } #ifndef WOLFMQTT_STATIC_MEMORY - if (completed) { + if (onWire) { BrokerOrphan_ReconcileSent(bc->broker, bc->client_id, sent); } #else (void)sent; - (void)completed; + (void)onWire; #endif bc->out_q_pending_len = 0; BROKER_FORCE_ZERO(bc->tx_buf, enc_len); return 1; } + /* Send as many QUEUED entries from out_q as the inflight cap allows. * * Ordering: walks from out_q_head, never reorders. Already-sent entries @@ -2341,7 +2344,8 @@ static int BrokerClient_DrainOutQueue(BrokerClient* bc) } wr_rc = MqttPacket_Write(&bc->client, bc->tx_buf, rel_rc); if (BrokerClient_OutQueueMoved(bc, cur, - wr_rc == rel_rc, rel_rc)) { + wr_rc == rel_rc || bc->client.write.pos > 0, + rel_rc)) { return sent; } if (wr_rc == MQTT_CODE_CONTINUE) { @@ -2443,7 +2447,8 @@ static int BrokerClient_DrainOutQueue(BrokerClient* bc) int wr_rc; wr_rc = MqttPacket_Write(&bc->client, bc->tx_buf, enc_rc); if (BrokerClient_OutQueueMoved(bc, cur, - wr_rc == enc_rc, enc_rc)) { + wr_rc == enc_rc || bc->client.write.pos > 0, + enc_rc)) { return sent; } /* Scrub the forwarded PUBLISH (which may carry a replayed will or @@ -3798,12 +3803,12 @@ static int BrokerSession_CountOthers(MqttBroker* broker, return count; } -/* Account for a write that completed after the hand-off moved the queue here. - * A delivered QoS 0 entry is spent - MQTT 3.1.1 section 4.3.1 allows no retry - - * so retire it rather than let the resume send it again. A QoS > 0 entry is - * still unacknowledged and stays, but its resume copy is a re-delivery and - * needs DUP [MQTT-3.3.1-1]. Found by walking the carrier, so an entry the - * hand-off dropped on its way in is simply absent. */ +/* Account for a write that reached the subscriber, whole or in part, after the + * hand-off moved the queue here. A QoS 0 entry is spent either way - MQTT 3.1.1 + * section 4.3.1 allows no retry - so retire it rather than let the resume send + * it again. A QoS > 0 entry is still unacknowledged and stays, but its resume + * copy is a re-delivery and needs DUP [MQTT-3.3.1-1]. Found by walking the + * carrier, so an entry the hand-off dropped on its way in is simply absent. */ static void BrokerOrphan_ReconcileSent(MqttBroker* broker, const char* client_id, BrokerOutPub* sent) { diff --git a/src/mqtt_client.c b/src/mqtt_client.c index 6b84fe39c..0d8be0d51 100644 --- a/src/mqtt_client.c +++ b/src/mqtt_client.c @@ -937,21 +937,26 @@ static int MqttClient_Replay_NewConnSafe(MqttClient* client) return MQTT_CODE_SUCCESS; } -/* Discard the retained copies of a Session the server did not resume, keeping - * anything published on this connection: a later resume still owes that - * [MQTT-4.4.0-1]. Reports a lock failure, unlike the wrappers above: a skipped - * discard would replay the old Session's messages into this one. */ +/* Discard the retained copies of a Session the server did not resume. Anything + * published on this connection is kept only for a persistent Session, which is + * the one a later resume still owes a re-send [MQTT-4.4.0-1]; a Clean Start or + * a zero Session Expiry ends with this connection, so its entries go too. + * Reports a lock failure, unlike the wrappers above: a skipped discard would + * replay the old Session's messages into this one. */ static int MqttClient_Replay_ResetSafe(MqttClient* client) { int i; + byte keepThisConn; #ifdef WOLFMQTT_MULTITHREAD int rc = wm_SemLock(&client->lockClient); if (rc != MQTT_CODE_SUCCESS) { return rc; } #endif + keepThisConn = (client->flags & MQTT_CLIENT_FLAG_PERSISTENT_SESSION) ? + 1 : 0; for (i = 0; i < MQTT_MAX_REPLAY_MSGS; i++) { - if (client->replay[i].onThisConn) { + if (keepThisConn && client->replay[i].onThisConn) { continue; } /* Release the handshake hold. */ diff --git a/tests/test_broker_connect.c b/tests/test_broker_connect.c index 32d0141ed..ca0eec3bb 100644 --- a/tests/test_broker_connect.c +++ b/tests/test_broker_connect.c @@ -3087,6 +3087,78 @@ TEST(handoff_marks_sent_qos1_for_redelivery) MqttBroker_Stop(&broker); MqttBroker_Free(&broker); } + +/* The same hand-off, but the write only placed part of the PUBLISH. Those + * bytes still reached the subscriber, so the resume copy is a re-delivery and + * [MQTT-3.3.1-1] requires DUP, exactly as for a completed write. The drain + * would have set the flag below, but the entry is already gone. */ +TEST(handoff_marks_partly_sent_qos1_for_redelivery) +{ + MqttBroker broker; + MqttBrokerNet net; + BrokerClient* sub_bc; + BrokerOrphanSession* o; + int i; + static const byte connect_pub[] = { + 0x10, 0x0D, 0x00, 0x04, 'M', 'Q', 'T', 'T', 0x04, 0x02, 0x00, 0x3C, + 0x00, 0x01, 'P' + }; + static const byte connect_sub[] = { + 0x10, 0x0D, 0x00, 0x04, 'M', 'Q', 'T', 'T', 0x04, 0x00, 0x00, 0x3C, + 0x00, 0x01, 'S' + }; + static const byte subscribe_x[] = { + 0x82, 0x06, 0x00, 0x01, 0x00, 0x01, 'x', 0x01 + }; + static const byte publish_x[] = { + 0x32, 0x08, 0x00, 0x01, 'x', 0x00, 0x07, 'A', 'B', 'C' + }; + + install_mock_net(&net); + XMEMSET(&broker, 0, sizeof(broker)); + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttBroker_Init(&broker, &net)); + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttBroker_Start(&broker)); + + reset_mock_clients(2); + mock_client_input_append(0, connect_pub, sizeof(connect_pub)); + mock_client_input_append(1, connect_sub, sizeof(connect_sub)); + mock_client_input_append(1, subscribe_x, sizeof(subscribe_x)); + for (i = 0; i < 16; i++) { + (void)MqttBroker_Step(&broker); + } + sub_bc = find_broker_client(&broker, "S"); + ASSERT_NOT_NULL(sub_bc); + ASSERT_NULL(sub_bc->out_q_head); + + broker_test_watch_free(NULL); + g_handoff_broker = &broker; + g_handoff_client = sub_bc; + g_write_publish_hook = handoff_out_queue_mid_write; + /* One byte of the PUBLISH goes out, then the hand-off takes the queue. */ + g_clients[1].write_limit_on_publish = 1; + + mock_client_input_append(0, publish_x, sizeof(publish_x)); + (void)MqttBroker_Step(&broker); + + ASSERT_NULL(g_write_publish_hook); + + o = broker.orphan_sessions; + ASSERT_NOT_NULL(o); + ASSERT_EQ(1, o->out_q_count); + ASSERT_NOT_NULL(o->out_q_head); + ASSERT_EQ(0, broker_test_watched_frees()); + ASSERT_EQ(MQTT_QOS_1, (int)o->out_q_head->qos); + ASSERT_EQ(1, (int)o->out_q_head->retransmit_dup); + + ASSERT_EQ(0, sub_bc->out_q_count); + ASSERT_NULL(sub_bc->out_q_head); + + g_handoff_broker = NULL; + g_handoff_client = NULL; + broker_test_watch_free(NULL); + MqttBroker_Stop(&broker); + MqttBroker_Free(&broker); +} #endif /* WOLFMQTT_MAX_QOS >= 1 */ #ifdef WOLFMQTT_BROKER_WILL @@ -17120,6 +17192,7 @@ int main(int argc, char** argv) RUN_TEST(drain_stops_when_session_handoff_takes_queue); #if WOLFMQTT_MAX_QOS >= 1 RUN_TEST(handoff_marks_sent_qos1_for_redelivery); + RUN_TEST(handoff_marks_partly_sent_qos1_for_redelivery); #endif #ifdef WOLFMQTT_NONBLOCK RUN_TEST(outbound_queue_short_write_resumes_on_next_step); diff --git a/tests/test_mqtt_client.c b/tests/test_mqtt_client.c index a363779b8..1e9007788 100644 --- a/tests/test_mqtt_client.c +++ b/tests/test_mqtt_client.c @@ -3888,6 +3888,8 @@ TEST(net_connect_clears_connect_sent_flag) #if !defined(WOLFMQTT_NO_SESSION_REPLAY) && WOLFMQTT_MAX_QOS >= 1 /* Reconnect against a CONNACK carrying the given Session Present bit and * return the result, so a test can observe what the replay put on the wire. */ +static byte g_connect_clean_session; /* CleanSession for the helpers below */ + static int run_reconnect_with(MqttConnect* connect, int session_present, MqttNetWriteCb write_cb) { @@ -3920,7 +3922,7 @@ static int run_reconnect_with(MqttConnect* connect, int session_present, XMEMSET(connect, 0, sizeof(*connect)); connect->keep_alive_sec = 60; - connect->clean_session = 0; + connect->clean_session = g_connect_clean_session; connect->client_id = "replay_client"; rc = MQTT_CODE_CONTINUE; @@ -4117,6 +4119,56 @@ TEST(fresh_session_reset_keeps_publish_from_this_connection) ASSERT_STR_EQ("new/1", test_client.replay[1].topic); } +/* The same reset for a CONNECT that asked for no persistent Session: it ends + * with this Network Connection, so nothing is owed a re-send [MQTT-4.4.0-1] + * and this connection's own entries go with the old Session's. Keeping them + * would replay them into a later Session that never held them. */ +TEST(fresh_session_reset_drops_publish_when_not_persistent) +{ + int rc; + int i; + int held; + MqttConnect connect; + MqttPublish publish; + static byte payload[] = "hello"; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); +#ifdef WOLFMQTT_V5 + test_client.protocol_level = MQTT_CONNECT_PROTOCOL_LEVEL_4; +#endif + ASSERT_EQ(MQTT_CODE_SUCCESS, run_initial_connect(&connect)); + + init_qos_publish(&publish, MQTT_QOS_1, 0x1234, "sensor/temp", + payload, (word32)(sizeof(payload) - 1)); + rc = run_publish_unacked(&publish); + ASSERT_EQ(MQTT_CODE_ERROR_NETWORK, rc); + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttClient_NetDisconnect(&test_client)); + ASSERT_EQ(0x1234, (int)test_client.replay[0].packet_id); + + g_seed_slot = 1; + g_seed_packet_id = 0x5678; + g_seed_on_this_conn = 1; + g_seed_reserve_ack_type = MQTT_PACKET_TYPE_PUBLISH_ACK; + g_connect_clean_session = 1; + rc = run_reconnect_with(&connect, 0, mock_net_write_seed_publish); + g_connect_clean_session = 0; + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + ASSERT_EQ(-1, g_seed_slot); + + /* Both are gone, and neither still occupies its identifier. */ + ASSERT_EQ(0, (int)test_client.replay[0].packet_id); + ASSERT_EQ(0, (int)test_client.replay[1].packet_id); + held = 0; + for (i = 0; i < MQTT_MAX_SEND_INFLIGHT; i++) { + if (test_client.send_inflight[i].packet_id != 0) { + held++; + } + } + ASSERT_EQ(0, held); + ASSERT_EQ(MQTT_MAX_REPLAY_MSGS, test_client.replayIdx); +} + /* A retained entry's Packet Identifier stays reserved across the handshake: * [MQTT-2.3.1-2] allows one use at a time and [MQTT-4.4.0-1] still owes it a * re-send. A fresh Session gives it back. */ @@ -10238,6 +10290,7 @@ void run_mqtt_client_tests(void) #endif RUN_TEST(reconnect_without_session_present_replays_nothing); RUN_TEST(fresh_session_reset_keeps_publish_from_this_connection); + RUN_TEST(fresh_session_reset_drops_publish_when_not_persistent); RUN_TEST(handshake_holds_retained_packet_id); RUN_TEST(resume_does_not_replay_publish_from_this_connection); RUN_TEST(resume_keeps_retained_packet_id_reserved); From d5a6136ecfb375b461d1fa0ad051827ad74e8666 Mon Sep 17 00:00:00 2001 From: Kareem Date: Mon, 26 Oct 2026 17:22:17 -0700 Subject: [PATCH 10/11] Code review feedback --- src/mqtt_client.c | 29 ++++++++++++++------------ tests/test_mqtt_client.c | 44 ++++++++++++++++++++++++++++++---------- 2 files changed, 49 insertions(+), 24 deletions(-) diff --git a/src/mqtt_client.c b/src/mqtt_client.c index 0d8be0d51..b066b844b 100644 --- a/src/mqtt_client.c +++ b/src/mqtt_client.c @@ -909,23 +909,30 @@ static MqttPacketType MqttClient_Replay_AckType(const MqttReplayMsg* slot) } /* Everything retained so far belongs to the Session that is ending; a send - * admitted once CONNECT is on the wire tags itself. Reports a lock failure: - * stale tags would have the handshake skip or keep the wrong entries. */ + * admitted once CONNECT is on the wire tags itself. The persistence flag still + * describes that Session, and one that ended with its connection owes nothing + * [MQTT-4.4.0-1], so its entries go. Reports a lock failure: stale tags would + * have the handshake skip or keep the wrong entries. */ static int MqttClient_Replay_NewConnSafe(MqttClient* client) { int i; + byte persisted; #ifdef WOLFMQTT_MULTITHREAD int rc = wm_SemLock(&client->lockClient); if (rc != MQTT_CODE_SUCCESS) { return rc; } #endif + persisted = (client->flags & MQTT_CLIENT_FLAG_PERSISTENT_SESSION) ? 1 : 0; for (i = 0; i < MQTT_MAX_REPLAY_MSGS; i++) { client->replay[i].onThisConn = 0; + if (!persisted) { + MqttClient_Replay_FreeSlot(&client->replay[i]); + } /* Hold for the handshake: [MQTT-2.3.1-2] allows one use at a time * and [MQTT-4.4.0-1] still owes these a re-send. Released below on a * fresh Session. */ - if (client->replay[i].packet_id != 0) { + else if (client->replay[i].packet_id != 0) { (void)MqttClient_SendIdReserve_Locked(client, client->replay[i].packet_id, &client->replay[i], 1, MqttClient_Replay_AckType(&client->replay[i])); @@ -937,26 +944,22 @@ static int MqttClient_Replay_NewConnSafe(MqttClient* client) return MQTT_CODE_SUCCESS; } -/* Discard the retained copies of a Session the server did not resume. Anything - * published on this connection is kept only for a persistent Session, which is - * the one a later resume still owes a re-send [MQTT-4.4.0-1]; a Clean Start or - * a zero Session Expiry ends with this connection, so its entries go too. - * Reports a lock failure, unlike the wrappers above: a skipped discard would - * replay the old Session's messages into this one. */ +/* Discard the retained copies of a Session the server did not resume, keeping + * anything published on this connection: its exchange is still live here, so + * its identifier stays in use [MQTT-2.3.1-2]. Reports a lock failure, unlike + * the wrappers above: a skipped discard would replay the old Session's + * messages into this one. */ static int MqttClient_Replay_ResetSafe(MqttClient* client) { int i; - byte keepThisConn; #ifdef WOLFMQTT_MULTITHREAD int rc = wm_SemLock(&client->lockClient); if (rc != MQTT_CODE_SUCCESS) { return rc; } #endif - keepThisConn = (client->flags & MQTT_CLIENT_FLAG_PERSISTENT_SESSION) ? - 1 : 0; for (i = 0; i < MQTT_MAX_REPLAY_MSGS; i++) { - if (keepThisConn && client->replay[i].onThisConn) { + if (client->replay[i].onThisConn) { continue; } /* Release the handshake hold. */ diff --git a/tests/test_mqtt_client.c b/tests/test_mqtt_client.c index 1e9007788..5fbf81e49 100644 --- a/tests/test_mqtt_client.c +++ b/tests/test_mqtt_client.c @@ -4072,9 +4072,14 @@ static int mock_net_write_seed_publish(void *context, const byte* buf, (buf[0] >> 4) == MQTT_PACKET_TYPE_CONNECT) { seed_replay_slot(g_seed_slot, g_seed_packet_id, g_seed_on_this_conn); if (g_seed_reserve_ack_type != 0) { - test_client.send_inflight[0].packet_id = g_seed_packet_id; - test_client.send_inflight[0].ack_type = g_seed_reserve_ack_type; - test_client.send_inflight[0].owner = &test_client.msg; + int n = 0; + while (n < MQTT_MAX_SEND_INFLIGHT - 1 && + test_client.send_inflight[n].packet_id != 0) { + n++; + } + test_client.send_inflight[n].packet_id = g_seed_packet_id; + test_client.send_inflight[n].ack_type = g_seed_reserve_ack_type; + test_client.send_inflight[n].owner = &test_client.msg; } g_seed_slot = -1; } @@ -4119,11 +4124,12 @@ TEST(fresh_session_reset_keeps_publish_from_this_connection) ASSERT_STR_EQ("new/1", test_client.replay[1].topic); } -/* The same reset for a CONNECT that asked for no persistent Session: it ends - * with this Network Connection, so nothing is owed a re-send [MQTT-4.4.0-1] - * and this connection's own entries go with the old Session's. Keeping them - * would replay them into a later Session that never held them. */ -TEST(fresh_session_reset_drops_publish_when_not_persistent) +/* A CONNECT that asked for no persistent Session: a publish sent on it is still + * in flight on this connection, so the fresh-Session reset keeps it and its + * identifier [MQTT-2.3.1-2]. That Session ends with the connection, though, so + * the next CONNECT drops it rather than holding or re-sending it + * [MQTT-4.4.0-1], even if the server wrongly claims a resume. */ +TEST(nonpersistent_publish_kept_until_next_connect) { int rc; int i; @@ -4156,8 +4162,25 @@ TEST(fresh_session_reset_drops_publish_when_not_persistent) ASSERT_EQ(MQTT_CODE_SUCCESS, rc); ASSERT_EQ(-1, g_seed_slot); - /* Both are gone, and neither still occupies its identifier. */ + /* The old Session's entry is gone; this connection's is still live. */ ASSERT_EQ(0, (int)test_client.replay[0].packet_id); + ASSERT_EQ(0x5678, (int)test_client.replay[1].packet_id); + held = 0; + for (i = 0; i < MQTT_MAX_SEND_INFLIGHT; i++) { + if (test_client.send_inflight[i].packet_id == 0x5678) { + held++; + } + } + ASSERT_EQ(1, held); + + /* The connection drops before its PUBACK. */ + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttClient_NetDisconnect(&test_client)); + g_hold_probe_id = -1; + rc = run_reconnect_with(&connect, 1, mock_net_write_accept); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + + /* Nothing held during the handshake, and nothing left to replay. */ + ASSERT_EQ(0, g_hold_probe_id); ASSERT_EQ(0, (int)test_client.replay[1].packet_id); held = 0; for (i = 0; i < MQTT_MAX_SEND_INFLIGHT; i++) { @@ -4166,7 +4189,6 @@ TEST(fresh_session_reset_drops_publish_when_not_persistent) } } ASSERT_EQ(0, held); - ASSERT_EQ(MQTT_MAX_REPLAY_MSGS, test_client.replayIdx); } /* A retained entry's Packet Identifier stays reserved across the handshake: @@ -10290,7 +10312,7 @@ void run_mqtt_client_tests(void) #endif RUN_TEST(reconnect_without_session_present_replays_nothing); RUN_TEST(fresh_session_reset_keeps_publish_from_this_connection); - RUN_TEST(fresh_session_reset_drops_publish_when_not_persistent); + RUN_TEST(nonpersistent_publish_kept_until_next_connect); RUN_TEST(handshake_holds_retained_packet_id); RUN_TEST(resume_does_not_replay_publish_from_this_connection); RUN_TEST(resume_keeps_retained_packet_id_reserved); From 34c37415b81ee36855199b428d241ae9db82f24f Mon Sep 17 00:00:00 2001 From: Kareem Date: Wed, 28 Oct 2026 12:44:09 -0700 Subject: [PATCH 11/11] Only clear entry in Drop_Locked if owner matches. Add tests for this. Add appropriate WOLFMQTT_TEST_BROKER_ALLOC_H guards for tests. --- src/mqtt_client.c | 15 +++-- tests/test_mqtt_client.c | 115 ++++++++++++++++++++++++++++++++++++++- 2 files changed, 121 insertions(+), 9 deletions(-) diff --git a/src/mqtt_client.c b/src/mqtt_client.c index b066b844b..77027bea9 100644 --- a/src/mqtt_client.c +++ b/src/mqtt_client.c @@ -92,7 +92,7 @@ static int MqttClient_CancelMessage(MqttClient *client, MqttObject* msg); #endif #ifndef WOLFMQTT_NO_SESSION_REPLAY static void MqttClient_SendIds_Drop_Locked(MqttClient* client, - word16 packet_id); + word16 packet_id, const void* owner); static int MqttClient_SendIdReserve_Locked(MqttClient* client, word16 packet_id, void* owner, int isRetransmit, MqttPacketType ack_type); #endif @@ -963,7 +963,8 @@ static int MqttClient_Replay_ResetSafe(MqttClient* client) continue; } /* Release the handshake hold. */ - MqttClient_SendIds_Drop_Locked(client, client->replay[i].packet_id); + MqttClient_SendIds_Drop_Locked(client, client->replay[i].packet_id, + &client->replay[i]); MqttClient_Replay_FreeSlot(&client->replay[i]); } client->replayIdx = MQTT_MAX_REPLAY_MSGS; @@ -1115,8 +1116,10 @@ static int MqttClient_SendIdReserve(MqttClient* client, word16 packet_id, * exchange's replay state. */ #ifndef WOLFMQTT_NO_SESSION_REPLAY /* Caller holds client->lockClient. Unlike MqttClient_SendIdRelease, drops only - * the reservation; the caller is already disposing of the replay entry. */ -static void MqttClient_SendIds_Drop_Locked(MqttClient* client, word16 packet_id) + * the reservation, and only while owner still holds it: a re-send may have + * taken it over. The caller is already disposing of the replay entry. */ +static void MqttClient_SendIds_Drop_Locked(MqttClient* client, word16 packet_id, + const void* owner) { int i; @@ -1124,7 +1127,7 @@ static void MqttClient_SendIds_Drop_Locked(MqttClient* client, word16 packet_id) return; } i = MqttClient_SendIds_Find(client, packet_id); - if (i >= 0) { + if (i >= 0 && client->send_inflight[i].owner == owner) { client->send_inflight[i].packet_id = 0; client->send_inflight[i].owner = NULL; client->send_inflight[i].ack_type = (byte)MQTT_PACKET_TYPE_RESERVED; @@ -4140,7 +4143,7 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect) * honoured: drop it and release the identifier, which no * acknowledgement would free [MQTT-2.3.1-3]. */ MqttClient_SendIds_Drop_Locked(client, - client->replay[i].packet_id); + client->replay[i].packet_id, &client->replay[i]); MqttClient_Replay_FreeSlot(&client->replay[i]); } } diff --git a/tests/test_mqtt_client.c b/tests/test_mqtt_client.c index 5fbf81e49..a880843e7 100644 --- a/tests/test_mqtt_client.c +++ b/tests/test_mqtt_client.c @@ -49,7 +49,10 @@ static byte test_tx_buf[TEST_TX_BUF_SIZE]; static byte test_rx_buf[TEST_RX_BUF_SIZE]; /* The library allocator is routed through here (tests/test_broker_alloc.h) so - * a test can watch one allocation and record the state it was released in. */ + * a test can watch one allocation and record the state it was released in. + * Only builds that force-include that header, and compile the client into the + * test, have the hook. */ +#ifdef WOLFMQTT_TEST_BROKER_ALLOC_H static void* g_free_watch_ptr; static int g_free_watch_count; static int g_free_watch_lock_count; @@ -85,6 +88,7 @@ UT_MAYBE_UNUSED static void test_client_watch_free(void* ptr) g_free_watch_count = 0; g_free_watch_lock_count = -1; } +#endif /* WOLFMQTT_TEST_BROKER_ALLOC_H */ /* Mock network callbacks - just return errors since we're not actually * connecting to anything */ @@ -4282,6 +4286,107 @@ TEST(resume_keeps_retained_packet_id_reserved) ASSERT_EQ(1, held); } +/* A send is admitted once CONNECT is on the wire, so a DUP re-send of a + * retained PUBLISH can take over the handshake's hold before it re-tags the + * entry. This does what MqttClient_SendIdReserve does for that re-send. */ +static MqttPublish g_takeover_publish; +static word16 g_takeover_id; + +static int mock_net_write_take_over_hold(void *context, const byte* buf, + int buf_len, int timeout_ms) +{ + int rc = mock_net_write_accept(context, buf, buf_len, timeout_ms); + int n; + + if (g_takeover_id != 0 && buf_len > 0 && + (buf[0] >> 4) == MQTT_PACKET_TYPE_CONNECT) { + for (n = 0; n < MQTT_MAX_SEND_INFLIGHT; n++) { + if (test_client.send_inflight[n].packet_id == g_takeover_id) { + test_client.send_inflight[n].owner = &g_takeover_publish; + } + } + g_takeover_id = 0; + } + return rc; +} + +static int count_held_by(word16 packet_id, const void* owner) +{ + int n; + int held = 0; + + for (n = 0; n < MQTT_MAX_SEND_INFLIGHT; n++) { + if (test_client.send_inflight[n].packet_id == packet_id && + test_client.send_inflight[n].owner == owner) { + held++; + } + } + return held; +} + +/* The fresh-Session reset gives back only the handshake's own hold. Once a + * re-send owns the identifier its exchange is live [MQTT-2.3.1-2]. */ +TEST(fresh_session_reset_keeps_hold_taken_by_resend) +{ + int rc; + MqttConnect connect; + MqttPublish publish; + static byte payload[] = "hello"; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); +#ifdef WOLFMQTT_V5 + test_client.protocol_level = MQTT_CONNECT_PROTOCOL_LEVEL_4; +#endif + ASSERT_EQ(MQTT_CODE_SUCCESS, run_initial_connect(&connect)); + + init_qos_publish(&publish, MQTT_QOS_1, 0x1234, "sensor/temp", + payload, (word32)(sizeof(payload) - 1)); + rc = run_publish_unacked(&publish); + ASSERT_EQ(MQTT_CODE_ERROR_NETWORK, rc); + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttClient_NetDisconnect(&test_client)); + + g_takeover_id = 0x1234; + rc = run_reconnect_with(&connect, 0, mock_net_write_take_over_hold); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + ASSERT_EQ(0, g_takeover_id); + + ASSERT_EQ(0, (int)test_client.replay[0].packet_id); + ASSERT_EQ(1, count_held_by(0x1234, &g_takeover_publish)); +} + +/* The same for the resumed walk, which drops an entry it has no copy of. */ +TEST(resume_drop_keeps_hold_taken_by_resend) +{ + int rc; + MqttConnect connect; + MqttPublish publish; + static byte payload[] = "hello"; + + rc = test_init_client(); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); +#ifdef WOLFMQTT_V5 + test_client.protocol_level = MQTT_CONNECT_PROTOCOL_LEVEL_4; +#endif + ASSERT_EQ(MQTT_CODE_SUCCESS, run_initial_connect(&connect)); + + init_qos_publish(&publish, MQTT_QOS_1, 0x1234, "sensor/temp", + payload, (word32)(sizeof(payload) - 1)); + rc = run_publish_unacked(&publish); + ASSERT_EQ(MQTT_CODE_ERROR_NETWORK, rc); + ASSERT_EQ(MQTT_CODE_SUCCESS, MqttClient_NetDisconnect(&test_client)); + ASSERT_EQ(0x1234, (int)test_client.replay[0].packet_id); + test_client.replay[0].haveCopy = 0; + + g_takeover_id = 0x1234; + rc = run_reconnect_with(&connect, 1, mock_net_write_take_over_hold); + ASSERT_EQ(MQTT_CODE_SUCCESS, rc); + ASSERT_EQ(0, g_takeover_id); + + ASSERT_EQ(0, (int)test_client.replay[0].packet_id); + ASSERT_EQ(1, count_held_by(0x1234, &g_takeover_publish)); +} + /* The resumed walk re-sends only what the previous connection left * unacknowledged; an entry published after CONNECT is already in flight. */ TEST(resume_does_not_replay_publish_from_this_connection) @@ -4314,7 +4419,8 @@ TEST(resume_does_not_replay_publish_from_this_connection) #if defined(WOLFMQTT_MULTITHREAD) && defined(WOLFMQTT_POSIX_SEMAPHORES) && \ - !defined(WOLFMQTT_NO_COND_SIGNAL) && !defined(WOLFMQTT_STATIC_MEMORY) + !defined(WOLFMQTT_NO_COND_SIGNAL) && !defined(WOLFMQTT_STATIC_MEMORY) && \ + defined(WOLFMQTT_TEST_BROKER_ALLOC_H) /* A send thread reaches the same replay slots under the client lock, and both * sides release what they find there, so the discard has to hold it too. The * POSIX lock keeps its own depth; the allocator hook samples it as the slot's @@ -10316,8 +10422,11 @@ void run_mqtt_client_tests(void) RUN_TEST(handshake_holds_retained_packet_id); RUN_TEST(resume_does_not_replay_publish_from_this_connection); RUN_TEST(resume_keeps_retained_packet_id_reserved); + RUN_TEST(fresh_session_reset_keeps_hold_taken_by_resend); + RUN_TEST(resume_drop_keeps_hold_taken_by_resend); #if defined(WOLFMQTT_MULTITHREAD) && defined(WOLFMQTT_POSIX_SEMAPHORES) && \ - !defined(WOLFMQTT_NO_COND_SIGNAL) && !defined(WOLFMQTT_STATIC_MEMORY) + !defined(WOLFMQTT_NO_COND_SIGNAL) && !defined(WOLFMQTT_STATIC_MEMORY) && \ + defined(WOLFMQTT_TEST_BROKER_ALLOC_H) RUN_TEST(fresh_session_reset_frees_replay_under_client_lock); #endif RUN_TEST(reconnect_does_not_replay_acked_publish);