diff --git a/src/mqtt_broker.c b/src/mqtt_broker.c index 50ea1441e..9fb3913de 100644 --- a/src/mqtt_broker.c +++ b/src/mqtt_broker.c @@ -6367,6 +6367,24 @@ 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. 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; + } } 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..859e77eab 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,144 @@ 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) +/* 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 + * 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_sub[] = { + 0x10, 0x0D, 0x00, 0x04, 'M', 'Q', 'T', 'T', + 0x04, 0x02, 0x00, 0x3C, 0x00, 0x01, 'S' + }; + /* 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 + }; + + 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_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); + + sub_before = g_clients[1].out_len; + g_clients[1].write_limit_then_timeout = 3; + 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(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) /* 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 +17481,12 @@ 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); +#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. */ #if defined(WOLFMQTT_BROKER_PERSIST) && defined(WOLFMQTT_BROKER_RETAINED) && \