Reconcile (repair or expand) quorum queue membership periodically

deps/rabbit/BUILD.bazel

@@ -5,6 +5,7 @@ load("//:rabbitmq_home.bzl", "rabbitmq_home")
 load("//:rabbitmq_run.bzl", "rabbitmq_run")
 load(
     "//:rabbitmq.bzl",
+    "ENABLE_FEATURE_MAYBE_EXPR",
     "RABBITMQ_DIALYZER_OPTS",
     "assert_suites",
     "rabbitmq_app",
@@ -330,6 +331,14 @@ rabbitmq_integration_suite(
     size = "medium",
 )
 
+rabbitmq_integration_suite(
+    name = "quorum_queue_member_reconciliation_SUITE",
+    size = "medium",
+    additional_beam = [
+        ":test_quorum_queue_utils_beam",
+    ],
+)
+
 rabbitmq_integration_suite(
     name = "clustering_management_SUITE",
     size = "large",
@@ -665,6 +674,7 @@ rabbitmq_integration_suite(
     additional_beam = [
         ":test_quorum_queue_utils_beam",
     ],
+    flaky = True,
     shard_count = 6,
 )
 
@@ -1230,5 +1240,6 @@ eunit(
         ":test_test_util_beam",
         ":test_test_rabbit_event_handler_beam",
     ],
+    erl_extra_args = [ENABLE_FEATURE_MAYBE_EXPR],
     target = ":test_erlang_app",
 )

deps/rabbit/app.bzl

@@ -192,6 +192,7 @@ def all_beam_files(name = "all_beam_files"):
             "src/rabbit_queue_type_util.erl",
             "src/rabbit_quorum_memory_manager.erl",
             "src/rabbit_quorum_queue.erl",
+            "src/rabbit_quorum_queue_periodic_membership_reconciliation.erl",
             "src/rabbit_ra_registry.erl",
             "src/rabbit_ra_systems.erl",
             "src/rabbit_reader.erl",
@@ -436,6 +437,7 @@ def all_test_beam_files(name = "all_test_beam_files"):
             "src/rabbit_queue_type_util.erl",
             "src/rabbit_quorum_memory_manager.erl",
             "src/rabbit_quorum_queue.erl",
+            "src/rabbit_quorum_queue_periodic_membership_reconciliation.erl",
             "src/rabbit_ra_registry.erl",
             "src/rabbit_ra_systems.erl",
             "src/rabbit_reader.erl",
@@ -698,6 +700,7 @@ def all_srcs(name = "all_srcs"):
             "src/rabbit_queue_type_util.erl",
             "src/rabbit_quorum_memory_manager.erl",
             "src/rabbit_quorum_queue.erl",
+            "src/rabbit_quorum_queue_periodic_membership_reconciliation.erl",
             "src/rabbit_ra_registry.erl",
             "src/rabbit_ra_systems.erl",
             "src/rabbit_reader.erl",
@@ -802,6 +805,7 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         erlc_opts = "//:test_erlc_opts",
         deps = ["//deps/amqp_client:erlang_app"],
     )
+
     erlang_bytecode(
         name = "cluster_rename_SUITE_beam_files",
         testonly = True,
@@ -1972,3 +1976,12 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         erlc_opts = "//:test_erlc_opts",
         deps = ["//deps/amqp_client:erlang_app"],
     )
+    erlang_bytecode(
+        name = "quorum_queue_member_reconciliation_SUITE_beam_files",
+        testonly = True,
+        srcs = ["test/quorum_queue_member_reconciliation_SUITE.erl"],
+        outs = ["test/quorum_queue_member_reconciliation_SUITE.beam"],
+        app_name = "rabbit",
+        erlc_opts = "//:test_erlc_opts",
+        deps = ["//deps/amqp_client:erlang_app", "//deps/rabbitmq_ct_helpers:erlang_app"],
+    )

deps/rabbit/priv/schema/rabbit.schema

@@ -2487,6 +2487,30 @@ end}.
 {mapping, "quorum_queue.property_equivalence.relaxed_checks_on_redeclaration", "rabbit.quorum_relaxed_checks_on_redeclaration", [
     {datatype, {enum, [true, false]}}]}.
 
+
+%%
+%% Quorum Queue membership reconciliation
+%%
+
+{mapping, "quorum_queue.continuous_membership_reconciliation.enabled", "rabbit.quorum_membership_reconciliation_enabled", [
+    {datatype, {enum, [true, false]}}]}.
+
+{mapping, "quorum_queue.continuous_membership_reconciliation.auto_remove", "rabbit.quorum_membership_reconciliation_auto_remove", [
+    {datatype, {enum, [true, false]}}]}.
+
+{mapping, "quorum_queue.continuous_membership_reconciliation.interval", "rabbit.quorum_membership_reconciliation_interval", [
+    {datatype, integer}, {validators, ["non_negative_integer"]}
+]}.
+
+{mapping, "quorum_queue.continuous_membership_reconciliation.trigger_interval", "rabbit.quorum_membership_reconciliation_trigger_interval", [
+    {datatype, integer}, {validators, ["non_negative_integer"]}
+]}.
+
+{mapping, "quorum_queue.continuous_membership_reconciliation.target_group_size", "rabbit.quorum_membership_reconciliation_target_group_size", [
+    {datatype, integer}, {validators, ["non_negative_integer"]}
+]}.
+
+
 %%
 %% Runtime parameters
 %%

deps/rabbit/src/rabbit.erl

@@ -175,6 +175,12 @@
                     {requires,    [rabbit_alarm, guid_generator]},
                     {enables,     core_initialized}]}).
 
+-rabbit_boot_step({rabbit_quorum_queue_periodic_membership_reconciliation,
+                   [{description, "Quorums Queue membership reconciliation"},
+                    {mfa,         {rabbit_sup, start_restartable_child,
+                                   [rabbit_quorum_queue_periodic_membership_reconciliation]}},
+                    {requires, [database]}]}).
+
 -rabbit_boot_step({rabbit_epmd_monitor,
                    [{description, "epmd monitor"},
                     {mfa,         {rabbit_sup, start_restartable_child,

deps/rabbit/src/rabbit_node_monitor.erl

@@ -815,6 +815,7 @@ handle_dead_rabbit(Node, State = #state{partitions = Partitions,
     ok = rabbit_amqqueue:on_node_down(Node),
     ok = rabbit_alarm:on_node_down(Node),
     ok = rabbit_mnesia:on_node_down(Node),
+    ok = rabbit_quorum_queue_periodic_membership_reconciliation:on_node_down(Node),
     %% If we have been partitioned, and we are now in the only remaining
     %% partition, we no longer care about partitions - forget them. Note
     %% that we do not attempt to deal with individual (other) partitions
@@ -843,7 +844,8 @@ ensure_keepalive_timer(State) ->
 handle_live_rabbit(Node) ->
     ok = rabbit_amqqueue:on_node_up(Node),
     ok = rabbit_alarm:on_node_up(Node),
-    ok = rabbit_mnesia:on_node_up(Node).
+    ok = rabbit_mnesia:on_node_up(Node),
+    ok = rabbit_quorum_queue_periodic_membership_reconciliation:on_node_up(Node).
 
 maybe_autoheal(State = #state{partitions = []}) ->
     State;

deps/rabbit/src/rabbit_nodes.erl

@@ -282,14 +282,17 @@ do_filter_reachable(Members) ->
       Members).
 
 -spec is_running(Node) -> IsRunning when
-      Node :: node(),
+      Node :: node() | [node()],
       IsRunning :: boolean().
 %% @doc Indicates if the given node is running.
 %%
 %% @see filter_running/1.
 
 is_running(Node) when is_atom(Node) ->
-    [Node] =:= filter_running([Node]).
+    [Node] =:= filter_running([Node]);
+is_running(Nodes) when is_list(Nodes) ->
+    lists:sort(Nodes) =:= lists:sort(filter_running(Nodes)).
+
 
 -spec list_running() -> Nodes when
       Nodes :: [node()].

deps/rabbit/src/rabbit_policy.erl

@@ -367,11 +367,13 @@ validate(_VHost, <<"operator_policy">>, Name, Term, _User) ->
 notify(VHost, <<"policy">>, Name, Term0, ActingUser) ->
     Term = rabbit_data_coercion:atomize_keys(Term0),
     update_matched_objects(VHost, Term, ActingUser),
+    rabbit_quorum_queue_periodic_membership_reconciliation:policy_set(),
     rabbit_event:notify(policy_set, [{name, Name}, {vhost, VHost},
                                      {user_who_performed_action, ActingUser} | Term]);
 notify(VHost, <<"operator_policy">>, Name, Term0, ActingUser) ->
     Term = rabbit_data_coercion:atomize_keys(Term0),
     update_matched_objects(VHost, Term, ActingUser),
+    rabbit_quorum_queue_periodic_membership_reconciliation:policy_set(),
     rabbit_event:notify(policy_set, [{name, Name}, {vhost, VHost},
                                      {user_who_performed_action, ActingUser} | Term]).
 

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -8,6 +8,8 @@
 -module(rabbit_quorum_queue).
 
 -behaviour(rabbit_queue_type).
+-behaviour(rabbit_policy_validator).
+-behaviour(rabbit_policy_merge_strategy).
 
 -export([init/1,
          close/1,
@@ -35,8 +37,8 @@
 -export([format/1]).
 -export([open_files/1]).
 -export([peek/2, peek/3]).
--export([add_member/4]).
--export([delete_member/3]).
+-export([add_member/4, add_member/2]).
+-export([delete_member/3, delete_member/2]).
 -export([requeue/3]).
 -export([policy_changed/1]).
 -export([format_ra_event/3]).
@@ -65,6 +67,7 @@
          is_compatible/3,
          declare/2,
          is_stateful/0]).
+-export([validate_policy/1, merge_policy_value/3]).
 
 -export([force_shrink_member_to_current_member/2,
          force_all_queues_shrink_member_to_current_member/0]).
@@ -114,6 +117,34 @@
 -define(ADD_MEMBER_TIMEOUT, 5000).
 -define(SNAPSHOT_INTERVAL, 8192). %% the ra default is 4096
 
+%%----------- QQ policies ---------------------------------------------------
+
+-rabbit_boot_step(
+   {?MODULE,
+    [{description, "QQ target group size policies. "
+      "target-group-size controls the targeted number of "
+      "member nodes for the queue. If set, RabbitMQ will try to "
+      "grow the queue members to the target size. "
+      "See module rabbit_queue_member_eval."},
+     {mfa, {rabbit_registry, register,
+            [policy_validator, <<"target-group-size">>, ?MODULE]}},
+     {mfa, {rabbit_registry, register,
+            [operator_policy_validator, <<"target-group-size">>, ?MODULE]}},
+     {mfa, {rabbit_registry, register,
+            [policy_merge_strategy, <<"target-group-size">>, ?MODULE]}},
+     {requires, rabbit_registry},
+     {enables, recovery}]}).
+
+validate_policy(Args) ->
+    Count = proplists:get_value(<<"target-group-size">>, Args, none),
+    case is_integer(Count) andalso Count > 0 of
+        true -> ok;
+        false -> {error, "~tp is not a valid qq target count value", [Count]}
+    end.
+
+merge_policy_value(<<"target-group-size">>, Val, OpVal) ->
+    max(Val, OpVal).
+
 %%----------- rabbit_queue_type ---------------------------------------------
 
 -spec is_enabled() -> boolean().
@@ -215,6 +246,7 @@ start_cluster(Q) ->
                     ok = rabbit_fifo_client:update_machine_state(LeaderId,
                                                                  ra_machine_config(NewQ)),
                     notify_decorators(QName, startup),
+                    rabbit_quorum_queue_periodic_membership_reconciliation:queue_created(NewQ),
                     rabbit_event:notify(queue_created,
                                         [{name, QName},
                                          {durable, Durable},
@@ -1093,6 +1125,8 @@ add_member(VHost, Name, Node, Timeout) ->
                     E
     end.
 
+add_member(Q, Node) ->
+    add_member(Q, Node, ?ADD_MEMBER_TIMEOUT).
 add_member(Q, Node, Timeout) when ?amqqueue_is_quorum(Q) ->
     {RaName, _} = amqqueue:get_pid(Q),
     QName = amqqueue:get_name(Q),