Merge pull request #3572 from rabbitmq/mergify/bp/v3.9.x/pr-3560

Fix memory leak when using mandatory flag in CQs (backport #3560)

Author: michaelklishin

deps/rabbit/src/rabbit_classic_queue.erl

@@ -299,9 +299,9 @@ settlement_action(Type, QRef, MsgSeqs, Acc) ->
 deliver(Qs0, #delivery{flow = Flow,
                        msg_seq_no = MsgNo,
                        message = #basic_message{exchange_name = _Ex},
-                       confirm = _Confirm} = Delivery) ->
+                       confirm = Confirm} = Delivery) ->
     %% TODO: record master and slaves for confirm processing
-    {MPids, SPids, Qs, Actions} = qpids(Qs0, MsgNo),
+    {MPids, SPids, Qs, Actions} = qpids(Qs0, Confirm, MsgNo),
     QPids = MPids ++ SPids,
     case Flow of
         %% Here we are tracking messages sent by the rabbit_channel
@@ -361,22 +361,22 @@ purge(Q) when ?is_amqqueue(Q) ->
     QPid = amqqueue:get_pid(Q),
     delegate:invoke(QPid, {gen_server2, call, [purge, infinity]}).
 
-qpids(Qs, MsgNo) ->
+qpids(Qs, Confirm, MsgNo) ->
     lists:foldl(
       fun ({Q, S0}, {MPidAcc, SPidAcc, Qs0, Actions0}) ->
               QPid = amqqueue:get_pid(Q),
               SPids = amqqueue:get_slave_pids(Q),
               QRef = amqqueue:get_name(Q),
               Actions = [{monitor, QPid, QRef}
                          | [{monitor, P, QRef} || P <- SPids]] ++ Actions0,
-              %% confirm record only if MsgNo isn't undefined
+              %% confirm record only if necessary
               S = case S0 of
                       #?STATE{unconfirmed = U0} ->
                           Rec = [QPid | SPids],
-                          U = case MsgNo of
-                                  undefined ->
+                          U = case Confirm of
+                                  false ->
                                       U0;
-                                  _ ->
+                                  true ->
                                       U0#{MsgNo => #msg_status{pending = Rec}}
                               end,
                           S0#?STATE{pid = QPid,

release-notes/3.9.8.md

@@ -23,12 +23,12 @@ Contributors are encouraged to update them together with their changes. This hel
 * When the mandatory flag was used when publishing to classic queues,
   but publisher confirms were not, channels memory usage would grow indefinitely.
 
-   GitHub issue: [#3560](https://github.com/rabbitmq/rabbitmq-server/issues/3560)
+  GitHub issue: [#3560](https://github.com/rabbitmq/rabbitmq-server/issues/3560)
 
 * `rabbitmq-diagnostics memory_breakdown` failed to read memory of connection
   reader, writer and channel processes.
 
-   GitHub issue: [#3570](https://github.com/rabbitmq/rabbitmq-server/issues/3570)
+  GitHub issue: [#3570](https://github.com/rabbitmq/rabbitmq-server/issues/3570)
 
 
 ## Dependency Upgrades