QQ: track discarded bytes and take snapshots based on that.

Author: kjnilsson

deps/rabbit/src/rabbit_fifo.erl

@@ -180,6 +180,31 @@
          unused_3 = ?NIL
         }).
 
+-record(messages,
+        {
+         messages = rabbit_fifo_q:new() :: rabbit_fifo_q:state(),
+         messages_total = 0 :: non_neg_integer(),
+         % queue of returned msg_in_ids - when checking out it picks from
+         returns = lqueue:new() :: lqueue:lqueue(term())
+        }).
+
+-record(dlx_consumer,
+        {pid :: pid(),
+         prefetch :: non_neg_integer(),
+         checked_out = #{} :: #{msg_id() =>
+                                optimised_tuple(rabbit_dead_letter:reason(), msg())},
+         next_msg_id = 0 :: msg_id()}).
+
+-record(rabbit_fifo_dlx,
+        {consumer :: option(#dlx_consumer{}),
+         %% Queue of dead-lettered messages.
+         discards = lqueue:new() :: lqueue:lqueue(optimised_tuple(rabbit_dead_letter:reason(), msg())),
+         %% Raft indexes of messages in both discards queue and dlx_consumer's checked_out map
+         %% so that we get the smallest ra index in O(1).
+         ra_indexes = rabbit_fifo_index:empty() :: rabbit_fifo_index:state(),
+         msg_bytes = 0 :: non_neg_integer(),
+         msg_bytes_checkout = 0 :: non_neg_integer()}).
+
 -record(rabbit_fifo,
         {cfg :: #cfg{},
          % unassigned messages
@@ -207,7 +232,7 @@
          % consumers that require further service are queued here
          service_queue = priority_queue:new() :: priority_queue:q(),
          %% state for at-least-once dead-lettering
-         dlx = rabbit_fifo_dlx:init() :: rabbit_fifo_dlx:state(),
+         dlx = #rabbit_fifo_dlx{} :: #rabbit_fifo_dlx{},
          msg_bytes_enqueue = 0 :: non_neg_integer(),
          msg_bytes_checkout = 0 :: non_neg_integer(),
          %% one is picked if active consumer is cancelled or dies

deps/rabbit/src/rabbit_fifo.hrl

@@ -147,13 +147,13 @@ enqueue(QName, Correlation, Msg,
             {reject_publish, State0};
         {error, {shutdown, delete}} ->
             ?LOG_DEBUG("~ts: QQ ~ts tried to register enqueuer during delete shutdown",
-                             [?MODULE, rabbit_misc:rs(QName)]),
+                       [?MODULE, rabbit_misc:rs(QName)]),
             {reject_publish, State0};
         {timeout, _} ->
             {reject_publish, State0};
         Err ->
             ?LOG_DEBUG("~ts: QQ ~ts error when registering enqueuer ~p",
-                             [?MODULE, rabbit_misc:rs(QName), Err]),
+                       [?MODULE, rabbit_misc:rs(QName), Err]),
             exit(Err)
     end;
 enqueue(_QName, _Correlation, _Msg,

deps/rabbit/src/rabbit_fifo_client.erl

@@ -7,7 +7,7 @@
 -module(rabbit_fifo_dlx).
 
 -include("rabbit_fifo_dlx.hrl").
--include("rabbit_fifo.hrl").
+-include("rabbit_fifo_v7.hrl").
 -include_lib("kernel/include/logger.hrl").
 -compile({no_auto_import, [apply/3]}).
 

deps/rabbit/src/rabbit_fifo_dlx.erl

@@ -7,7 +7,8 @@
 -record(dlx_consumer,
         {pid :: pid(),
          prefetch :: non_neg_integer(),
-         checked_out = #{} :: #{msg_id() => optimised_tuple(rabbit_dead_letter:reason(), msg())},
+         checked_out = #{} :: #{msg_id() =>
+                                optimised_tuple(rabbit_dead_letter:reason(), msg())},
          next_msg_id = 0 :: msg_id()}).
 
 -record(rabbit_fifo_dlx,

deps/rabbit/src/rabbit_fifo_dlx.hrl

@@ -375,13 +375,11 @@ ra_machine_config(Q) when ?is_amqqueue(Q) ->
     PolicyConfig = gather_policy_config(Q, true),
     QName = amqqueue:get_name(Q),
     {Name, _} = amqqueue:get_pid(Q),
-    PolicyConfig#{
-      name => Name,
-      queue_resource => QName,
-      become_leader_handler => {?MODULE, become_leader, [QName]},
-      single_active_consumer_on => single_active_consumer_on(Q),
-      created => erlang:system_time(millisecond)
-     }.
+    PolicyConfig#{name => Name,
+                  queue_resource => QName,
+                  single_active_consumer_on => single_active_consumer_on(Q),
+                  created => erlang:system_time(millisecond)
+                 }.
 
 resolve_delivery_limit(PolVal, ArgVal)
   when PolVal < 0 orelse ArgVal < 0 ->
@@ -679,13 +677,13 @@ handle_tick(QName,
               catch
                   _:Err ->
                       ?LOG_DEBUG("~ts: handle tick failed with ~p",
-                                       [rabbit_misc:rs(QName), Err]),
+                                 [rabbit_misc:rs(QName), Err]),
                       ok
               end
       end);
 handle_tick(QName, Config, _Nodes) ->
     ?LOG_DEBUG("~ts: handle tick received unexpected config format ~tp",
-                     [rabbit_misc:rs(QName), Config]).
+               [rabbit_misc:rs(QName), Config]).
 
 repair_leader_record(Q, Name) ->
     Node = node(),
@@ -696,7 +694,7 @@ repair_leader_record(Q, Name) ->
         _ ->
             QName = amqqueue:get_name(Q),
             ?LOG_DEBUG("~ts: updating leader record to current node ~ts",
-                             [rabbit_misc:rs(QName), Node]),
+                       [rabbit_misc:rs(QName), Node]),
             ok = become_leader0(QName, Name),
             ok
     end,

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -3633,16 +3633,18 @@ message_bytes_metrics(Config) ->
     wait_for_messages_pending_ack(Servers, RaName, 0),
     rabbit_ct_helpers:await_condition(
       fun() ->
-              {3, 3, 0} == get_message_bytes(Leader, QRes)
+              {M, M, 0} = get_message_bytes(Leader, QRes),
+              M > 0
       end, 30000),
 
+    {MsgSize, _, _} = get_message_bytes(Leader, QRes),
     subscribe(Ch, QQ, false),
 
     wait_for_messages_ready(Servers, RaName, 0),
     wait_for_messages_pending_ack(Servers, RaName, 1),
     rabbit_ct_helpers:await_condition(
       fun() ->
-              {3, 0, 3} == get_message_bytes(Leader, QRes)
+              {MsgSize, 0, MsgSize} == get_message_bytes(Leader, QRes)
       end, 30000),
 
     receive
@@ -3667,7 +3669,7 @@ message_bytes_metrics(Config) ->
     wait_for_messages_pending_ack(Servers, RaName, 1),
     rabbit_ct_helpers:await_condition(
       fun() ->
-              {3, 0, 3} == get_message_bytes(Leader, QRes)
+              {MsgSize, 0, MsgSize} == get_message_bytes(Leader, QRes)
       end, 30000),
 
     rabbit_ct_client_helpers:close_channel(Ch),
@@ -3676,7 +3678,7 @@ message_bytes_metrics(Config) ->
     wait_for_messages_pending_ack(Servers, RaName, 0),
     rabbit_ct_helpers:await_condition(
       fun() ->
-              {3, 3, 0} == get_message_bytes(Leader, QRes)
+              {MsgSize, MsgSize, 0} == get_message_bytes(Leader, QRes)
       end, 30000),
     ok.