QQ: fix SAC activation bug

unit test and initial "fix"

TODO: needs a new machine version
TODO: needs the same behaviour for the requeue{} command.

Author: kjnilsson

deps/rabbit/src/rabbit_fifo.erl

@@ -1760,7 +1760,8 @@ return(#{} = Meta, ConsumerKey, MsgIds, IncrDelCount, Anns,
                  _ ->
                      State1
              end,
-    checkout(Meta, State0, State2, Effects1).
+    {State3, Effects2} = activate_next_consumer({State2, Effects1}),
+    checkout(Meta, State0, State3, Effects2).
 
 % used to process messages that are finished
 complete(Meta, ConsumerKey, [MsgId],

deps/rabbit/test/rabbit_fifo_SUITE.erl

@@ -1916,6 +1916,43 @@ single_active_consumer_higher_waiting_disconnected_test(Config) ->
 
     ok.
 
+single_active_consumer_higher_waiting_return_test(Config) ->
+    S0 = init(#{name => ?FUNCTION_NAME,
+                queue_resource => rabbit_misc:r("/", queue, ?FUNCTION_NAME_B),
+                single_active_consumer_on => true}),
+
+    Pid1 = test_util:fake_pid(node()),
+    C1Pid = test_util:fake_pid(n1@banana),
+    C2Pid = test_util:fake_pid(n2@banana),
+    % % adding some consumers
+    {CK1, C1} = {?LINE, {?LINE_B, C1Pid}},
+    {CK2, C2} = {?LINE, {?LINE_B, C2Pid}},
+    Entries =
+    [
+     %% add a consumer
+     {CK1, make_checkout(C1, {auto, {simple_prefetch, 1}}, #{priority => 1})},
+     ?ASSERT(#rabbit_fifo{consumers = #{CK1 := #consumer{status = up}},
+                          waiting_consumers = []}),
+
+     %% enqueue a message
+     {?LINE , rabbit_fifo:make_enqueue(Pid1, 1, msg1)},
+
+     %% add a consumer with a higher priority, current is quiescing
+     {CK2, make_checkout(C2, {auto, {simple_prefetch, 1}}, #{priority => 2})},
+     ?ASSERT(#rabbit_fifo{consumers = #{CK1 := #consumer{status = quiescing}},
+                          waiting_consumers = [{CK2, _}]}),
+     %% C1 returns message
+     {?LINE, rabbit_fifo:make_return(CK1, [0])},
+     %% C2 should activated
+     ?ASSERT(#rabbit_fifo{consumers = #{CK2 := #consumer{status = up,
+                                                         checked_out = Ch,
+                                                         credit = 0}},
+                          waiting_consumers = [_]} when map_size(Ch) == 1)
+    ],
+    {_S1, _} = run_log(Config, S0, Entries, fun single_active_invariant/1),
+
+    ok.
+
 single_active_consumer_quiescing_disconnected_test(Config) ->
     S0 = init(#{name => ?FUNCTION_NAME,
                 queue_resource => rabbit_misc:r("/", queue, ?FUNCTION_NAME_B),