From 324fb6d76e1e1b68ec2fbd45956ef9e424b15429 Mon Sep 17 00:00:00 2001 From: Daniil Fedotov Date: Tue, 4 Jun 2019 14:08:40 -0400 Subject: Reject a message before discarding it from BQ. discard/4 confirms a message, which can cause races between confirms and rejects. --- src/rabbit_amqqueue_process.erl | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl index 735093f818..25dfdc91cd 100644 --- a/src/rabbit_amqqueue_process.erl +++ b/src/rabbit_amqqueue_process.erl @@ -739,14 +739,21 @@ maybe_drop_head(AlreadyDropped, State = #q{backing_queue = BQ, end. send_reject_publish(#delivery{confirm = true, - sender = SenderPid, - msg_seq_no = MsgSeqNo} = Delivery, + sender = SenderPid, + flow = Flow, + msg_seq_no = MsgSeqNo, + message = #basic_message{id = MsgId}}, _Delivered, State = #q{ backing_queue = BQ, backing_queue_state = BQS, msg_id_to_channel = MTC}) -> - {BQS1, MTC1} = discard(Delivery, BQ, BQS, MTC), gen_server2:cast(SenderPid, {reject_publish, MsgSeqNo, self()}), + + MTC1 = case gb_trees:is_defined(MsgId, MTC) of + true -> gb_trees:delete(MsgId, MTC); + false -> MTC + end, + BQS1 = BQ:discard(MsgId, SenderPid, Flow, BQS), State#q{ backing_queue_state = BQS1, msg_id_to_channel = MTC1 }; send_reject_publish(#delivery{confirm = false}, _Delivered, State) -> -- cgit v1.2.1