WolfSSL
/
wolfMQTT
mirror of https://github.com/wolfSSL/wolfMQTT.git
1
0
Fork 0
Code Issues Releases Wiki Activity

Fixes from review

pull/615/head
Eric Blankenhorn 2026-09-08 13:55:28 -05:00
parent 1301252ff6
commit 6ec061779d
4 changed files with 115 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
src/mqtt_broker.c
View File

@ -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

18
src/mqtt_client.c
View File

@ -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);

58
tests/test_broker_connect.c
View File

@ -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);

17
wolfmqtt/mqtt_broker.h
View File

@ -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;