QQ: Message expiration scans all priorities

Expiration is now done for all priorities using a separate command
that performs a "shallow" expiry run. I.e. each priority will have
messages dropped until it encounters the next non expired one.

This means that it can still leave expired messages on queue if they
are preceeded by message that are not yet or never will be expired.

This is consistent with existing expiration logic.

Author: kjnilsson

deps/rabbit/src/rabbit_fifo.erl

@@ -175,7 +175,6 @@
 
 -record(enqueuer,
         {next_seqno = 1 :: msg_seqno(),
-         % out of order enqueues - sorted list
          unused = ?NIL,
          status = up :: up | suspected_down,
          %% it is useful to have a record of when this was blocked
@@ -250,7 +249,7 @@
          % rabbit_fifo_index can be slow when calculating the smallest
          % index when there are large gaps but should be faster than gb_trees
          % for normal appending operations as it's backed by a map
-         unused_0 = ?NIL,
+         last_command_time = 0,
          unused_1 = ?NIL,
          % consumers need to reflect consumer state at time of snapshot
          consumers = #{} :: #{consumer_key() => consumer()},

deps/rabbit/src/rabbit_fifo.hrl

@@ -20,7 +20,10 @@
          from_lqueue/1,
          indexes/1,
          get_lowest_index/1,
-         overview/1
+         overview/1,
+         take_while/2,
+         any_priority_next/2,
+         fold_priorities_next/3
         ]).
 
 -define(STATE, pq).
@@ -156,8 +159,75 @@ overview(#?STATE{len = Len,
       num_active_priorities => map_size(Buckets),
       lowest_index => get_lowest_index(State)}.
 
+-spec take_while(fun ((msg()) -> boolean()), state()) ->
+    {[msg()], state()}.
+take_while(Fun, #?STATE{len = Len,
+                        buckets = Buckets0} = State)
+  when is_function(Fun) ->
+    {Buckets, Acc} = maps:fold(
+                       fun (P, Q0, {B0, Items0}) ->
+                               case take_while(Q0, Fun, Items0) of
+                                   {?EMPTY, Items} ->
+                                       {maps:remove(P, B0), Items};
+                                   {Q, Items} ->
+                                       {B0#{P => Q}, Items}
+                               end
+                       end,
+                       {Buckets0, []},
+                       maps:iterator(Buckets0, ordered)),
+
+    %% TODO: optimise updates
+    %% update bitmap
+    Bitmap = maps:fold(fun (P, _Q, B) -> B bor (1 bsl P) end, 0, Buckets),
+
+    {lists:reverse(Acc),
+     State#?STATE{len = Len - length(Acc),
+                  buckets = Buckets,
+                  bitmap = Bitmap}}.
+
+-spec any_priority_next(fun ((msg()) -> boolean()), state()) ->
+    boolean().
+any_priority_next(Fun, #?STATE{buckets = Buckets0})
+  when is_function(Fun) ->
+    maps_any(Fun, maps:next(maps:iterator(Buckets0))).
+
+-spec fold_priorities_next(fun ((msg(), Acc) -> Acc), Acc, state()) ->
+    Acc when Acc :: term().
+fold_priorities_next(Fun, Acc, #?STATE{buckets = Buckets0})
+  when is_function(Fun) ->
+    maps:fold(fun (_P, Q, A) ->
+                      Fun(peek(Q), A)
+              end, Acc, Buckets0).
+
 %% INTERNAL
 
+maps_any(_Fun, none) ->
+    false;
+maps_any(Fun, {_, Q, I}) ->
+    case Fun(peek(Q)) of
+        true ->
+            true;
+        false ->
+            maps_any(Fun, maps:next(I))
+    end.
+
+take_while(?EMPTY, _Fun, Acc) ->
+    {?EMPTY, Acc};
+take_while(Q, Fun, Acc) ->
+    case peek(Q) of
+        empty ->
+            {Q, Acc};
+        Msg ->
+            case Fun(Msg) of
+                true ->
+                    take_while(drop(Q), Fun, [Msg | Acc]);
+                false ->
+                    {Q, Acc}
+            end
+    end.
+
+
+
 %% invariant, if the queue is non empty so is the Out (right) list.
 in(X, ?EMPTY) ->
     {1, [], [X]};

deps/rabbit/src/rabbit_fifo_pq.erl

@@ -1987,7 +1987,7 @@ make_ra_conf(Q, ServerId, Membership, MacVersion)
                  Membership, MacVersion).
 
 make_ra_conf(Q, ServerId, TickTimeout,
-             _SnapshotInterval, CheckpointInterval,
+             _SnapshotInterval, _CheckpointInterval,
              Membership, MacVersion) ->
     QName = amqqueue:get_name(Q),
     #resource{name = QNameBin} = QName,
@@ -1998,8 +1998,9 @@ make_ra_conf(Q, ServerId, TickTimeout,
     Formatter = {?MODULE, format_ra_event, [QName]},
     LogCfg = #{uid => UId,
                min_snapshot_interval => 0,
-               min_checkpoint_interval => CheckpointInterval,
-               max_checkpoints => 3},
+               % min_checkpoint_interval => CheckpointInterval,
+               % max_checkpoints => 3,
+               major_compaction_strategy => {num_minors, 32}},
     rabbit_misc:maps_put_truthy(membership, Membership,
                                 #{cluster_name => ClusterName,
                                   id => ServerId,

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -2147,6 +2147,8 @@ dead_letter_policy(Config) ->
 %% Test that messages are at most once dead letter in the correct order
 %% for reason 'maxlen'.
 at_most_once_dead_letter_order_maxlen(Config) ->
+    check_quorum_queues_v8_compat(Config),
+
     [Server | _] = Servers = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
 
     Ch = rabbit_ct_client_helpers:open_channel(Config, Server),
@@ -2261,16 +2263,18 @@ at_most_once_dead_letter_order_delivery_limit(Config) ->
                            #'basic.publish'{routing_key = QQ},
                            #amqp_msg{payload = <<"m2">>}),
 
-    ok = subscribe(Ch, QQ, false),
+    Ch2 = rabbit_ct_client_helpers:open_channel(Config, Server),
+    ok = subscribe(Ch2, QQ, false),
     receive {_, #amqp_msg{payload = P1}} ->
                 ?assertEqual(<<"m1">>, P1)
     end,
     receive {_, #amqp_msg{payload = P2}} ->
                 ?assertEqual(<<"m2">>, P2)
     end,
-    ok = amqp_channel:call(Ch, #'basic.nack'{delivery_tag = 0,
-                                             multiple = true,
-                                             requeue = true}),
+    amqp_channel:close(Ch2),
+    % ok = amqp_channel:call(Ch, #'basic.nack'{delivery_tag = 0,
+    %                                          multiple = true,
+    %                                          requeue = true}),
 
     wait_for_consensus(DLQ, Config),
     wait_for_messages_ready(Servers,  ra_name(DLQ), 2),
@@ -5337,7 +5341,7 @@ check_quorum_queues_v8_compat(Config) ->
         true ->
             ok;
         false ->
-            throw({skip, "test will only work on QQ machine version > 8"})
+            throw({skip, "test will only work on QQ machine version >= 8"})
     end.
 
 lists_interleave([], _List) ->

deps/rabbit/test/quorum_queue_SUITE.erl

@@ -1620,8 +1620,11 @@ single_active_consumer_state_enter_leader_include_waiting_consumers_test(Config)
               ],
     {State1, _} = run_log(Config, State0, Entries),
     Effects = rabbit_fifo:state_enter(leader, State1),
-    %% 2 effects for each consumer process (channel process), 1 effect for the node,
-    ?assertEqual(2 * 3 + 1 + 1 + 1, length(Effects)).
+    ct:pal("Efx ~p", [Effects]),
+    %% 2 effects for each consumer process (channel process),
+    %% 1 effect for the node,
+    %% 1 for decorators
+    ?assertEqual(2 * 3 + 1 + 1, length(Effects)).
 
 single_active_consumer_state_enter_eol_include_waiting_consumers_test(Config) ->
     Resource = rabbit_misc:r("/", queue, ?FUNCTION_NAME_B),
@@ -3223,6 +3226,39 @@ modify_test(Config) ->
 
     ok.
 
+priorities_expire_test(Config) ->
+    State0 = init(#{name => ?FUNCTION_NAME,
+                    queue_resource => rabbit_misc:r("/", queue,
+                                                    ?FUNCTION_NAME_B)}),
+    Pid1 = spawn(fun() -> ok end),
+
+    Entries =
+    [
+     {?LINE, make_enqueue(Pid1, 1,
+                          mk_mc(<<"p1">>, #'P_basic'{priority = 9,
+                                                     expiration = <<"100">>}))},
+     {?LINE, make_enqueue(Pid1, 2,
+                          mk_mc(<<"p1">>, #'P_basic'{priority = 9,
+                                                     expiration = <<"100000">>}))},
+     {?LINE, make_enqueue(Pid1, 3,
+                          mk_mc(<<"p7">>, #'P_basic'{priority = 7,
+                                                     expiration = <<"100">>}))},
+     {?LINE, make_enqueue(Pid1, 4,
+                          mk_mc(<<"p7">>, #'P_basic'{priority = 7,
+                                                     expiration = <<"100000">>}))},
+     {?LINE, make_enqueue(Pid1, 5,
+                          mk_mc(<<"p7b">>, #'P_basic'{priority = 3}))},
+
+     {?LINE + 101, {timeout, {expire_msgs, shallow}}},
+
+     ?ASSERT(_, fun(State) ->
+                        ?assertMatch(#{num_messages := 3},
+                                     rabbit_fifo:overview(State))
+                end)
+    ],
+    {_State2, _} = run_log(Config, State0, Entries),
+    ok.
+
 %% Utility
 %%
 
@@ -3232,6 +3268,7 @@ apply(Meta, Entry, State) -> rabbit_fifo:apply(Meta, Entry, State).
 init_aux(Conf) -> rabbit_fifo:init_aux(Conf).
 handle_aux(S, T, C, A, A2) -> rabbit_fifo:handle_aux(S, T, C, A, A2).
 make_checkout(C, S, M) -> rabbit_fifo:make_checkout(C, S, M).
+make_enqueue(P, S, M) -> rabbit_fifo:make_enqueue(P, S, M).
 
 cid(A) when is_atom(A) ->
     atom_to_binary(A, utf8).
@@ -3242,10 +3279,13 @@ single_active_invariant( #rabbit_fifo{consumers = Cons}) ->
                               end, Cons)).
 
 mk_mc(Body) ->
+    mk_mc(Body, #'P_basic'{}).
+
+mk_mc(Body, BasicProps) ->
     mc_amqpl:from_basic_message(
       #basic_message{routing_keys = [<<"">>],
                      exchange_name = #resource{name = <<"x">>,
                                                kind = exchange,
                                                virtual_host = <<"v">>},
-                     content = #content{properties = #'P_basic'{},
+                     content = #content{properties = BasicProps,
                                         payload_fragments_rev = [Body]}}).

deps/rabbit/test/rabbit_fifo_SUITE.erl

@@ -17,6 +17,8 @@ all() ->
 all_tests() ->
     [
      basics,
+     take_while,
+     any_priority_next,
      property
     ].
 
@@ -80,6 +82,73 @@ basics(_Config) ->
     empty = rabbit_fifo_pq:out(Q6),
     ok.
 
+take_while(_Config) ->
+    Q1 = lists:foldl(
+           fun ({P, I}, Q) ->
+                   rabbit_fifo_pq:in(P, I, Q)
+           end, rabbit_fifo_pq:new(),
+           [
+            {1, ?MSG(1)}, {1, ?MSG(2)}, {1, ?MSG(3)},
+            {2, ?MSG(1)}, {2, ?MSG(2)}, {2, ?MSG(3)},
+            {3, ?MSG(1)}, {3, ?MSG(2)}, {3, ?MSG(3)},
+            {4, ?MSG(1)}, {4, ?MSG(2)}, {4, ?MSG(3)},
+            {5, ?MSG(1, 10)}, {5, ?MSG(2, 20)}, {5, ?MSG(3, 30)}
+           ]),
+
+    {Taken, Q2} = rabbit_fifo_pq:take_while(fun (?MSG(I, _)) ->
+                                                    I < 3
+                                            end, Q1),
+    ?assertMatch([
+                  ?MSG(1, 10), ?MSG(2, 20),
+                  ?MSG(1, 1), ?MSG(2, 2),
+                  ?MSG(1, 1), ?MSG(2, 2),
+                  ?MSG(1, 1), ?MSG(2, 2),
+                  ?MSG(1, 1), ?MSG(2, 2)
+                 ], Taken),
+
+
+    ?assertEqual(5, rabbit_fifo_pq:len(Q2)),
+    ?assertEqual(10, length(Taken)),
+    {?MSG(3, 30), Q3} = rabbit_fifo_pq:out(Q2),
+    {?MSG(3), Q4} = rabbit_fifo_pq:out(Q3),
+    {?MSG(3), Q5} = rabbit_fifo_pq:out(Q4),
+    {?MSG(3), Q6} = rabbit_fifo_pq:out(Q5),
+    {?MSG(3), _Q7} = rabbit_fifo_pq:out(Q6),
+
+
+    {_Taken2, Q} = rabbit_fifo_pq:take_while(fun (?MSG(_, _)) ->
+                                                    true
+                                            end, Q2),
+
+    ct:pal("Q ~p", [Q]),
+
+    ok.
+
+any_priority_next(_Config) ->
+    Q0 = rabbit_fifo_pq:new(),
+
+    ?assertNot(rabbit_fifo_pq:any_priority_next(fun (_) -> true end, Q0)),
+
+    Q1 = lists:foldl(fun ({P, I}, Q) ->
+                             rabbit_fifo_pq:in(P, I, Q)
+                     end, Q0,
+                     [
+                      {1, ?MSG(1)}, {1, ?MSG(2)}, {1, ?MSG(3)},
+                      {2, ?MSG(1)}, {2, ?MSG(2)}, {2, ?MSG(3)},
+                      {3, ?MSG(2)}, {3, ?MSG(3)},
+                      {4, ?MSG(1)}, {4, ?MSG(2)}, {4, ?MSG(3)},
+                      {5, ?MSG(1)}, {5, ?MSG(2)}, {5, ?MSG(3)}
+                     ]),
+
+    ?assert(rabbit_fifo_pq:any_priority_next(fun (?MSG(I, _)) ->
+                                                     I > 1
+                                             end, Q1)),
+    ?assertNot(rabbit_fifo_pq:any_priority_next(fun (?MSG(I, _)) ->
+                                                        I > 6
+                                                end, Q1)),
+
+    ok.
+
 hi_is_prioritised(_Config) ->
     Q0 = rabbit_fifo_q:new(),
     %% when `hi' has a lower index than the next 'no' then it is still