diff --git a/src/mqtt_broker.c b/src/mqtt_broker.c index 60c56e9..545da99 100644 --- a/src/mqtt_broker.c +++ b/src/mqtt_broker.c @@ -2191,6 +2191,14 @@ static int BrokerClient_DrainOutQueue(BrokerClient* bc) } bc->out_q_count--; BrokerOutPub_Free(free_me); + /* MqttSocket_Write resumes a partial send at + * client.write.pos inside the caller's buffer and only + * clears it after a completed write. The entry those bytes + * belonged to is gone, so leaving the offset set would make + * the next queued PUBLISH start mid-buffer and go out + * truncated. */ + bc->client.write.pos = 0; + bc->client.write.len = 0; } return (wr_rc < 0) ? wr_rc : MQTT_CODE_ERROR_NETWORK; } @@ -4407,14 +4415,19 @@ static int BrokerStaticOutId_InUse(const BrokerClient* bc, word16 packet_id) int i; for (i = 0; i < BROKER_MAX_INFLIGHT_PER_SUB; i++) { - if (bc->out_inflight[i] == packet_id) { + if (bc->out_inflight[i].packet_id == packet_id) { return 1; } } return 0; } -static void BrokerStaticOutId_Release(BrokerClient* bc, word16 packet_id) +/* Free the slot only when the acknowledgement matches the QoS the delivery + * went out with: MQTT 3.1.1 section 4.3.3 completes a QoS 2 delivery with + * PUBCOMP, so a PUBACK naming that identifier must not release it while + * PUBREC/PUBREL/PUBCOMP are still outstanding. */ +static void BrokerStaticOutId_Release(BrokerClient* bc, word16 packet_id, + MqttQoS qos) { int i; @@ -4422,8 +4435,10 @@ static void BrokerStaticOutId_Release(BrokerClient* bc, word16 packet_id) return; } for (i = 0; i < BROKER_MAX_INFLIGHT_PER_SUB; i++) { - if (bc->out_inflight[i] == packet_id) { - bc->out_inflight[i] = 0; + if (bc->out_inflight[i].packet_id == packet_id && + bc->out_inflight[i].qos == qos) { + bc->out_inflight[i].packet_id = 0; + bc->out_inflight[i].qos = MQTT_QOS_0; return; } } @@ -4433,14 +4448,15 @@ static void BrokerStaticOutId_Release(BrokerClient* bc, word16 packet_id) * Returns 0 when every value is outstanding or the table is full, which the * callers treat as "cannot deliver this QoS > 0 PUBLISH now" rather than * reusing an identifier the client has not acknowledged. */ -static word16 BrokerStaticOutId_Take(MqttBroker* broker, BrokerClient* bc) +static word16 BrokerStaticOutId_Take(MqttBroker* broker, BrokerClient* bc, + MqttQoS qos) { word16 candidate; word16 first; int i; for (i = 0; i < BROKER_MAX_INFLIGHT_PER_SUB; i++) { - if (bc->out_inflight[i] == 0) { + if (bc->out_inflight[i].packet_id == 0) { break; } } @@ -4459,7 +4475,8 @@ static word16 BrokerStaticOutId_Take(MqttBroker* broker, BrokerClient* bc) if (broker->next_packet_id == 0) { broker->next_packet_id = 1; } - bc->out_inflight[i] = candidate; + bc->out_inflight[i].packet_id = candidate; + bc->out_inflight[i].qos = qos; return candidate; } candidate++; @@ -5529,7 +5546,7 @@ static int BrokerRetained_DeliverToClient(MqttBroker* broker, out_pub.buffer = (rm->payload_len > 0) ? rm->payload : NULL; out_pub.total_len = rm->payload_len; if (eff_qos >= MQTT_QOS_1) { - out_pub.packet_id = BrokerStaticOutId_Take(broker, bc); + out_pub.packet_id = BrokerStaticOutId_Take(broker, bc, eff_qos); if (out_pub.packet_id == 0) { /* [MQTT-2.3.1-4] No identifier is free for this subscriber * because every one is still awaiting its acknowledgement. @@ -5563,6 +5580,12 @@ static int BrokerRetained_DeliverToClient(MqttBroker* broker, BROKER_FORCE_ZERO(bc->tx_buf, enc_rc); } } + else { + /* The PUBLISH never reached the wire, so no PUBACK or + * PUBCOMP will ever release the identifier. Give the slot back + * now [MQTT-2.3.1-3]. */ + BrokerStaticOutId_Release(bc, out_pub.packet_id, eff_qos); + } } } #else @@ -5854,7 +5877,7 @@ static void BrokerClient_PublishWillImmediate(MqttBroker* broker, out_pub.buffer = (payload_len > 0) ? (byte*)payload : NULL; out_pub.total_len = payload_len; if (eff_qos >= MQTT_QOS_1) { - out_pub.packet_id = BrokerStaticOutId_Take(broker, wc); + out_pub.packet_id = BrokerStaticOutId_Take(broker, wc, eff_qos); if (out_pub.packet_id == 0) { /* [MQTT-2.3.1-4] No identifier is free for this subscriber * because every one is still awaiting its acknowledgement. @@ -5877,6 +5900,12 @@ static void BrokerClient_PublishWillImmediate(MqttBroker* broker, BROKER_FORCE_ZERO(wc->tx_buf, enc_rc); } } + else { + /* The PUBLISH never reached the wire, so no PUBACK or + * PUBCOMP will ever release the identifier. Give the slot back + * now [MQTT-2.3.1-3]. */ + BrokerStaticOutId_Release(wc, out_pub.packet_id, eff_qos); + } } } else if ((sub->client == NULL || !sub->client->connected) && @@ -7725,7 +7754,7 @@ static int BrokerHandle_Publish(BrokerClient* bc, int rx_len, out_pub.qos = eff_qos; if (eff_qos >= MQTT_QOS_1) { out_pub.packet_id = BrokerStaticOutId_Take(broker, - sub->client); + sub->client, eff_qos); if (out_pub.packet_id == 0) { /* [MQTT-2.3.1-4] No identifier is free for this subscriber * because every one is still awaiting its acknowledgement. @@ -7765,6 +7794,11 @@ static int BrokerHandle_Publish(BrokerClient* bc, int rx_len, BROKER_FORCE_ZERO(sub->client->tx_buf, sub_rc); } else { + /* The PUBLISH never reached the wire, so no PUBACK or + * PUBCOMP will ever release the identifier. Give the slot back + * now [MQTT-2.3.1-3]. */ + BrokerStaticOutId_Release(sub->client, out_pub.packet_id, + eff_qos); WBLOG_ERR(broker, "broker: PUBLISH fwd encode failed " "sock=%d -> sock=%d rc=%d", @@ -8485,7 +8519,7 @@ static int BrokerClient_Process(MqttBroker* broker, BrokerClient* bc) #ifdef WOLFMQTT_STATIC_MEMORY /* [MQTT-2.3.1-3] PUBACK completes the QoS 1 exchange, so * the identifier is free for the next delivery. */ - BrokerStaticOutId_Release(bc, ack_resp.packet_id); + BrokerStaticOutId_Release(bc, ack_resp.packet_id, MQTT_QOS_1); BrokerStaticOrphan_OnPubAck(broker, bc, ack_resp.packet_id); #else @@ -8542,7 +8576,7 @@ static int BrokerClient_Process(MqttBroker* broker, BrokerClient* bc) if (comp_rc >= 0) { #ifdef WOLFMQTT_STATIC_MEMORY /* [MQTT-2.3.1-3] PUBCOMP completes the QoS 2 exchange. */ - BrokerStaticOutId_Release(bc, comp_resp.packet_id); + BrokerStaticOutId_Release(bc, comp_resp.packet_id, MQTT_QOS_2); BrokerStaticOrphan_OnPubComp(broker, bc, comp_resp.packet_id); #else diff --git a/src/mqtt_client.c b/src/mqtt_client.c index 9f6da0d..6c515c4 100644 --- a/src/mqtt_client.c +++ b/src/mqtt_client.c @@ -2525,13 +2525,6 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect) } if (mc_connect->stat.write == MQTT_MSG_BEGIN) { - /* A new handshake starts a fresh outbound Packet Identifier space: - * the client keeps no unacknowledged outbound PUBLISH/PUBREL across a - * Network Connection, so nothing from the previous one is still in - * flight [MQTT-2.3.1-3]. Must stay outside the WOLFMQTT_V5 block - * below - the table exists in every build. */ - MqttClient_SendIdsReset(client); - /* [MQTT-3.1.0-2] A Client can only send the CONNECT Packet once over a * Network Connection; a Server must treat a second one as a protocol * violation and disconnect. A partially written CONNECT re-enters with @@ -2543,6 +2536,17 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect) MQTT_CLIENT_FLAG_CONNECT_SENT) != 0) { return MQTT_TRACE_ERROR(MQTT_CODE_ERROR_STAT); } + + /* Past the guard, so this really is a new handshake: start a fresh + * outbound Packet Identifier space. The client keeps no + * unacknowledged outbound PUBLISH/PUBREL across a Network Connection, + * so nothing from a previous one is still in flight [MQTT-2.3.1-3]. + * Must run after the guard above - a refused duplicate CONNECT would + * otherwise wipe identifiers still in flight on the live connection - + * and outside the WOLFMQTT_V5 block below, since the table exists in + * every build. */ + MqttClient_SendIdsReset(client); + #ifdef WOLFMQTT_V5 #ifdef WOLFMQTT_MULTITHREAD rc = wm_SemLock(&client->lockClient); diff --git a/tests/test_broker_connect.c b/tests/test_broker_connect.c index 980911d..13c97cf 100644 --- a/tests/test_broker_connect.c +++ b/tests/test_broker_connect.c @@ -4341,7 +4341,7 @@ static int static_out_id_in_use(const BrokerClient* bc, word16 packet_id) int i; for (i = 0; i < BROKER_MAX_INFLIGHT_PER_SUB; i++) { - if (bc->out_inflight[i] == packet_id) { + if (bc->out_inflight[i].packet_id == packet_id) { return 1; } } @@ -4470,6 +4470,25 @@ TEST(static_fanout_does_not_reuse_unacked_packet_id) #endif /* WOLFMQTT_STATIC_MEMORY */ +#ifndef WOLFMQTT_STATIC_MEMORY +/* Find the orphan session carrying a disconnected client's Session state. The + * BrokerClient itself is freed by the close, so anything a test wants to check + * about the preserved queue must be read from here. */ +static BrokerOrphanSession* find_orphan_session(MqttBroker* broker, + const char* id) +{ + BrokerOrphanSession* o = broker->orphan_sessions; + + while (o != NULL) { + if (o->client_id != NULL && XSTRCMP(o->client_id, id) == 0) { + return o; + } + o = o->next; + } + return NULL; +} +#endif /* !WOLFMQTT_STATIC_MEMORY */ + #if defined(WOLFMQTT_NONBLOCK) && !defined(WOLFMQTT_STATIC_MEMORY) /* A zero-progress would-block result means none of the first transmission has * reached the network, so its later retry must still carry DUP=0 @@ -4556,6 +4575,7 @@ TEST(outbound_qos0_partial_failure_not_replayed_on_reconnect) MqttBroker broker; MqttBrokerNet net; BrokerClient* sub_bc; + BrokerOrphanSession* orphan; int i; static const byte connect_pub[] = { 0x10, 0x0D, @@ -4607,18 +4627,22 @@ TEST(outbound_qos0_partial_failure_not_replayed_on_reconnect) ASSERT_EQ(MQTT_QOS_0, sub_bc->out_q_head->qos); ASSERT_TRUE(sub_bc->client.write.pos > 0); + /* The failing write closes the client, so sub_bc is freed from here on and + * the preserved queue must be inspected through the orphan session. */ g_clients[1].write_err = 1; - (void)MqttBroker_Step(&broker); - /* The attempted QoS 0 message is gone rather than queued for replay. */ - ASSERT_NULL(sub_bc->out_q_head); - ASSERT_EQ(0, sub_bc->out_q_count); - for (i = 0; i < 4; i++) { (void)MqttBroker_Step(&broker); } ASSERT_TRUE(g_clients[1].closed); ASSERT_EQ(1, broker.orphan_session_count); + /* The attempted QoS 0 message did not follow the Session into the orphan; + * only QoS > 0 state is preserved for replay. */ + orphan = find_orphan_session(&broker, "S"); + ASSERT_NOT_NULL(orphan); + ASSERT_NULL(orphan->out_q_head); + ASSERT_EQ(0, orphan->out_q_count); + /* Reconnect the same persistent Session: only the CONNACK comes back, * with no second copy of the QoS 0 PUBLISH. */ mock_client_input_append(2, connect_sub, sizeof(connect_sub)); @@ -4729,6 +4753,7 @@ TEST(outbound_blocking_partial_failure_reconnect_sets_dup) MqttBroker broker; MqttBrokerNet net; BrokerClient* sub_bc; + BrokerOrphanSession* orphan; PublishInfo info; word16 packet_id; int i; @@ -4776,16 +4801,14 @@ TEST(outbound_blocking_partial_failure_reconnect_sets_dup) ASSERT_NOT_NULL(sub_bc); /* One byte of the queued delivery goes out, then the socket errors - - * both within the single blocking write. */ + * both within the single blocking write. That failure closes the client, + * so sub_bc is freed and the entry must be inspected via the orphan. */ g_clients[1].write_limit_then_err = 1; mock_client_input_append(0, publish_x, sizeof(publish_x)); (void)MqttBroker_Step(&broker); - ASSERT_NOT_NULL(sub_bc->out_q_head); - ASSERT_TRUE(sub_bc->client.write.pos > 0); - ASSERT_EQ(1, sub_bc->out_q_head->retransmit_dup); - packet_id = sub_bc->out_q_head->packet_id; - ASSERT_TRUE(packet_id != 0); + /* The fan-out drain discards its result, so the close lands on a later + * step. It frees sub_bc, hence the orphan lookup below. */ g_clients[1].write_err = 1; for (i = 0; i < 4; i++) { (void)MqttBroker_Step(&broker); @@ -4793,6 +4816,13 @@ TEST(outbound_blocking_partial_failure_reconnect_sets_dup) ASSERT_TRUE(g_clients[1].closed); ASSERT_EQ(1, broker.orphan_session_count); + orphan = find_orphan_session(&broker, "S"); + ASSERT_NOT_NULL(orphan); + ASSERT_NOT_NULL(orphan->out_q_head); + ASSERT_EQ(1, orphan->out_q_head->retransmit_dup); + packet_id = orphan->out_q_head->packet_id; + ASSERT_TRUE(packet_id != 0); + /* Reconnect the same persistent Session and inspect its replayed PUBLISH. */ mock_client_input_append(2, connect_sub, sizeof(connect_sub)); g_clients_active = 3; @@ -8408,9 +8438,11 @@ int main(int argc, char** argv) #if defined(WOLFMQTT_V5) && defined(WOLFMQTT_BROKER_WILL) RUN_TEST(connect_v5_oversize_will_payload_emits_connack); #ifdef WOLFMQTT_STATIC_MEMORY - RUN_TEST(static_fanout_does_not_reuse_unacked_packet_id); RUN_TEST(connect_v5_oversize_will_topic_emits_connack); #endif +#endif +#ifdef WOLFMQTT_STATIC_MEMORY + RUN_TEST(static_fanout_does_not_reuse_unacked_packet_id); #endif RUN_TEST(connect_v311_explicit_auto_prefix_refused); RUN_TEST(connect_unsupported_level_3_refused); diff --git a/wolfmqtt/mqtt_broker.h b/wolfmqtt/mqtt_broker.h index c727872..79b94a1 100644 --- a/wolfmqtt/mqtt_broker.h +++ b/wolfmqtt/mqtt_broker.h @@ -568,6 +568,12 @@ typedef struct BrokerOrphanSession { /* -------------------------------------------------------------------------- */ /* Broker client tracking */ /* -------------------------------------------------------------------------- */ +/* One outbound QoS > 0 delivery this client has not acknowledged yet. */ +typedef struct BrokerStaticOutId { + word16 packet_id; /* 0 = empty slot */ + MqttQoS qos; /* QoS the delivery went out with */ +} BrokerStaticOutId; + typedef struct BrokerClient { #ifdef WOLFMQTT_STATIC_MEMORY byte in_use; @@ -645,10 +651,13 @@ typedef struct BrokerClient { /* Outbound QoS 1/2 Packet Identifiers sent to this client and not yet * released by its PUBACK or PUBCOMP. [MQTT-2.3.1-4] applies the Client * identifier rule to a Server sending a QoS > 0 PUBLISH, so a new one - * must not reuse an identifier still awaiting acknowledgement. Slot value - * 0 is empty. Dynamic-memory builds derive this from the per-subscriber - * out_q instead, via BrokerNextPacketIdForQueue. */ - word16 out_inflight[BROKER_MAX_INFLIGHT_PER_SUB]; + * must not reuse an identifier still awaiting acknowledgement. The QoS is + * kept with it so a mismatched acknowledgement (a PUBACK for a QoS 2 + * delivery, say) cannot free a slot whose PUBREC/PUBCOMP flow is still + * running. packet_id 0 marks an empty slot. Dynamic-memory builds derive + * this from the per-subscriber out_q instead, via + * BrokerNextPacketIdForQueue. */ + BrokerStaticOutId out_inflight[BROKER_MAX_INFLIGHT_PER_SUB]; #endif #ifndef WOLFMQTT_STATIC_MEMORY /* Per-subscriber outbound publish queue. FIFO from head to tail;