rumqttc: potential auto ack lost

[ting-ms]

Here's a part of the code in the readb function of framed.rs:

                    if let Some(outgoing) = state.handle_incoming_packet(packet)? {
                        self.write(outgoing).await?;
                    }

Assume the outgoing packet is an auto ack for the incoming publish packet, if the self.write(outgoing).await? operation is canceled because the next_request task is completed first, this would cause the outgoing packet to be lost.

[swanandx]

hey, lmk if i understand this correctly, as readb internally uses Framed .feed() method, which is similar to .send() and their docs mention

futures_util::sink::SinkExt::send: if send is used as the event in a tokio::select! statement and some other branch completes first, then it is guaranteed that the message was not sent, but the message itself is lost.

so if send and feed are considered under same category, then there is a chance that message might be lost if this future is canceled right?

here, we can verify if feed() is cancel safe in this case or not ( idk how to guarantee that though )
or, an better approach will be to just read packets in the condition of select branch, and process then in the body, like:

select! { 
    o = self.readb(..) => { // this will just read the packets
        self.handle_batch(o).await?;
        // ..
    }
    // ..
}

this should also solve the issue right?

cc: @tekjar