Quorum Queue notifies AMQP client of Single Active Consumer state change (#15736)

* Notify AMQP 1.0 clients of Single Active Consumer status changes

Solves #14726

Quorum queues support the Single Active Consumer (SAC) feature, but
until now clients had no reliable way to learn whether they are the
active consumer or a waiting one. This commit adds consumer activity
notifications so that AMQP 1.0 consumers learn about their active/inactive
status after attaching and whenever it changes.

The notification is pushed via the `properties` field of the AMQP 1.0
flow frame, using a key `rabbitmq:active` with a boolean value.
This property is pushed to the client only upon SAC activity state
change, i.e. not in every flow frame to keep network traffic low.

On the server side, the existing `credit_reply` message sent from the
queue process to the session process is extended from a 6-tuple to a
7-tuple, where the 7th element is a map of link state properties
(e.g. `#{active => true}`). This reuses the existing communication
mechanism rather than introducing a new Ra event, and maps naturally to
the flow frame's `properties` field. It also allows future extensibility
for other queue types to convey additional link state information.

`rabbit_quorum_queue:consume/3` opts in by setting
`link_state_properties => true` in the consumer metadata. `rabbit_fifo`
uses this flag to decide whether to emit the 7-tuple (with properties)
or the old 6-tuple (without). This is how mixed-version cluster
compatibility is maintained: an old queue process sends the old 6-tuple
which `rabbit_amqp_session` converts to a 7-tuple with an empty map, and
a new queue process only sends the 7-tuple when the consumer opted in.

On the client side, amqp10_client emits a generic
`{amqp10_event, {link, LinkRef, {state_properties, Map}}}` event
whenever the flow frame's properties field changes, without baking in
any RabbitMQ-specific knowledge.

* Send link state properties also for non-SAC QQ

Author: ansd

deps/amqp10_client/src/amqp10_client.erl

@@ -230,7 +230,7 @@ attach_sender_link(Session, Name, Target, SettleMode, Durability) ->
                                       durable => Durability}},
                    snd_settle_mode => SettleMode,
                    rcv_settle_mode => first},
-    amqp10_client_session:attach(Session, AttachArgs).
+    attach_link(Session, AttachArgs).
 
 %% @doc Attaches a receiver link to a source.
 %% This is asynchronous and will notify completion of the attach request to the
@@ -307,7 +307,7 @@ attach_receiver_link(Session, Name, Source, SettleMode, Durability, Filter, Prop
                    filter => Filter,
                    properties => Properties,
                    raw_mode => RawMode},
-    amqp10_client_session:attach(Session, AttachArgs).
+    attach_link(Session, AttachArgs).
 
 -spec attach_link(pid(), attach_args()) -> {ok, link_ref()}.
 attach_link(Session, AttachArgs) ->

deps/amqp10_client/src/amqp10_client_session.erl

@@ -101,7 +101,8 @@
                          max_message_size => max_message_size(),
                          handle => output_handle(),
                          footer_opt => footer_opt(),
-                         raw_mode => boolean()
+                         raw_mode => boolean(),
+                         notify_when_state_properties_changed => boolean()
                         }.
 
 -type transfer_error() :: {error,
@@ -144,7 +145,9 @@
                                Credit :: pos_integer()},
          incoming_unsettled = #{} :: #{delivery_number() => ok},
          footer_opt :: footer_opt() | undefined,
-         raw_mode = false :: boolean()
+         raw_mode = false :: boolean(),
+         %% link state properties from the flow frame
+         state_properties :: ignore | #{binary() => term()}
         }).
 
 -record(state,
@@ -414,9 +417,9 @@ mapped(cast, #'v1_0.flow'{handle = {uint, InHandle}} = Flow,
     {ok, #link{output_handle = OutHandle} = Link0} =
         find_link_by_input_handle(InHandle, State),
 
-     % TODO: handle `send_flow` return tag
-    {ok, Link} = handle_link_flow(Flow, Link0),
-    ok = maybe_notify_link_credit(Link0, Link),
+    {ok, Link1} = handle_link_flow(Flow, Link0),
+    ok = maybe_notify_link_credit(Link0, Link1),
+    Link = maybe_notify_link_state_properties(Flow, Link1),
     Links1 = Links#{OutHandle := Link},
     State1 = State#state{links = Links1},
     {keep_state, State1};
@@ -976,6 +979,10 @@ send_attach(Send, #{name := Name, role := RoleTuple} = Args, {FromPid, _},
     ok = Send(Attach, State),
 
     Ref = make_link_ref(Role, self(), OutHandle),
+    StateProps =  case maps:get(notify_when_state_properties_changed, Args, false) of
+                      true -> maps:new();
+                      _ -> ignore
+                  end,
     Link = #link{name = Name,
                  ref = Ref,
                  output_handle = OutHandle,
@@ -987,8 +994,8 @@ send_attach(Send, #{name := Name, role := RoleTuple} = Args, {FromPid, _},
                  delivery_count = unpack(InitialDeliveryCount),
                  max_message_size = unpack(MaxMessageSize),
                  footer_opt = maps:get(footer_opt, Args, undefined),
-                 raw_mode = maps:get(raw_mode, Args, false)},
-
+                 raw_mode = maps:get(raw_mode, Args, false),
+                 state_properties = StateProps},
     {State#state{links = Links#{OutHandle => Link},
                  next_link_handle = NextLinkHandle,
                  link_index = LinkIndex#{{Role, Name} => OutHandle}}, Ref}.
@@ -1031,11 +1038,17 @@ handle_link_flow(#'v1_0.flow'{delivery_count = TheirDC,
                               link_credit = {uint, TheirCredit},
                               available = Available,
                               drain = Drain0},
-                 Link0 = #link{role = receiver}) ->
+                 #link{role = receiver,
+                       link_credit = OurCredit} = Link0) ->
     Drain = default(Drain0, false),
     Link = case Drain andalso TheirCredit =:= 0 of
                true ->
-                   notify_credit_exhausted(Link0),
+                   case OurCredit > 0 of
+                       true ->
+                           notify_credit_exhausted(Link0);
+                       false ->
+                           ok
+                   end,
                    Link0#link{delivery_count = unpack(TheirDC),
                               link_credit = 0,
                               available = unpack(Available),
@@ -1056,9 +1069,11 @@ find_link_by_input_handle(InHandle, #state{link_handle_index = LHI,
             case Links of
                 #{OutHandle := Link} ->
                     {ok, Link};
-                _ -> not_found
+                _ ->
+                    not_found
             end;
-        _ -> not_found
+        _ ->
+            not_found
     end.
 
 with_link(InHandle, State, Fun) ->
@@ -1120,6 +1135,19 @@ maybe_notify_link_credit(#link{role = sender,
 maybe_notify_link_credit(_Old, _New) ->
     ok.
 
+maybe_notify_link_state_properties(#'v1_0.flow'{properties = {map, KVList}},
+                                   #link{state_properties = #{} = OldProps} = Link)
+  when is_list(KVList) ->
+    case #{Key => unpack(Val) || {{symbol, Key}, Val} <- KVList} of
+        OldProps ->
+            Link;
+        NewProps ->
+            notify_link(Link, {state_properties, NewProps}),
+            Link#link{state_properties = NewProps}
+    end;
+maybe_notify_link_state_properties(#'v1_0.flow'{}, Link) ->
+    Link.
+
 notify_link_attached(Link, Perf, #state{connection_config = Cfg}) ->
     What = case Cfg of
                #{notify_with_performative := true} ->

deps/rabbit/include/rabbit_queue_type.hrl

@@ -0,0 +1,21 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2026 Broadcom. All Rights Reserved. The term "Broadcom"
+%% refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+
+-include_lib("amqp10_common/include/amqp10_types.hrl").
+
+%% see AMQP 1.0 §2.6.7
+-type delivery_count() :: sequence_no().
+-type credit() :: uint().
+
+-type link_state_properties() :: #{atom() => term()}.
+
+-record(credit_reply, {ctag :: rabbit_types:ctag(),
+                       delivery_count :: delivery_count(),
+                       credit :: credit(),
+                       available :: non_neg_integer(),
+                       drain :: boolean(),
+                       properties :: link_state_properties()}).