QQ Reconciliator - move hardcoded triggers to events subscription

Author: ikavgo

deps/rabbit/src/rabbit_node_monitor.erl

@@ -857,7 +857,6 @@ handle_dead_rabbit(Node, State) ->
     %% statements on *one* node, rather than all of them.
     ok = rabbit_amqqueue:on_node_down(Node),
     ok = rabbit_alarm:on_node_down(Node),
-    ok = rabbit_quorum_queue_periodic_membership_reconciliation:on_node_down(Node),
     State1 = case rabbit_khepri:is_enabled() of
                  true  -> State;
                  false -> on_node_down_using_mnesia(Node, State)
@@ -898,8 +897,7 @@ handle_live_rabbit(Node) ->
         true  -> ok;
         false -> on_node_up_using_mnesia(Node)
     end,
-    ok = rabbit_vhosts:on_node_up(Node),
-    ok = rabbit_quorum_queue_periodic_membership_reconciliation:on_node_up(Node).
+    ok = rabbit_vhosts:on_node_up(Node).
 
 on_node_up_using_mnesia(Node) ->
     ok = rabbit_mnesia:on_node_up(Node).

deps/rabbit/src/rabbit_policy.erl

@@ -378,13 +378,11 @@ 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_event_subscriber.erl

@@ -0,0 +1,51 @@
+%% 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-2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+%%
+
+-module(rabbit_quorum_event_subscriber).
+
+-behaviour(gen_event).
+
+-export([init/1, handle_event/2, handle_call/2]).
+-export([register/0, unregister/0]).
+
+-include_lib("rabbit_common/include/rabbit.hrl").
+
+-rabbit_boot_step({rabbit_quorum_event_subscriber,
+                   [{description, "quorum queue event subscriber"},
+                    {mfa,         {?MODULE, register, []}},
+                    {cleanup,     {?MODULE, unregister, []}},
+                    {requires,    rabbit_event},
+                    {enables,     recovery}]}).
+
+register() ->
+    gen_event:add_handler(rabbit_alarm, ?MODULE, []),
+    gen_event:add_handler(rabbit_event, ?MODULE, []).
+
+unregister() ->
+    gen_event:delete_handler(rabbit_alarm, ?MODULE, []),
+    gen_event:delete_handler(rabbit_event, ?MODULE, []).
+
+init([]) ->
+    {ok, []}.
+
+handle_call( _, State) ->
+    {ok, ok, State}.
+
+handle_event({node_up, Node}, State) ->
+    rabbit_quorum_queue_periodic_membership_reconciliation:on_node_up(Node),
+    {ok, State};
+handle_event({node_down, Node}, State) ->
+    rabbit_quorum_queue_periodic_membership_reconciliation:on_node_down(Node),
+    {ok, State};
+handle_event(#event{type = policy_set}, State) ->
+    rabbit_quorum_queue_periodic_membership_reconciliation:policy_set(),
+    {ok, State};
+handle_event(#event{type = operator_policy_set}, State) ->
+    rabbit_quorum_queue_periodic_membership_reconciliation:policy_set(),
+    {ok, State};
+handle_event(_, State) ->
+    {ok, State}.