Broker: close static retained delivery after a short write

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

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) && \