QQ: record consumer metrics on the Ra leader, not the consumer's node

kjnilsson · kjnilsson · commit 479f654eb31a · 2026-07-15T14:52:21.000+01:00
consumer_created/updated/deleted rows were recorded on the consuming
channel's node, unlike other queue metrics which live on the Ra
leader. Cross-node updates went through an erpc:cast that silently
dropped on a disconnected node, leaving stale rows with no retry.

Move ownership to the leader: the checkout apply clause and
state_enter(leader/follower, ...) now write/clear these rows locally,
since mod_call effects only ever run on the current leader and must
never crash it. Also fan out a one-off cleanup on the v9 machine
version bump, to remove stale rows any node may have written under
the old scheme.
diff --git a/deps/rabbit/src/rabbit_fifo.erl b/deps/rabbit/src/rabbit_fifo.erl
@@ -260,7 +260,21 @@ update_config(Conf, State) ->
 apply(Meta, {machine_version, FromVersion, ToVersion}, VXState) ->
     %% machine version upgrades cant be done in apply_
     State = convert(Meta, FromVersion, ToVersion, VXState),
-    {State, ok, [{aux, {dlx, setup}}]};
+    Effects = case ToVersion of
+                 9 ->
+                     %% v9 moved consumer metrics ownership from the
+                     %% consuming channel's node to the current Ra leader;
+                     %% clear out any stale rows earlier versions may have
+                     %% left on other nodes and repopulate authoritatively
+                     #?STATE{cfg = #cfg{resource = QRes}} = State,
+                     [{mod_call, rabbit_quorum_queue,
+                       consumer_metrics_v9_upgrade,
+                       [QRes, consumer_metrics_rows(State)]},
+                      {aux, {dlx, setup}}];
+                 _ ->
+                     [{aux, {dlx, setup}}]
+             end,
+    {State, ok, Effects};
 apply(#{system_time := Ts} = Meta, Cmd,
       #?STATE{reclaimable_bytes = ReclBytes} = State) ->
     %% Add estimated reclaimable bytes.
@@ -492,7 +506,23 @@ apply_(#{index := Idx} = Meta,
     {Consumer, State1} = update_consumer(Meta, ConsumerKey, ConsumerId,
                                          ConsumerMeta, Spec, Priority,
                                          Timeout, State0),
-    {State2, Effs} = activate_next_consumer(State1, []),
+    WasActive = is_active(ConsumerKey, State1),
+    {State2, Effs0} = activate_next_consumer(State1, []),
+    {Active, ActivityStatus} = consumer_activity_status(ConsumerKey, State2),
+    Effs = case {WasActive, Active} of
+               {false, true} ->
+                   %% activate_next_consumer/2 has already emitted the
+                   %% metrics effect as part of promoting this consumer
+                   Effs0;
+               _ ->
+                   %% record (or refresh) the consumer metrics on the
+                   %% leader; needed here as activate_next_consumer/2 does
+                   %% not emit anything for the competing consumer strategy
+                   %% nor for a consumer that stays/becomes non-active
+                   [consumer_metrics_effect(State2, Consumer, Active,
+                                            ActivityStatus)
+                    | Effs0]
+           end,
 
     %% reply with a consumer infos
     Reply = {ok, consumer_info(ConsumerKey, Consumer, State2)},
@@ -1055,7 +1085,7 @@ state_enter(leader,
                     waiting_consumers = WaitingConsumers,
                     cfg = #cfg{resource = QRes,
                                dead_letter_handler = DLH},
-                    dlx = DlxState}) ->
+                    dlx = DlxState} = State) ->
     % return effects to monitor all current consumers and enqueuers
     Pids = lists:usort(maps:keys(Enqs)
                        ++ [P || ?CONSUMER_PID(P) <- maps:values(Cons)]
@@ -1064,7 +1094,12 @@ state_enter(leader,
     Nots = [{send_msg, P, leader_change, ra_event} || P <- Pids],
     NodeMons = lists:usort([{monitor, node, node(P)} || P <- Pids]),
     NotifyDecs = notify_decorators_startup(QRes),
-    Effects = Mons ++ Nots ++ NodeMons ++ [NotifyDecs],
+    %% (re-)populate the local consumer metrics table with the current
+    %% consumers of this queue: this node is now authoritative for them
+    ConsumerMetrics = {mod_call, rabbit_quorum_queue,
+                       bulk_update_consumer_metrics,
+                       [QRes, consumer_metrics_rows(State)]},
+    Effects = Mons ++ Nots ++ NodeMons ++ [NotifyDecs, ConsumerMetrics],
 
     case DLH of
         at_least_once ->
@@ -1073,6 +1108,18 @@ state_enter(leader,
             ok
     end,
     Effects;
+state_enter(follower, #?STATE{cfg = #cfg{resource = QRes,
+                                         dead_letter_handler = DLH},
+                              dlx = DlxState}) ->
+    %% this node is no longer authoritative for this queue's consumer
+    %% metrics, drop the locally cached rows
+    case DLH of
+        at_least_once ->
+            ensure_worker_terminated(DlxState);
+        _ ->
+            ok
+    end,
+    [{mod_call, rabbit_quorum_queue, delete_local_consumer_metrics, [QRes]}];
 state_enter(eol, #?STATE{enqueuers = Enqs,
                          consumers = Cons0,
                          waiting_consumers = WaitingConsumers0}) ->
@@ -1786,21 +1833,56 @@ cancel_consumer(Meta, ConsumerKey,
             end
     end.
 
-consumer_update_active_effects(#?STATE{cfg = #cfg{resource = QName}} = State,
+%% is this consumer currently entitled to receive deliveries and what its
+%% externally-visible activity status is, given the queue's consumer strategy
+consumer_activity_status(_ConsumerKey,
+                         #?STATE{cfg = #cfg{consumer_strategy = competing}}) ->
+    {true, up};
+consumer_activity_status(ConsumerKey,
+                         #?STATE{cfg = #cfg{consumer_strategy = single_active}} = State) ->
+    case is_active(ConsumerKey, State) of
+        true -> {true, single_active};
+        false -> {false, waiting}
+    end.
+
+consumer_metrics_effect(#?STATE{cfg = #cfg{resource = QName}},
+                        #consumer{cfg = #consumer_cfg{meta = Meta,
+                                                      pid = CPid,
+                                                      tag = CTag}},
+                        Active, ActivityStatus) ->
+    Ack = maps:get(ack, Meta, undefined),
+    Prefetch = maps:get(prefetch, Meta, undefined),
+    Args = maps:get(args, Meta, []),
+    {mod_call, rabbit_quorum_queue, update_consumer_handler,
+     [QName, {CTag, CPid}, false, Ack, Prefetch, Active, ActivityStatus, Args]}.
+
+%% metrics rows for every consumer of this queue, active or waiting, used to
+%% (re-)populate the local consumer metrics table when this node becomes leader
+consumer_metrics_rows(#?STATE{consumers = Cons,
+                              waiting_consumers = WaitingConsumers} = State) ->
+    [consumer_metrics_row(ConsumerKey, Consumer, State)
+     || {ConsumerKey, Consumer} <- maps:to_list(Cons) ++ WaitingConsumers].
+
+consumer_metrics_row(ConsumerKey,
+                     #consumer{cfg = #consumer_cfg{pid = Pid,
+                                                   tag = Tag,
+                                                   meta = Meta}},
+                     State) ->
+    {Active, ActivityStatus} = consumer_activity_status(ConsumerKey, State),
+    {Pid, Tag, maps:get(ack, Meta, undefined), maps:get(prefetch, Meta, undefined),
+     Active, ActivityStatus, maps:get(args, Meta, [])}.
+
+consumer_update_active_effects(State,
                                #consumer{cfg = #consumer_cfg{meta = Meta,
                                                              pid = CPid,
                                                              tag = CTag,
                                                              credit_mode = Mode},
                                          delivery_count = DeliveryCount,
                                          credit = Credit,
-                                         drain = Drain},
+                                         drain = Drain} = Consumer,
                                Active, ActivityStatus, Effects0) ->
-    Ack = maps:get(ack, Meta, undefined),
-    Prefetch = maps:get(prefetch, Meta, undefined),
-    Args = maps:get(args, Meta, []),
-    Effects = [{mod_call, rabbit_quorum_queue, update_consumer_handler,
-                [QName, {CTag, CPid}, false, Ack, Prefetch,
-                 Active, ActivityStatus, Args]} | Effects0],
+    Effects = [consumer_metrics_effect(State, Consumer, Active, ActivityStatus)
+               | Effects0],
     case Mode of
         {credited, _} when map_get(link_state_properties, Meta) =:= true ->
             Avail = case Active of
diff --git a/deps/rabbit/src/rabbit_quorum_queue.erl b/deps/rabbit/src/rabbit_quorum_queue.erl
@@ -34,8 +34,10 @@
          deliver/3]).
 -export([dead_letter_publish/5]).
 -export([cluster_state/1, status/1, status/2]).
--export([update_consumer_handler/8, update_consumer/9]).
--export([cancel_consumer_handler/2, cancel_consumer/3]).
+-export([update_consumer_handler/8]).
+-export([cancel_consumer_handler/2]).
+-export([bulk_update_consumer_metrics/2, delete_local_consumer_metrics/1]).
+-export([consumer_metrics_v9_upgrade/2]).
 -export([become_leader/2, handle_tick/3, spawn_deleter/1]).
 -export([rpc_delete_metrics/1,
          key_metrics_rpc/1]).
@@ -461,29 +463,70 @@ single_active_consumer_on(Q) ->
     QArguments = amqqueue:get_arguments(Q),
     table_lookup(QArguments, <<"x-single-active-consumer">>, false).
 
+%% these mod_call effects are always applied on the current Ra leader
+%% (see ra_server_proc:handle_effect/5), which is where quorum queue
+%% consumer metrics are recorded, so no cross-node dispatch is needed
 update_consumer_handler(QName, {ConsumerTag, ChPid}, Exclusive, AckRequired,
                         Prefetch, Active, ActivityStatus, Args) ->
-    catch rabbit_queue_type_util:local_or_remote_handler(ChPid, ?MODULE, update_consumer,
-                                                         [QName, ChPid, ConsumerTag,
-                                                          Exclusive, AckRequired,
-                                                          Prefetch, Active,
-                                                          ActivityStatus, Args]).
-
-update_consumer(QName, ChPid, ConsumerTag, Exclusive, AckRequired, Prefetch,
-                Active, ActivityStatus, Args) ->
     catch rabbit_core_metrics:consumer_updated(ChPid, ConsumerTag, Exclusive,
-                                               AckRequired,
-                                               QName, Prefetch, Active,
-                                               ActivityStatus, Args).
+                                               AckRequired, QName, Prefetch,
+                                               Active, ActivityStatus, Args).
 
 cancel_consumer_handler(QName, {ConsumerTag, ChPid}) ->
-    catch rabbit_queue_type_util:local_or_remote_handler(ChPid, ?MODULE, cancel_consumer,
-                                                         [QName, ChPid, ConsumerTag]).
-
-cancel_consumer(QName, ChPid, ConsumerTag) ->
     catch rabbit_core_metrics:consumer_deleted(ChPid, ConsumerTag, QName),
     rabbit_queue_type_util:notify_consumer_deleted(ChPid, ConsumerTag, QName, ?INTERNAL_USER).
 
+%% called when this node becomes the Ra leader for QName: the previous
+%% leader's local rows (if any) were already dropped when it stepped down
+%% (see delete_local_consumer_metrics/1), but bulk-replace defensively so
+%% this table is never a mix of two leader terms' data
+%%
+%% this mod_call effect (like all mod_call effects) is not protected by
+%% ra_server_proc, so a crash here would take down the whole Ra server;
+%% the consumer_created table may also legitimately not exist, e.g. in
+%% rabbit_fifo_int_SUITE which runs bare ra clusters without core_metrics
+bulk_update_consumer_metrics(QName, ConsumerMetrics) ->
+    catch bulk_update_consumer_metrics0(QName, ConsumerMetrics).
+
+bulk_update_consumer_metrics0(QName, ConsumerMetrics) ->
+    delete_local_consumer_metrics(QName),
+    %% quorum queues don't support exclusive consumers
+    [rabbit_core_metrics:consumer_created(ChPid, ConsumerTag, false,
+                                          AckRequired, QName, Prefetch,
+                                          Active, ActivityStatus, Args)
+     || {ChPid, ConsumerTag, AckRequired, Prefetch, Active,
+         ActivityStatus, Args} <- ConsumerMetrics],
+    ok.
+
+delete_local_consumer_metrics(QName) ->
+    catch ets:match_delete(consumer_created,
+                           {{QName, '_', '_'}, '_', '_', '_', '_', '_', '_'}),
+    ok.
+
+%% one-off cleanup run when a queue's machine version crosses into v9: prior
+%% versions wrote consumer_created from the consuming channel's node instead
+%% of the Ra leader's, so any node in the cluster could be holding a stale
+%% or duplicate row for this queue; wipe them all and let the current
+%% leader (the only node this mod_call effect runs on) repopulate
+%% authoritatively
+%%
+%% see bulk_update_consumer_metrics/2 for why this must never crash
+consumer_metrics_v9_upgrade(QName, ConsumerMetrics) ->
+    catch consumer_metrics_v9_upgrade0(QName, ConsumerMetrics).
+
+consumer_metrics_v9_upgrade0(QName, ConsumerMetrics) ->
+    case rabbit_amqqueue:lookup(QName) of
+        {ok, Q} when ?is_amqqueue(Q) ->
+            Nodes = get_nodes(Q),
+            _ = [_ = erpc_call(Node, ?MODULE, delete_local_consumer_metrics,
+                              [QName], ?RPC_TIMEOUT)
+                 || Node <- Nodes, Node =/= node()],
+            ok;
+        _ ->
+            ok
+    end,
+    bulk_update_consumer_metrics(QName, ConsumerMetrics).
+
 become_leader(_QName, _Name) ->
     %% noop now as we instead rely on the promt tick_timeout + repair to update
     %% the meta data store after a leader change
@@ -1254,38 +1297,16 @@ consume(Q, Spec, QState0) when ?amqqueue_is_quorum(Q) ->
                      %% in credit_reply
                      link_state_properties => true},
     case rabbit_fifo_client:checkout(ConsumerTag, Mode, ConsumerMeta, QState0) of
-        {ok, Infos, QState} ->
-            %% this info key was added in QQ v8
-            IsSac = maps:get(consumer_strategy, Infos, competing) == single_active,
-            case IsSac orelse single_active_consumer_on(Q) of
-                true ->
-                    ActivityStatus = case Infos of
-                                         #{is_active := true} ->
-                                             single_active;
-                                         _ ->
-                                             waiting
-                                     end,
-                    rabbit_core_metrics:consumer_created(ChPid, ConsumerTag,
-                                                         ExclusiveConsume,
-                                                         AckRequired, QName,
-                                                         Prefetch,
-                                                         ActivityStatus == single_active,
-                                                         ActivityStatus, Args),
-                    rabbit_queue_type_util:notify_consumer_created(
-                      ChPid, ConsumerTag, ExclusiveConsume, AckRequired,
-                      QName, Prefetch, Args, none, ActingUser),
-                    {ok, QState};
-                false ->
-                    rabbit_core_metrics:consumer_created(ChPid, ConsumerTag,
-                                                         ExclusiveConsume,
-                                                         AckRequired, QName,
-                                                         Prefetch, true,
-                                                         up, Args),
-                    rabbit_queue_type_util:notify_consumer_created(
-                      ChPid, ConsumerTag, ExclusiveConsume, AckRequired,
-                      QName, Prefetch, Args, none, ActingUser),
-                    {ok, QState}
-            end;
+        {ok, _Infos, QState} ->
+            %% the consumer_created/updated rabbit_core_metrics rows for this
+            %% consumer are recorded on the Ra leader as part of applying the
+            %% checkout command (see rabbit_fifo:consumer_metrics_effect/4),
+            %% not here, as this code may run on a different node than the
+            %% leader
+            rabbit_queue_type_util:notify_consumer_created(
+              ChPid, ConsumerTag, ExclusiveConsume, AckRequired,
+              QName, Prefetch, Args, none, ActingUser),
+            {ok, QState};
         Err ->
             consume_error(Err, QName)
     end.
diff --git a/deps/rabbit/test/rabbit_fifo_SUITE.erl b/deps/rabbit/test/rabbit_fifo_SUITE.erl
@@ -1231,6 +1231,7 @@ return_auto_checked_out_test(Config) ->
     {State1, #{key := CKey,
                next_msg_id := MsgId},
      [_Monitor,
+      {mod_call, rabbit_quorum_queue, update_consumer_handler, _},
       {log_ext, [1], _Fun1, _} ]} = checkout(Config, ?LINE, Cid, 1, State0),
     % return should include another delivery
     {State2, _, Effects} = apply(meta(Config, 3),
@@ -1246,7 +1247,7 @@ return_auto_checked_out_test(Config) ->
 
     {State4, #{key := CKey2,
                 next_msg_id := MsgId3},
-     [_, {log_ext, [1], _Fun3, _} ]} = checkout(Config, ?LINE, Cid, 1, State3),
+     [_, _, {log_ext, [1], _Fun3, _} ]} = checkout(Config, ?LINE, Cid, 1, State3),
 
     [{_, {_, #{delivery_count := 1,
                acquired_count := 2}}}]
@@ -1262,6 +1263,7 @@ requeue_test(Config) ->
     {State1, #{key := CKey,
                next_msg_id := MsgId},
      [_Monitor,
+      {mod_call, rabbit_quorum_queue, update_consumer_handler, _},
       {log_ext, [1], _Fun, _}]} = checkout(Config, ?LINE, Cid, 1, State0),
 
     [{MsgId, {H1, _}}] = rabbit_fifo:get_checked_out(CKey, MsgId, MsgId, State1),
@@ -2096,8 +2098,9 @@ single_active_consumer_state_enter_leader_include_waiting_consumers_test(Config)
     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)).
+    %% 1 for decorators,
+    %% 1 for the bulk consumer metrics resync
+    ?assertEqual(2 * 3 + 1 + 1 + 1, length(Effects)).
 
 single_active_consumer_state_enter_eol_include_waiting_consumers_test(Config) ->
     Resource = rabbit_misc:r("/", queue, ?FUNCTION_NAME_B),
@@ -2480,10 +2483,13 @@ single_active_consumer_priority_test(Config) ->
     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(4, ModCalls)),
+    %% C3 is added as waiting (the active consumer, C2, still has a message
+    %% checked out so cannot be pre-empted immediately)
+    assert_update_consumer_handler_state_transition(C3, Resource, false, waiting, lists:nth(4, ModCalls)),
+    %% C3 is promoted to single_active once C2's message is settled
+    assert_update_consumer_handler_state_transition(C3, Resource, true, single_active, lists:nth(5, ModCalls)),
     %% C2 should transition as waiting
-    assert_update_consumer_handler_state_transition(C2, Resource, false, waiting, lists:nth(5, ModCalls)),
+    assert_update_consumer_handler_state_transition(C2, Resource, false, waiting, lists:nth(6, ModCalls)),
 
     ok.
 
