Update stream SAC configuration in tick

Not in aux.

Author: acogoluegnes

deps/rabbit/src/rabbit_stream_coordinator.erl

@@ -810,9 +810,9 @@ all_member_nodes(Streams) ->
                 maps:merge(Acc, M)
         end, #{}, Streams)).
 
-tick(_Ts, _State) ->
-    [{aux, maybe_resize_coordinator_cluster},
-     {aux, maybe_update_sac_configuration}].
+tick(_Ts, #?MODULE{single_active_consumer = SacState}) ->
+    [{aux, maybe_resize_coordinator_cluster} |
+     maybe_update_sac_configuration(SacState)].
 
 members() ->
     %% TODO: this can be replaced with a ra_leaderboard
@@ -897,18 +897,13 @@ maybe_handle_stale_nodes(MemberNodes, ExpectedNodes,
 maybe_handle_stale_nodes(_, _, _, _) ->
     ok.
 
-maybe_update_sac_configuration(RaAux, MacVersion) when MacVersion >= 5 ->
-    #?MODULE{single_active_consumer = SacState} = ra_aux:machine_state(RaAux),
+maybe_update_sac_configuration(SacState) ->
     case sac_check_conf_change(SacState) of
         {new, UpdatedConf} ->
-            Leader = ra_aux:leader_id(RaAux),
-            ra:pipeline_command(Leader, sac_make_update_conf(UpdatedConf)),
-            ok;
+            [{append, sac_make_update_conf(UpdatedConf), noreply}];
         _ ->
-            ok
-    end;
-maybe_update_sac_configuration(_, _) ->
-    ok.
+            []
+    end.
 
 add_member(Members, Node) ->
     MinMacVersion = erpc:call(Node, ?MODULE, version, []),
@@ -992,11 +987,6 @@ handle_aux(leader, _, maybe_resize_coordinator_cluster,
            AuxState, RaAux) ->
     %% Coordinator resizing is still happening, let's ignore this tick event
     {no_reply, AuxState, RaAux};
-handle_aux(leader, _, maybe_update_sac_configuration,
-           AuxState, RaAux) ->
-    MachineVersion = ra_aux:effective_machine_version(RaAux),
-    maybe_update_sac_configuration(RaAux, MachineVersion),
-    {no_reply, AuxState, RaAux};
 handle_aux(leader, _, {down, Pid, _},
            #aux{resizer = Pid} = Aux, RaAux) ->
     %% Coordinator resizing has finished

deps/rabbit/src/rabbit_stream_sac_coordinator.erl

@@ -861,8 +861,9 @@ import_state(4, #{<<"groups">> := Groups, <<"pids_groups">> := PidsGroups}) ->
              pids_groups = map_to_pids_groups(PidsGroups),
              conf = #{disconnected_timeout => ?DISCONNECTED_TIMEOUT_MS}}.
 
--spec check_conf_change(state()) -> {new, conf()} | unchanged.
-check_conf_change(#?MODULE{conf = Conf}) ->
+-spec check_conf_change(state() | term()) -> {new, conf()} | unchanged.
+check_conf_change(State) when is_record(State, ?MODULE)->
+    #?MODULE{conf = Conf} = State,
     DisconTimeout = lookup_disconnected_timeout(),
     case Conf of
         #{?DISCONNECTED_TIMEOUT_CONF_KEY := DT}
@@ -872,7 +873,9 @@ check_conf_change(#?MODULE{conf = Conf}) ->
             {new, #{?DISCONNECTED_TIMEOUT_CONF_KEY => DisconTimeout}};
         _ ->
             unchanged
-    end.
+    end;
+check_conf_change(_) ->
+    unchanged.
 
 - spec make_purge_nodes([node()]) -> {sac, command()}.
 make_purge_nodes(Nodes) ->

deps/rabbit/test/rabbit_stream_sac_coordinator_SUITE.erl

@@ -70,6 +70,8 @@ check_conf_test(_) ->
                  ?MOD:check_conf_change(state_with_conf(#{K => 42}))),
     ?assertMatch(unchanged,
                  ?MOD:check_conf_change(state_with_conf(#{K => Def}))),
+    ?assertMatch(unchanged,
+                 ?MOD:check_conf_change(#{K => Def})),
     ok.
 
 simple_sac_test(_) ->