Stop outbound drain when Session hand-off takes its queue. Claim Will before publishing it. Do not report a successful cancel while a reader owns the request. Take the client lock over the Session replay pool in MqttClient_Connect. #622
+799
−53
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -88,6 +88,9 @@ static int MqttClient_AuthEx(MqttClient *client, MqttAuth* auth, | |
| #if !defined(WOLFMQTT_MULTITHREAD) && !defined(WOLFMQTT_NONBLOCK) | ||
| static int MqttClient_CancelMessage(MqttClient *client, MqttObject* msg); | ||
| #endif | ||
| #ifndef WOLFMQTT_NO_SESSION_REPLAY | ||
| static int MqttClient_SendIds_Find(const MqttClient* client, word16 packet_id); | ||
| #endif | ||
| #ifdef WOLFMQTT_MULTITHREAD | ||
|
|
||
| #ifdef WOLFMQTT_USER_THREADING | ||
|
|
@@ -797,6 +800,37 @@ static void MqttClient_Replay_AddSafe(MqttClient* client, MqttPublish* publish) | |
| #endif | ||
| } | ||
|
|
||
| /* Discard the retained copies of a Session the server did not resume. A send | ||
| * is allowed once CONNECT reaches the transport, so an entry may instead belong | ||
| * to the Session being established: MqttClient_Connect empties the Packet | ||
| * Identifier table before sending CONNECT, so a still-reserved identifier marks | ||
| * one of those, and it is kept for a later resume [MQTT-4.4.0-1]. Reports a | ||
| * lock failure, unlike the wrappers above: a skipped discard would replay the | ||
| * previous Session's messages into this one. */ | ||
| static int MqttClient_Replay_ResetSafe(MqttClient* client) | ||
| { | ||
| int i; | ||
| #ifdef WOLFMQTT_MULTITHREAD | ||
| int rc = wm_SemLock(&client->lockClient); | ||
| if (rc != MQTT_CODE_SUCCESS) { | ||
| return rc; | ||
| } | ||
| #endif | ||
| for (i = 0; i < MQTT_MAX_REPLAY_MSGS; i++) { | ||
| if (client->replay[i].packet_id != 0 && | ||
| MqttClient_SendIds_Find(client, | ||
| client->replay[i].packet_id) >= 0) { | ||
| continue; /* published on this connection */ | ||
|
Comment on lines
+820
to
+823
|
||
| } | ||
| MqttClient_Replay_FreeSlot(&client->replay[i]); | ||
| } | ||
| client->replayIdx = MQTT_MAX_REPLAY_MSGS; | ||
| #ifdef WOLFMQTT_MULTITHREAD | ||
| wm_SemUnlock(&client->lockClient); | ||
| #endif | ||
| return MQTT_CODE_SUCCESS; | ||
| } | ||
|
|
||
| static void MqttClient_Replay_RemoveSafe(MqttClient* client, word16 packet_id) | ||
| { | ||
| #ifdef WOLFMQTT_MULTITHREAD | ||
|
|
@@ -861,8 +895,9 @@ static int MqttClient_SendIds_Find(const MqttClient* client, word16 packet_id) | |
| * MQTT_CODE_SUCCESS when it was free (or when isRetransmit says this is a | ||
| * re-send of the same Control Packet, which [MQTT-2.3.1-3] requires to keep | ||
| * its original identifier), and MQTT_CODE_ERROR_PACKET_ID when it is still | ||
| * awaiting its acknowledgement. The _Locked form is for a caller already | ||
| * holding client->lockClient, which is not recursive. */ | ||
| static int MqttClient_SendIdReserve_Locked(MqttClient* client, word16 packet_id, | ||
| void* owner, int isRetransmit, MqttPacketType ack_type) | ||
| { | ||
| int rc = MQTT_CODE_SUCCESS; | ||
|
|
@@ -871,12 +906,6 @@ static int MqttClient_SendIdReserve(MqttClient* client, word16 packet_id, | |
| if (packet_id == 0) { | ||
| return MQTT_CODE_SUCCESS; /* nothing to track */ | ||
| } | ||
| i = MqttClient_SendIds_Find(client, packet_id); | ||
| if (i >= 0) { | ||
| /* Only a re-send of the same Control Packet may keep an identifier | ||
|
|
@@ -906,6 +935,25 @@ static int MqttClient_SendIdReserve(MqttClient* client, word16 packet_id, | |
| * packets in flight, so the check is best effort past that point - | ||
| * raise MQTT_MAX_SEND_INFLIGHT to widen the window. */ | ||
| } | ||
| return rc; | ||
| } | ||
|
|
||
| static int MqttClient_SendIdReserve(MqttClient* client, word16 packet_id, | ||
| void* owner, int isRetransmit, MqttPacketType ack_type) | ||
| { | ||
| int rc; | ||
|
|
||
| if (packet_id == 0) { | ||
| return MQTT_CODE_SUCCESS; /* nothing to track */ | ||
| } | ||
| #ifdef WOLFMQTT_MULTITHREAD | ||
| rc = wm_SemLock(&client->lockClient); | ||
| if (rc != MQTT_CODE_SUCCESS) { | ||
| return rc; | ||
| } | ||
| #endif | ||
| rc = MqttClient_SendIdReserve_Locked(client, packet_id, owner, | ||
| isRetransmit, ack_type); | ||
| #ifdef WOLFMQTT_MULTITHREAD | ||
| wm_SemUnlock(&client->lockClient); | ||
| #endif | ||
|
|
@@ -2365,6 +2413,16 @@ static int MqttClient_WaitType(MqttClient *client, void *packet_obj, | |
| rc = MQTT_CODE_SUCCESS; | ||
| } | ||
| else { | ||
| #ifdef WOLFMQTT_MULTITHREAD | ||
| /* Terminal, so the claim must not outlive this reader. The | ||
| * CONTINUE path above keeps it: that reader resumes. */ | ||
| if (pendResp != NULL) { | ||
| if (wm_SemLock(&client->lockClient) == 0) { | ||
| pendResp->packetProcessing = 0; | ||
| wm_SemUnlock(&client->lockClient); | ||
| } | ||
| } | ||
| #endif | ||
| /* error, break */ | ||
| break; | ||
| } | ||
|
|
@@ -3595,15 +3653,25 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect) | |
| if (!(mc_connect->ack.flags & MQTT_CONNECT_ACK_FLAG_SESSION_PRESENT) || | ||
| !session_id_matched) { | ||
| /* A fresh Session starts with no outbound state to re-send. */ | ||
| rc = MqttClient_Replay_ResetSafe(client); | ||
| if (rc != MQTT_CODE_SUCCESS) { | ||
| return rc; | ||
| } | ||
| } | ||
| else { | ||
| int i; | ||
|
|
||
| /* The server resumed the Session, so these messages are still in | ||
| * flight as far as it is concerned: keep their Packet Identifiers | ||
| * reserved (MqttClient_Connect cleared the table above) and | ||
| * re-send them [MQTT-4.4.0-1]. A send thread reaches these same | ||
| * slots, so the walk holds the client lock. */ | ||
| #ifdef WOLFMQTT_MULTITHREAD | ||
| rc = wm_SemLock(&client->lockClient); | ||
| if (rc != MQTT_CODE_SUCCESS) { | ||
| return rc; | ||
| } | ||
| #endif | ||
| for (i = 0; i < MQTT_MAX_REPLAY_MSGS; i++) { | ||
| if (client->replay[i].packet_id == 0) { | ||
| continue; | ||
|
|
@@ -3617,14 +3685,17 @@ int MqttClient_Connect(MqttClient *client, MqttConnect *mc_connect) | |
| MqttClient_Replay_FreeSlot(&client->replay[i]); | ||
| continue; | ||
| } | ||
| (void)MqttClient_SendIdReserve_Locked(client, | ||
| client->replay[i].packet_id, &client->replay[i], 1, | ||
| client->replay[i].pubrelSent ? | ||
| MQTT_PACKET_TYPE_PUBLISH_COMP : | ||
| ((client->replay[i].qos == MQTT_QOS_2) ? | ||
| MQTT_PACKET_TYPE_PUBLISH_COMP : | ||
| MQTT_PACKET_TYPE_PUBLISH_ACK)); | ||
|
Comment on lines
+3688
to
3694
|
||
| } | ||
| #ifdef WOLFMQTT_MULTITHREAD | ||
| wm_SemUnlock(&client->lockClient); | ||
| #endif | ||
| client->replayIdx = 0; | ||
| mc_connect->stat.write = MQTT_MSG_PAYLOAD; | ||
| rc = MqttClient_ReplaySession(client, mc_connect); | ||
|
|
@@ -5335,6 +5406,57 @@ int MqttClient_CancelMessage(MqttClient *client, MqttObject* msg) | |
| PRINTF("Cancel Msg: %p", msg); | ||
| #endif | ||
|
|
||
| #ifdef WOLFMQTT_MULTITHREAD | ||
| /* Remove any pending responses expected. Runs before the resets below so | ||
| * that a refusal leaves the message exactly as it was found. | ||
| * | ||
| * A reading thread claims an entry with packetProcessing while it decodes | ||
| * the response into the packet_obj this message owns, having dropped | ||
| * lockClient first. Success is the caller's signal to release or reuse the | ||
| * object, so it is withheld while that claim stands; the entry stays | ||
| * listed and the caller retries until the reader marks it done. */ | ||
| rc = wm_SemLock(&client->lockClient); | ||
| if (rc != MQTT_CODE_SUCCESS) { | ||
| return rc; | ||
| } | ||
|
|
||
| for (tmpResp = client->firstPendResp; | ||
| tmpResp != NULL; | ||
| tmpResp = tmpResp->next) | ||
| { | ||
| #ifdef WOLFMQTT_DEBUG_CLIENT | ||
| PRINTF("\tMsg: %p (obj %p), Type %s (%d), ID %d, InProc %d, Done %d", | ||
| tmpResp, tmpResp->packet_obj, | ||
| MqttPacket_TypeDesc(tmpResp->packet_type), | ||
| tmpResp->packet_type, tmpResp->packet_id, | ||
| tmpResp->packetProcessing, tmpResp->packetDone); | ||
| #endif | ||
| if ((size_t)tmpResp->packet_obj == (size_t)msg || | ||
| (size_t)tmpResp - OFFSETOF(MqttMessage, pendResp) == (size_t)msg) { | ||
| #ifdef WOLFMQTT_DEBUG_CLIENT | ||
| PRINTF("Found Cancel Msg: %p (obj %p), Type %s (%d), ID %d, " | ||
| "InProc %d, Done %d", | ||
| tmpResp, tmpResp->packet_obj, | ||
| MqttPacket_TypeDesc(tmpResp->packet_type), | ||
| tmpResp->packet_type, tmpResp->packet_id, | ||
| tmpResp->packetProcessing, tmpResp->packetDone); | ||
| #endif | ||
| if (tmpResp->packetProcessing && !tmpResp->packetDone) { | ||
|
There was a problem hiding this comment. New CONTINUE return from CancelMessage is ignored by internal error paths, leaving a stale pendResp linked · API contract violations
Suggested fix: Update every internal caller of Related known findings (similar but distinct; listed for context, not part of this finding)
|
||
| wm_SemUnlock(&client->lockClient); | ||
| return MQTT_CODE_CONTINUE; | ||
|
kareem-wolfssl marked this conversation as resolved.
|
||
| } | ||
| /* Do not credit any reserved Receive Maximum unit here: the PUBLISH | ||
| * may already be on the wire, where the server keeps counting it | ||
| * [MQTT-4.9], so crediting it on a local cancel could exceed the | ||
| * negotiated quota. The unit is released on the acknowledgement, or | ||
| * recovered when the connection resets server_recv_max. */ | ||
| MqttClient_RespList_Remove(client, tmpResp); | ||
| break; | ||
| } | ||
| } | ||
| wm_SemUnlock(&client->lockClient); | ||
| #endif /* WOLFMQTT_MULTITHREAD */ | ||
|
|
||
| /* Whether this message's packet finished going out. MQTT_MSG_WAIT is only | ||
| * reached once the whole Control Packet has been written. */ | ||
| onWire = (mms_stat->write == MQTT_MSG_WAIT) ? 1 : 0; | ||
|
|
@@ -5377,46 +5499,6 @@ int MqttClient_CancelMessage(MqttClient *client, MqttObject* msg) | |
| mms_stat->recvQuotaHeld = 0; | ||
| #endif | ||
|
|
||
| /* cancel any active flags / locks */ | ||
| if (mms_stat->isReadActive) { | ||
| #ifdef WOLFMQTT_DEBUG_CLIENT | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.