From fe10546024852637053fa031040ce868a6aaec59 Mon Sep 17 00:00:00 2001 From: Martin Mirchev Date: Mon, 2 Nov 2026 19:11:42 +0800 Subject: [PATCH 1/2] Broker: close static retained delivery after a short write In WOLFMQTT_STATIC_MEMORY builds BrokerRetained_DeliverToClient encodes each matching retained PUBLISH into bc->tx_buf and writes it directly. When a write stops part way (CONTINUE, or a select() timeout in BrokerPosix_Write), MqttSocket_Write leaves client.write.pos at the bytes already sent. The loop then encodes the next retained message into the same tx_buf, and the next write resumes at that offset inside the new packet. The subscriber receives the head of one PUBLISH joined to the tail of another; with different payload lengths it loses framing for the rest of the connection. Static builds keep nothing to resume a cut frame from. Stop the delivery once part of a frame is on the wire and fail the SUBSCRIBE, so the connection is closed the same way BrokerHandle_Publish handles a short fan-out write. A write that fails before sending any bytes keeps the existing behaviour. Add a mock mode that short-writes and then times out with the socket still open, and a test that fails without this change (15 bytes reach the subscriber instead of 3: the cut PUBLISH, the tail of the next one and the SUBACK). --- src/mqtt_broker.c | 15 ++++++++ tests/test_broker_connect.c | 77 +++++++++++++++++++++++++++++++++++++ 2 files changed, 92 insertions(+) diff --git a/src/mqtt_broker.c b/src/mqtt_broker.c index 50ea1441e..7fbd25f2f 100644 --- a/src/mqtt_broker.c +++ b/src/mqtt_broker.c @@ -6367,6 +6367,21 @@ static int BrokerRetained_DeliverToClient(MqttBroker* broker, * slot would leak it [MQTT-2.3.1-3]. */ BrokerStaticOutId_Release(bc, out_pub.packet_id, eff_qos); } + if (wr_rc != enc_rc && bc->client.write.pos != 0) { + /* Part of the PUBLISH is on the wire (a short write that + * then returned CONTINUE, timed out or failed). Static + * memory keeps nothing to resume it from, and the next + * write would continue at client.write.pos inside a + * different packet, splicing the two frames. Stop and + * fail the SUBSCRIBE so the connection is closed, as + * BrokerHandle_Publish does for a short fan-out write. */ + if (wr_rc == MQTT_CODE_CONTINUE && eff_qos >= MQTT_QOS_1) { + BrokerStaticOutId_Release(bc, out_pub.packet_id, + eff_qos); + } + deliver_rc = MQTT_CODE_ERROR_NETWORK; + break; + } } else { /* The PUBLISH never reached the wire, so no PUBACK or diff --git a/tests/test_broker_connect.c b/tests/test_broker_connect.c index ca0eec3bb..c67127869 100644 --- a/tests/test_broker_connect.c +++ b/tests/test_broker_connect.c @@ -219,6 +219,10 @@ typedef struct MockClient { * a sequence the one-shot knobs above cannot express on their own. */ int write_limit_then_err; int write_err_pending; /* next mock_write call returns a network error */ + /* positive: short-write the next callback to this many bytes and arm + * write_timeout_pending: part of a packet goes out, then select() times + * out while the socket stays alive (BrokerPosix_Write's timeout path). */ + int write_limit_then_timeout; /* When set, the next PUBLISH frame (and only that) fails. Lets a test aim * at a delivery without also failing the SUBACK that precedes it. */ int write_err_on_publish; @@ -365,6 +369,13 @@ static int mock_write(void* ctx, BROKER_SOCKET_T sock, mc->write_limit_then_err = 0; mc->write_err_pending = 1; } + if (mc->write_limit_then_timeout > 0) { + if (write_len > mc->write_limit_then_timeout) { + write_len = mc->write_limit_then_timeout; + } + mc->write_limit_then_timeout = 0; + mc->write_timeout_pending = 1; + } if (mc->write_limit_once > 0) { if (write_len > mc->write_limit_once) { write_len = mc->write_limit_once; @@ -5478,6 +5489,69 @@ TEST(retained_short_write_preserves_following_delivery) #endif /* WOLFMQTT_NONBLOCK */ #endif /* WOLFMQTT_BROKER_RETAINED && !WOLFMQTT_STATIC_MEMORY */ +#if defined(WOLFMQTT_BROKER_RETAINED) && defined(WOLFMQTT_STATIC_MEMORY) +/* Static builds cannot resume a cut direct write. When a retained PUBLISH + * stops part way, nothing else may follow it on the wire: the next retained + * PUBLISH used to be written from the cut offset, so the subscriber read the + * head of r/a joined to the tail of r/b. The connection must close instead. */ +TEST(static_retained_short_write_closes_without_splicing) +{ + MqttBroker broker; + MqttBrokerNet net; + size_t sub_before; + 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, 0x02, 0x00, 0x3C, 0x00, 0x01, 'S' + }; + /* Hand-built retained QoS 0 PUBLISH packets for r/a and r/b. */ + static const byte publish_one[] = { + 0x31, 0x08, 0x00, 0x03, 'r', '/', 'a', 'O', 'N', 'E' + }; + static const byte publish_two[] = { + 0x31, 0x08, 0x00, 0x03, 'r', '/', 'b', 'T', 'W', 'O' + }; + static const byte subscribe_all[] = { + 0x82, 0x08, 0x00, 0x01, 0x00, 0x03, 'r', '/', '#', 0x00 + }; + + 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(0, publish_one, sizeof(publish_one)); + mock_client_input_append(0, publish_two, sizeof(publish_two)); + mock_client_input_append(1, connect_sub, sizeof(connect_sub)); + for (i = 0; i < 12; i++) { + (void)MqttBroker_Step(&broker); + } + ASSERT_FALSE(g_clients[1].closed); + + /* The first retained PUBLISH gets 3 bytes out, then the write times out + * with the socket still open. */ + sub_before = g_clients[1].out_len; + g_clients[1].write_limit_then_timeout = 3; + mock_client_input_append(1, subscribe_all, sizeof(subscribe_all)); + for (i = 0; i < 4; i++) { + (void)MqttBroker_Step(&broker); + } + + ASSERT_EQ(3, (int)(g_clients[1].out_len - sub_before)); + ASSERT_MEM_EQ(publish_one, g_clients[1].out_buf + sub_before, 3); + ASSERT_TRUE(g_clients[1].closed); + + MqttBroker_Stop(&broker); + MqttBroker_Free(&broker); +} +#endif /* WOLFMQTT_BROKER_RETAINED && WOLFMQTT_STATIC_MEMORY */ + #if defined(WOLFMQTT_BROKER_WILL) && !defined(WOLFMQTT_STATIC_MEMORY) /* The immediate-will fan-out must scrub the subscriber tx_buf even when the * delivery write fails hard (only an in-progress MQTT_CODE_CONTINUE is @@ -17332,6 +17406,9 @@ int main(int argc, char** argv) RUN_TEST(retained_short_write_preserves_following_delivery); #endif #endif +#if defined(WOLFMQTT_BROKER_RETAINED) && defined(WOLFMQTT_STATIC_MEMORY) + RUN_TEST(static_retained_short_write_closes_without_splicing); +#endif /* Both allocation profiles: the retained store and its expiry paths have * separate implementations, so each has to be swept correctly. */ #if defined(WOLFMQTT_BROKER_PERSIST) && defined(WOLFMQTT_BROKER_RETAINED) && \ From 92b94743fe5925864503d3f27c592db3632b541f Mon Sep 17 00:00:00 2001 From: Martin Mirchev Date: Mon, 2 Nov 2026 19:15:42 +0800 Subject: [PATCH 2/2] Broker: close the transport before the Will after a short retained write Returning MQTT_CODE_ERROR_NETWORK alone left the socket open until the dispatcher's BrokerClient_AbnormalClose, which publishes the client's Will first. If the client is subscribed to its own Will topic, the Will fan-out wrote into tx_buf at the stale client.write.pos and spliced the frame again. Call BrokerStaticClient_Close before failing the SUBSCRIBE, as the queue-full branch of the same loop does; the closed client is no longer connected, so the Will fan-out skips it. Tests: use exact-topic filters so they also run with --disable-broker-wildcards (r/# was rejected there and the short write cut the SUBACK instead), and add static_retained_short_write_own_will_not_spliced, which fails without this change. --- src/mqtt_broker.c | 9 ++- tests/test_broker_connect.c | 132 ++++++++++++++++++++++++++++-------- 2 files changed, 111 insertions(+), 30 deletions(-) diff --git a/src/mqtt_broker.c b/src/mqtt_broker.c index 7fbd25f2f..9fb3913de 100644 --- a/src/mqtt_broker.c +++ b/src/mqtt_broker.c @@ -6372,13 +6372,16 @@ static int BrokerRetained_DeliverToClient(MqttBroker* broker, * then returned CONTINUE, timed out or failed). Static * memory keeps nothing to resume it from, and the next * write would continue at client.write.pos inside a - * different packet, splicing the two frames. Stop and - * fail the SUBSCRIBE so the connection is closed, as - * BrokerHandle_Publish does for a short fan-out write. */ + * different packet, splicing the two frames. Close the + * transport now, as BrokerHandle_Publish does for a short + * fan-out write, so nothing else (such as this client's + * own Will, published when the failed SUBSCRIBE closes + * the connection) is written after the cut frame. */ if (wr_rc == MQTT_CODE_CONTINUE && eff_qos >= MQTT_QOS_1) { BrokerStaticOutId_Release(bc, out_pub.packet_id, eff_qos); } + BrokerStaticClient_Close(broker, bc); deliver_rc = MQTT_CODE_ERROR_NETWORK; break; } diff --git a/tests/test_broker_connect.c b/tests/test_broker_connect.c index c67127869..859e77eab 100644 --- a/tests/test_broker_connect.c +++ b/tests/test_broker_connect.c @@ -5490,6 +5490,52 @@ TEST(retained_short_write_preserves_following_delivery) #endif /* WOLFMQTT_BROKER_RETAINED && !WOLFMQTT_STATIC_MEMORY */ #if defined(WOLFMQTT_BROKER_RETAINED) && defined(WOLFMQTT_STATIC_MEMORY) +/* Exact-topic filters keep these tests valid with + * --disable-broker-wildcards. */ +static const byte static_short_write_connect_pub[] = { + 0x10, 0x0D, 0x00, 0x04, 'M', 'Q', 'T', 'T', + 0x04, 0x02, 0x00, 0x3C, 0x00, 0x01, 'P' +}; +/* Hand-built retained QoS 0 PUBLISH packets for r/a and r/b. */ +static const byte static_short_write_publish_a[] = { + 0x31, 0x08, 0x00, 0x03, 'r', '/', 'a', 'O', 'N', 'E' +}; +static const byte static_short_write_publish_b[] = { + 0x31, 0x08, 0x00, 0x03, 'r', '/', 'b', 'T', 'W', 'O' +}; + +/* Connect "P", store the retained messages, then connect the subscriber. */ +static int static_short_write_setup(MqttBroker* broker, MqttBrokerNet* net, + const byte* connect_sub, size_t connect_sub_len) +{ + int i; + int rc; + + install_mock_net(net); + XMEMSET(broker, 0, sizeof(*broker)); + rc = MqttBroker_Init(broker, net); + if (rc != MQTT_CODE_SUCCESS) { + return rc; + } + rc = MqttBroker_Start(broker); + if (rc != MQTT_CODE_SUCCESS) { + return rc; + } + + reset_mock_clients(2); + mock_client_input_append(0, static_short_write_connect_pub, + sizeof(static_short_write_connect_pub)); + mock_client_input_append(0, static_short_write_publish_a, + sizeof(static_short_write_publish_a)); + mock_client_input_append(0, static_short_write_publish_b, + sizeof(static_short_write_publish_b)); + mock_client_input_append(1, connect_sub, connect_sub_len); + for (i = 0; i < 12; i++) { + (void)MqttBroker_Step(broker); + } + return MQTT_CODE_SUCCESS; +} + /* Static builds cannot resume a cut direct write. When a retained PUBLISH * stops part way, nothing else may follow it on the wire: the next retained * PUBLISH used to be written from the cut offset, so the subscriber read the @@ -5500,56 +5546,85 @@ TEST(static_retained_short_write_closes_without_splicing) MqttBrokerNet net; size_t sub_before; 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, 0x02, 0x00, 0x3C, 0x00, 0x01, 'S' }; - /* Hand-built retained QoS 0 PUBLISH packets for r/a and r/b. */ - static const byte publish_one[] = { - 0x31, 0x08, 0x00, 0x03, 'r', '/', 'a', 'O', 'N', 'E' - }; - static const byte publish_two[] = { - 0x31, 0x08, 0x00, 0x03, 'r', '/', 'b', 'T', 'W', 'O' - }; - static const byte subscribe_all[] = { - 0x82, 0x08, 0x00, 0x01, 0x00, 0x03, 'r', '/', '#', 0x00 + /* SUBSCRIBE r/a and r/b at QoS 0. */ + static const byte subscribe_ab[] = { + 0x82, 0x0E, 0x00, 0x01, + 0x00, 0x03, 'r', '/', 'a', 0x00, + 0x00, 0x03, 'r', '/', 'b', 0x00 }; - 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)); + ASSERT_EQ(MQTT_CODE_SUCCESS, static_short_write_setup(&broker, &net, + connect_sub, sizeof(connect_sub))); + ASSERT_FALSE(g_clients[1].closed); - reset_mock_clients(2); - mock_client_input_append(0, connect_pub, sizeof(connect_pub)); - mock_client_input_append(0, publish_one, sizeof(publish_one)); - mock_client_input_append(0, publish_two, sizeof(publish_two)); - mock_client_input_append(1, connect_sub, sizeof(connect_sub)); - for (i = 0; i < 12; i++) { + /* The first retained PUBLISH gets 3 bytes out, then the write times out + * with the socket still open. */ + sub_before = g_clients[1].out_len; + g_clients[1].write_limit_then_timeout = 3; + mock_client_input_append(1, subscribe_ab, sizeof(subscribe_ab)); + for (i = 0; i < 4; i++) { (void)MqttBroker_Step(&broker); } + + ASSERT_EQ(3, (int)(g_clients[1].out_len - sub_before)); + ASSERT_MEM_EQ(static_short_write_publish_a, + g_clients[1].out_buf + sub_before, 3); + ASSERT_TRUE(g_clients[1].closed); + + MqttBroker_Stop(&broker); + MqttBroker_Free(&broker); +} + +#ifdef WOLFMQTT_BROKER_WILL +/* Closing that connection publishes the subscriber's Will. When the + * subscriber is subscribed to its own Will topic, the Will must not be + * written after the cut retained PUBLISH either. */ +TEST(static_retained_short_write_own_will_not_spliced) +{ + MqttBroker broker; + MqttBrokerNet net; + size_t sub_before; + int i; + /* v3.1.1 CONNECT "S", flags 0x06 (will | clean), Will topic "r/w", + * Will payload "BYE". */ + static const byte connect_sub[] = { + 0x10, 0x17, 0x00, 0x04, 'M', 'Q', 'T', 'T', + 0x04, 0x06, 0x00, 0x3C, 0x00, 0x01, 'S', + 0x00, 0x03, 'r', '/', 'w', + 0x00, 0x03, 'B', 'Y', 'E' + }; + /* SUBSCRIBE r/w (its own Will topic) then r/a (retained) at QoS 0. + * Filters are added in order, so r/w is in place when r/a is cut. */ + static const byte subscribe_wa[] = { + 0x82, 0x0E, 0x00, 0x01, + 0x00, 0x03, 'r', '/', 'w', 0x00, + 0x00, 0x03, 'r', '/', 'a', 0x00 + }; + + ASSERT_EQ(MQTT_CODE_SUCCESS, static_short_write_setup(&broker, &net, + connect_sub, sizeof(connect_sub))); ASSERT_FALSE(g_clients[1].closed); - /* The first retained PUBLISH gets 3 bytes out, then the write times out - * with the socket still open. */ sub_before = g_clients[1].out_len; g_clients[1].write_limit_then_timeout = 3; - mock_client_input_append(1, subscribe_all, sizeof(subscribe_all)); + mock_client_input_append(1, subscribe_wa, sizeof(subscribe_wa)); for (i = 0; i < 4; i++) { (void)MqttBroker_Step(&broker); } ASSERT_EQ(3, (int)(g_clients[1].out_len - sub_before)); - ASSERT_MEM_EQ(publish_one, g_clients[1].out_buf + sub_before, 3); + ASSERT_MEM_EQ(static_short_write_publish_a, + g_clients[1].out_buf + sub_before, 3); ASSERT_TRUE(g_clients[1].closed); MqttBroker_Stop(&broker); MqttBroker_Free(&broker); } +#endif /* WOLFMQTT_BROKER_WILL */ #endif /* WOLFMQTT_BROKER_RETAINED && WOLFMQTT_STATIC_MEMORY */ #if defined(WOLFMQTT_BROKER_WILL) && !defined(WOLFMQTT_STATIC_MEMORY) @@ -17408,6 +17483,9 @@ int main(int argc, char** argv) #endif #if defined(WOLFMQTT_BROKER_RETAINED) && defined(WOLFMQTT_STATIC_MEMORY) RUN_TEST(static_retained_short_write_closes_without_splicing); +#ifdef WOLFMQTT_BROKER_WILL + RUN_TEST(static_retained_short_write_own_will_not_spliced); +#endif #endif /* Both allocation profiles: the retained store and its expiry paths have * separate implementations, so each has to be swept correctly. */