Merge pull request #15166 from rabbitmq/rabbitmq-server-13873

#13873 by @Ayanda-D, polished and rebased

Author: michaelklishin

.github/workflows/authorization-server-make.yaml

@@ -11,15 +11,15 @@ on:
     paths:
       - .github/workflows/authorization-server-make.yaml
       - selenium/authorization-server
-  
+
 env:
   REGISTRY_IMAGE: pivotalrabbitmq/spring-authorization-server
   IMAGE_TAG: 0.0.11
 jobs:
   docker:
     runs-on: ubuntu-latest
     steps:
-      
+
       - name: CHECKOUT REPOSITORY
         uses: actions/checkout@v6
 
@@ -28,10 +28,10 @@ jobs:
         with:
           username: ${{ secrets.DOCKERHUB_USERNAME }}
           password: ${{ secrets.DOCKERHUB_PASSWORD }}
-      
+
       - name: Build and push
         uses: docker/build-push-action@v6
         with:
-          context: selenium/authorization-server        
+          context: selenium/authorization-server
           push: true
           tags: ${{ env.REGISTRY_IMAGE }}:${{ env.IMAGE_TAG }}

.github/workflows/test-management-ui-for-pr.yaml

@@ -78,4 +78,4 @@ jobs:
       with:
         name: test-artifacts-${{ matrix.browser }}-${{ matrix.erlang_version }}
         path: ${{ env.SELENIUM_ARTIFACTS }}/*
- 
+

deps/rabbit/priv/schema/rabbit.schema

@@ -2859,6 +2859,11 @@ end}.
 {mapping, "stream.read_ahead", "rabbit.stream_read_ahead",
     [{datatype, [{enum, [true, false]}, integer, string]}]}.
 
+{mapping, "stream.read_ahead_limit", "rabbit.stream_read_ahead_limit", [
+    {datatype, [integer, string]},
+    {validators, ["is_supported_information_unit"]}
+]}.
+
 {mapping, "cluster_tags.$tag", "rabbit.cluster_tags", [
     {datatype, [binary]}
 ]}.

deps/rabbit/src/rabbit_amqp_session.erl

@@ -2613,6 +2613,7 @@ rejected(QNameBin, down) ->
                           [{{symbol, <<"queue">>}, {utf8, QNameBin}},
                            {{symbol, <<"reason">>}, {symbol, <<"unavailable">>}}]}}}.
 
+
 maybe_grant_link_credit(Credit, MaxLinkCredit, DeliveryCount, NumUnconfirmed, Handle) ->
     case grant_link_credit(Credit, MaxLinkCredit, NumUnconfirmed) of
         true ->

deps/rabbit/src/rabbit_fifo.erl

@@ -1638,27 +1638,27 @@ drop_head(#?STATE{ra_indexes = Indexes0} = State0, Effects) ->
             #?STATE{cfg = #cfg{dead_letter_handler = DLH},
                     dlx = DlxState} = State = State3,
             {_, DlxEffects} = rabbit_fifo_dlx:discard([Msg], maxlen, DLH, DlxState),
-            {State, combine_effects(DlxEffects, Effects)};
+            {State, add_drop_head_effects(DlxEffects, Effects)};
         empty ->
             {State0, Effects}
     end.
 
-%% combine global counter update effects to avoid bulding a huge list of
-%% effects if many messages are dropped at the same time as could happen
-%% when the `max_length' is changed via a configuration update.
-combine_effects([{mod_call,
-                  rabbit_global_counters,
-                  messages_dead_lettered,
-                  [Reason, rabbit_quorum_queue, Type, NewLen]}],
-                [{mod_call,
-                  rabbit_global_counters,
-                  messages_dead_lettered,
-                  [Reason, rabbit_quorum_queue, Type, PrevLen]} | Rem]) ->
+add_drop_head_effects([{mod_call,
+                        rabbit_global_counters,
+                        messages_dead_lettered,
+                        [Reason, rabbit_quorum_queue, Type, NewLen]}],
+                      [{mod_call,
+                        rabbit_global_counters,
+                        messages_dead_lettered,
+                        [Reason, rabbit_quorum_queue, Type, PrevLen]} | Rem]) ->
+    %% combine global counter update effects to avoid bulding a huge list of
+    %% effects if many messages are dropped at the same time as could happen
+    %% when the `max_length' is changed via a configuration update.
     [{mod_call,
       rabbit_global_counters,
       messages_dead_lettered,
       [Reason, rabbit_quorum_queue, Type, PrevLen + NewLen]} | Rem];
-combine_effects(New, Old) ->
+add_drop_head_effects(New, Old) ->
     New ++ Old.
 
 maybe_set_msg_ttl(Msg, RaCmdTs, Header,

deps/rabbit/src/rabbit_plugins.erl

@@ -175,6 +175,7 @@ list() ->
     PluginsPath = plugins_dir(),
     list(PluginsPath).
 
+
 %% @doc Get the list of plugins which are ready to be enabled.
 
 -spec list(string()) -> [#plugin{}].

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -1515,29 +1515,20 @@ shrink_all(Node) ->
             amqqueue:get_type(Q) == ?MODULE,
             lists:member(Node, get_nodes(Q))].
 
-
+-spec grow(node() | integer(), binary(), binary(), all | even) ->
+    [{rabbit_amqqueue:name(),
+      {ok, pos_integer()} | {error, pos_integer(), term()}}].
 grow(Node, VhostSpec, QueueSpec, Strategy) ->
     grow(Node, VhostSpec, QueueSpec, Strategy, promotable).
 
--spec grow(node(), binary(), binary(), all | even, membership()) ->
+-spec grow(node() | integer(), binary(), binary(), all | even, membership()) ->
     [{rabbit_amqqueue:name(),
       {ok, pos_integer()} | {error, pos_integer(), term()}}].
-grow(Node, VhostSpec, QueueSpec, Strategy, Membership) ->
+grow(Node, VhostSpec, QueueSpec, Strategy, Membership) when is_atom(Node) ->
     Running = rabbit_nodes:list_running(),
     [begin
          Size = length(get_nodes(Q)),
-         QName = amqqueue:get_name(Q),
-         ?LOG_INFO("~ts: adding a new member (replica) on node ~w",
-                         [rabbit_misc:rs(QName), Node]),
-         case add_member(Q, Node, Membership) of
-             ok ->
-                 {QName, {ok, Size + 1}};
-             {error, Err} ->
-                 ?LOG_WARNING(
-                   "~ts: failed to add member (replica) on node ~w, error: ~w",
-                   [rabbit_misc:rs(QName), Node, Err]),
-                 {QName, {error, Size, Err}}
-         end
+         maybe_grow(Q, Node, Membership, Size)
      end
      || Q <- rabbit_amqqueue:list(),
         amqqueue:get_type(Q) == ?MODULE,
@@ -1547,7 +1538,91 @@ grow(Node, VhostSpec, QueueSpec, Strategy, Membership) ->
         lists:member(Node, Running),
         matches_strategy(Strategy, get_nodes(Q)),
         is_match(amqqueue:get_vhost(Q), VhostSpec) andalso
-        is_match(get_resource_name(amqqueue:get_name(Q)), QueueSpec) ].
+        is_match(get_resource_name(amqqueue:get_name(Q)), QueueSpec) ];
+
+grow(QuorumClusterSize, VhostSpec, QueueSpec, Strategy, Membership)
+  when is_integer(QuorumClusterSize), QuorumClusterSize > 0 ->
+    Running = rabbit_nodes:list_running(),
+    TotalRunning = length(Running),
+
+    TargetQuorumClusterSize =
+        if QuorumClusterSize > TotalRunning ->
+            %% we can't grow beyond total running nodes
+            TotalRunning;
+        true ->
+            QuorumClusterSize
+        end,
+
+    lists:flatten(
+        [begin
+            QNodes = get_nodes(Q),
+            case length(QNodes) of
+                Size when Size < TargetQuorumClusterSize ->
+                    TargetAvailableNodes = Running -- QNodes,
+                    N = length(TargetAvailableNodes),
+                    Node = lists:nth(rand:uniform(N), TargetAvailableNodes),
+                    maybe_grow(Q, Node, Membership, Size);
+                _ ->
+                    []
+            end
+        end
+        ||  _ <- lists:seq(1, TargetQuorumClusterSize),
+            Q <- rabbit_amqqueue:list(),
+            amqqueue:get_type(Q) == ?MODULE,
+            matches_strategy(Strategy, get_nodes(Q)),
+            is_match(amqqueue:get_vhost(Q), VhostSpec) andalso
+            is_match(get_resource_name(amqqueue:get_name(Q)), QueueSpec)]);
+
+grow(QuorumClusterSize, _VhostSpec, _QueueSpec, _Strategy, _Membership)
+  when is_integer(QuorumClusterSize) ->
+    rabbit_log:warning(
+            "cannot grow queues to a quorum cluster size less than zero (~tp)",
+            [QuorumClusterSize]),
+    {error, bad_quorum_cluster_size}.
+
+maybe_grow(Q, Node, Membership, Size) ->
+    QNodes = get_nodes(Q),
+    maybe_grow(Q, Node, Membership, Size, QNodes).
+
+maybe_grow(Q, Node, Membership, Size, QNodes) ->
+    QName = amqqueue:get_name(Q),
+    {ok, RaName} = qname_to_internal_name(QName),
+    case check_all_memberships(RaName, QNodes, voter) of
+        true ->
+            ?LOG_INFO("~ts: adding a new member (replica) on node ~w",
+                            [rabbit_misc:rs(QName), Node]),
+            case add_member(Q, Node, Membership) of
+                ok ->
+                    {QName, {ok, Size + 1}};
+                {error, Err} ->
+                    ?LOG_WARNING(
+                    "~ts: failed to add member (replica) on node ~w, error: ~w",
+                    [rabbit_misc:rs(QName), Node, Err]),
+                    {QName, {error, Size, Err}}
+            end;
+        false ->
+            Err = {error, non_voters_found},
+            ?LOG_WARNING(
+                    "~ts: failed to add member (replica) on node ~w, error: ~w",
+                    [rabbit_misc:rs(QName), Node, Err]),
+            {QName, {error, Size, Err}}
+    end.
+
+%% Compare local membership states of all nodes in parallel.
+%%
+%% Note a few things:
+%% 1. This function intentionally queries local member state and not the leader
+%% 2. ra:key_metrics/1 is sequential and not parallel
+%% 3. ra:key_metrics/1 is not multicall-friendly because it relies on erlang:node/0
+check_all_memberships(RaName, QNodes, CompareMembership) ->
+    case rpc:multicall(QNodes, ets, lookup, [ra_state, RaName]) of
+        {Result, []} ->
+            lists:all(
+                fun(M) -> M == CompareMembership end,
+                [Membership || [{_RaName, _RaState, Membership}] <- Result]);
+        _ ->
+            false
+    end.
 
 -spec transfer_leadership(amqqueue:amqqueue(), node()) ->
     {migrated, node()} | {not_migrated, atom()}.

deps/rabbit/src/rabbit_stream_queue.erl

@@ -535,6 +535,7 @@ credit(QName, CTag, DeliveryCountRcv, LinkCreditRcv, Drain,
 
 supports_stateful_delivery() -> true.
 
+
 deliver(QSs, Msg, Options) ->
     lists:foldl(
       fun({Q, stateless}, {Qs, Actions}) ->

deps/rabbit/test/quorum_queue_SUITE.erl

@@ -116,7 +116,8 @@ groups() ->
                                             node_removal_is_not_quorum_critical,
                                             select_nodes_with_least_replicas,
                                             select_nodes_with_least_replicas_node_down,
-                                            subscribe_from_each
+                                            subscribe_from_each,
+                                            grow_queue
 
 
                                            ]},
@@ -1768,6 +1769,116 @@ dont_leak_file_handles(Config) ->
     rabbit_ct_client_helpers:close_channel(C),
     ok.
 
+grow_queue(Config) ->
+    [Server0, Server1, Server2, _Server3, _Server4] =
+        rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+
+    Ch = rabbit_ct_client_helpers:open_channel(Config, Server0),
+    QQ = ?config(queue_name, Config),
+    AQ = ?config(alt_queue_name, Config),
+    ?assertEqual({'queue.declare_ok', QQ, 0, 0},
+                 declare(Ch, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>},
+                                  {<<"x-quorum-initial-group-size">>, long, 5}])),
+    ?assertEqual({'queue.declare_ok', AQ, 0, 0},
+                 declare(Ch, AQ, [{<<"x-queue-type">>, longstr, <<"quorum">>},
+                                  {<<"x-quorum-initial-group-size">>, long, 5}])),
+
+    QQs = [QQ, AQ],
+    MsgCount = 3,
+
+    [begin
+        RaName = ra_name(Q),
+        rabbit_ct_client_helpers:publish(Ch, Q, MsgCount),
+        wait_for_messages_ready([Server0], RaName, MsgCount),
+        {ok, Q0} = rpc:call(Server0, rabbit_amqqueue, lookup, [Q, <<"/">>]),
+        #{nodes := Nodes0} = amqqueue:get_type_state(Q0),
+        ?assertEqual(5, length(Nodes0))
+    end || Q <- QQs],
+
+    rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue,
+        force_all_queues_shrink_member_to_current_member, []),
+
+    TargetClusterSize_1 = 1,
+    assert_grown_queues(QQs, Server0, TargetClusterSize_1, MsgCount),
+
+    %% grow queues to node 'Server1'
+    TargetClusterSize_2 = 2,
+    Result1 = rpc:call(Server0, rabbit_quorum_queue, grow, [Server1, <<"/">>, <<".*">>, all]),
+    %% [{{resource,<<"/">>,queue,<<"grow_queue">>},{ok,2}},
+    %%  {{resource,<<"/">>,queue,<<"grow_queue_alt">>},{ok,2}},...]
+    ?assert(lists:all(fun({_, {R, _}}) -> R =:= ok end, Result1)),
+    assert_grown_queues(QQs, Server0, TargetClusterSize_2, MsgCount),
+
+    %% grow queues to quorum cluster size '2' has no effect
+    Result2 = rpc:call(Server0, rabbit_quorum_queue, grow, [TargetClusterSize_2, <<"/">>, <<".*">>, all]),
+    ?assertEqual([], Result2),
+    assert_grown_queues(QQs, Server0, TargetClusterSize_2, MsgCount),
+
+    %% grow queues to quorum cluster size '3'
+    TargetClusterSize_3 = 3,
+    Result3 = rpc:call(Server0, rabbit_quorum_queue, grow, [TargetClusterSize_3, <<"/">>, <<".*">>, all, voter]),
+    ?assert(lists:all(fun({_, {R, _}}) -> R =:= ok end, Result3)),
+    assert_grown_queues(QQs, Server0, TargetClusterSize_3, MsgCount),
+
+    %% grow queues to quorum cluster size '5'
+    TargetClusterSize_5 = 5,
+    Result4 = rpc:call(Server0, rabbit_quorum_queue, grow, [TargetClusterSize_5, <<"/">>, <<".*">>, all, voter]),
+    ?assert(lists:all(fun({_, {R, _}}) -> R =:= ok end, Result4)),
+    assert_grown_queues(QQs, Server0, TargetClusterSize_5, MsgCount),
+
+    %% shrink all queues again down to 1 member
+    rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue,
+        force_all_queues_shrink_member_to_current_member, []),
+    assert_grown_queues(QQs, Server0, TargetClusterSize_1, MsgCount),
+
+    %% grow queues to quorum cluster size > '5' (limit = 5).
+    TargetClusterSize_10 = 10,
+    Result5 = rpc:call(Server0, rabbit_quorum_queue, grow, [TargetClusterSize_10, <<"/">>, <<".*">>, all]),
+    ?assert(lists:all(fun({_, {R, _}}) -> R =:= ok end, Result5)),
+    assert_grown_queues(QQs, Server0, TargetClusterSize_5, MsgCount),
+
+    %% shrink all queues again down to 1 member
+    rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue,
+        force_all_queues_shrink_member_to_current_member, []),
+    assert_grown_queues(QQs, Server0, TargetClusterSize_1, MsgCount),
+
+    %% attempt to grow queues to quorum cluster size < '0'.
+    BadTargetClusterSize = -5,
+    ?assertEqual({error, bad_quorum_cluster_size},
+        rpc:call(Server0, rabbit_quorum_queue, grow, [BadTargetClusterSize, <<"/">>, <<".*">>, all])),
+
+    %% shrink all queues again down to 1 member
+    rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue,
+        force_all_queues_shrink_member_to_current_member, []),
+    assert_grown_queues(QQs, Server0, TargetClusterSize_1, MsgCount),
+
+    %% grow queues to node 'Server1': non_voter
+    rpc:call(Server0, rabbit_quorum_queue, grow, [Server1, <<"/">>, <<".*">>, all, non_voter]),
+    assert_grown_queues(QQs, Server0, TargetClusterSize_2, MsgCount),
+
+    %% grow queues to node 'Server2': fail, non_voters found
+    Result6 = rpc:call(Server0, rabbit_quorum_queue, grow, [Server2, <<"/">>, <<".*">>, all, voter]),
+    %% [{{resource,<<"/">>,queue,<<"grow_queue">>},{error, 2, {error, non_voters_found}},
+    %%  {{resource,<<"/">>,queue,<<"grow_queue_alt">>},{error, 2, {error, non_voters_found}},...]
+    ?assert(lists:all(
+        fun({_, Err}) -> Err =:= {error, TargetClusterSize_2, {error, non_voters_found}} end, Result6)),
+    assert_grown_queues(QQs, Server0, TargetClusterSize_2, MsgCount),
+
+    %% grow queues to target quorum cluster size '5': fail, non_voters found
+    Result7 = rpc:call(Server0, rabbit_quorum_queue, grow, [TargetClusterSize_5, <<"/">>, <<".*">>, all]),
+    ?assert(lists:all(
+        fun({_, Err}) -> Err =:= {error, TargetClusterSize_2, {error, non_voters_found}} end, Result7)),
+    assert_grown_queues(QQs, Server0, TargetClusterSize_2, MsgCount).
+
+assert_grown_queues(Qs, Node, TargetClusterSize, MsgCount) ->
+    [begin
+        RaName = ra_name(Q),
+        wait_for_messages_ready([Node], RaName, MsgCount),
+        {ok, Q0} = rpc:call(Node, rabbit_amqqueue, lookup, [Q, <<"/">>]),
+        #{nodes := Nodes0} = amqqueue:get_type_state(Q0),
+        ?assertEqual(TargetClusterSize, length(Nodes0))
+    end || Q <- Qs].
+
 gh_12635(Config) ->
     % https://github.com/rabbitmq/rabbitmq-server/issues/12635
     [Server0, _Server1, Server2] =