QQ: fix effects ordering

Not really a problem but effects passed to checkout/4 would be
reversed which could potentially cause future confusion.

Author: kjnilsson

deps/rabbit/src/rabbit_fifo.erl

@@ -272,7 +272,7 @@ apply_(_Meta, #register_enqueuer{pid = Pid},
           end,
     {State, Res, [{monitor, process, Pid}]};
 apply_(Meta, #settle{msg_ids = MsgIds,
-                    consumer_key = Key},
+                     consumer_key = Key},
       #?STATE{consumers = Consumers} = State) ->
     case find_consumer(Key, Consumers) of
         {ConsumerKey, Con0} ->
@@ -284,7 +284,7 @@ apply_(Meta, #settle{msg_ids = MsgIds,
             {State, ok}
     end;
 apply_(Meta, #discard{consumer_key = ConsumerKey,
-                     msg_ids = MsgIds},
+                      msg_ids = MsgIds},
       #?STATE{consumers = Consumers } = State0) ->
     case find_consumer(ConsumerKey, Consumers) of
         {ActualConsumerKey, #consumer{} = Con} ->
@@ -293,8 +293,8 @@ apply_(Meta, #discard{consumer_key = ConsumerKey,
             {State0, ok}
     end;
 apply_(Meta, #return{consumer_key = ConsumerKey,
-                    msg_ids = MsgIds},
-      #?STATE{consumers = Cons} = State) ->
+                     msg_ids = MsgIds},
+       #?STATE{consumers = Cons} = State) ->
     case find_consumer(ConsumerKey, Cons) of
         {ActualConsumerKey, #consumer{checked_out = Checked}} ->
             return(Meta, ActualConsumerKey, MsgIds, false,
@@ -303,11 +303,11 @@ apply_(Meta, #return{consumer_key = ConsumerKey,
             {State, ok}
     end;
 apply_(Meta, #modify{consumer_key = ConsumerKey,
-                    delivery_failed = DelFailed,
-                    undeliverable_here = Undel,
-                    annotations = Anns,
-                    msg_ids = MsgIds},
-      #?STATE{consumers = Cons} = State) ->
+                     delivery_failed = DelFailed,
+                     undeliverable_here = Undel,
+                     annotations = Anns,
+                     msg_ids = MsgIds},
+       #?STATE{consumers = Cons} = State) ->
     case find_consumer(ConsumerKey, Cons) of
         {ActualConsumerKey, #consumer{checked_out = Checked}}
           when Undel == false ->
@@ -1988,8 +1988,8 @@ checkout0(Meta, {success, ConsumerKey, MsgId,
               end,
     checkout0(Meta, checkout_one(Meta, ExpiredMsg, State, Effects), SendAcc);
 checkout0(_Meta, {_Activity, ExpiredMsg, State0, Effects0}, SendAcc) ->
-    Effects = add_delivery_effects(Effects0, SendAcc, State0),
-    {State0, ExpiredMsg, lists:reverse(Effects)}.
+    Effects = add_delivery_effects([], SendAcc, State0),
+    {State0, ExpiredMsg, Effects0 ++ lists:reverse(Effects)}.
 
 evaluate_limit(_Index, Result,
                #?STATE{cfg = #cfg{max_length = undefined,
@@ -2983,14 +2983,24 @@ do_snapshot(MacVer, Ts, #snapshot{index = _ChIdx,
             RaAux, DiscardedBytes, Force)
   when is_integer(MacVer) andalso MacVer >= 8 ->
     LastAppliedIdx = ra_aux:last_applied(RaAux),
-    #?STATE{} = MacState = ra_aux:machine_state(RaAux),
+    #?STATE{consumers = Consumers,
+            enqueuers = Enqueuers} = MacState = ra_aux:machine_state(RaAux),
     TimeSince = Ts - SnapTime,
     MsgsTot = messages_total(MacState),
     %% if the approximate snapshot size * 2 can be reclaimed it is worth
     %% taking a snapshot
-    %% TODO: take number of enqueues and consumers into account
-    ApproxSnapSize = 4096 + (32 * MsgsTot),
-    EnoughDataRemoved = DiscardedBytes - LastDiscardedBytes > (ApproxSnapSize * 2),
+    %% take number of enqueues and consumers into account
+    %% message: 32 bytes
+    %% enqueuer: 96 bytes
+    %% consumer: 256 bytes
+    NumEnqueuers = map_size(Enqueuers),
+    NumConsumers = map_size(Consumers),
+    ApproxSnapSize = 4096 +
+                     (MsgsTot * 32) +
+                     (NumEnqueuers * 96) +
+                     (NumConsumers * 256),
+
+    EnoughDataRemoved = DiscardedBytes - LastDiscardedBytes > (ApproxSnapSize * 3),
 
     {CheckMinInterval, _CheckMinIndexes, _CheckMaxIndexes} =
         persistent_term:get(quorum_queue_checkpoint_config,

deps/rabbit/test/rabbit_fifo_SUITE.erl

@@ -585,7 +585,6 @@ untracked_enq_deq_test(Config) ->
                            State0),
     {_State2, _, Effs} =
         apply(meta(Config, 3), make_checkout(Cid, {dequeue, settled}, #{}), State1),
-        ct:pal("Effs ~p", [State1]),
     ?ASSERT_EFF({log, [1], _}, Effs),
     ok.
 
@@ -821,7 +820,6 @@ requeue_test(Config) ->
      [_Monitor, {log_ext, [1], _Fun, _}]} = checkout(Config, ?LINE, Cid, 1, State0),
 
     [{MsgId, {H1, _}}] = rabbit_fifo:get_checked_out(CKey, MsgId, MsgId, State1),
-    ct:pal("query consumers ~p", [rabbit_fifo:query_consumers(State1)]),
 
     [{append, Requeue, _}] = rabbit_fifo:make_requeue(CKey, {notify, 1, self()},
                                                       [{MsgId, 1, H1, Msg1}], []),
@@ -991,7 +989,6 @@ discarded_message_with_dead_letter_handler_emits_log_effect_test(Config) ->
 
     ?assertEqual(undefined, mc:get_annotation(acquired_count, McOut)),
     ?assertEqual(1, mc:get_annotation(delivery_count, McOut)),
-
     ok.
 
 discard_after_cancel_test(Config) ->
@@ -1940,18 +1937,18 @@ single_active_consumer_priority_test(Config) ->
     ],
     {#rabbit_fifo{ cfg = #cfg{resource = Resource}}, StateMachineEvents} = run_log(Config, S0, Entries, fun single_active_invariant/1),
     ModCalls = [ S || S = {mod_call, rabbit_quorum_queue, update_consumer_handler, _} <- StateMachineEvents ],
-    
-    %% C1 should be added as single_active    
+
+    %% C1 should be added as single_active
     assert_update_consumer_handler_state_transition(C1, Resource, true, single_active, lists:nth(1, ModCalls)),
-    %% C1 should transition to waiting because ...
-    assert_update_consumer_handler_state_transition(C1, Resource, false, waiting, lists:nth(2, ModCalls)),
     %% C2 should become single_active
-    assert_update_consumer_handler_state_transition(C2, Resource, true, single_active, lists:nth(3, ModCalls)),
-    %% C2 should transition as waiting because ...
-    assert_update_consumer_handler_state_transition(C2, Resource, false, waiting, lists:nth(4, ModCalls)),
+    assert_update_consumer_handler_state_transition(C2, Resource, true, single_active, lists:nth(2, ModCalls)),
+    %% C1 should transition to waiting
+    assert_update_consumer_handler_state_transition(C1, Resource, false, waiting, lists:nth(3, ModCalls)),
     %% C3 is added as single_active
-    assert_update_consumer_handler_state_transition(C3, Resource, true, single_active, lists:nth(5, ModCalls)),
-    
+    assert_update_consumer_handler_state_transition(C3, Resource, true, single_active, lists:nth(4, ModCalls)),
+    %% C2 should transition as waiting
+    assert_update_consumer_handler_state_transition(C2, Resource, false, waiting, lists:nth(5, ModCalls)),
+
     ok.
 
 assert_update_consumer_handler_state_transition(ConsumerId, Resource, IsActive, UpdatedState, ModCall) ->
@@ -2405,7 +2402,6 @@ reject_publish_purge_test(Config) ->
                             rabbit_fifo:make_enqueue(Pid1, 2, two), State2),
     {State4, ok, Efx} = apply(meta(Config, 4, ?LINE, {notify, 2, Pid1}),
                               rabbit_fifo:make_enqueue(Pid1, 3, three), State3),
-    % ct:pal("Efx ~tp", [Efx]),
     ?ASSERT_EFF({send_msg, P, {queue_status, reject_publish}, [ra_event]}, P == Pid1, Efx),
     {_State5, {purge, 3}, Efx1} = apply(meta(Config, 5), rabbit_fifo:make_purge(), State4),
     ?ASSERT_EFF({send_msg, P, {queue_status, go}, [ra_event]}, P == Pid1, Efx1),