Handle stream SAC monitors in SAC coordinator

The stream SAC coordinator already keeps track of its monitors but
delegates some of the work to the stream coordinator (in its monitors
map).

This commit changes this to let the SAC coordinator handle its monitors
itself. This decouples the 2 coordinators. This also fixes a bug whereby
the SAC coordinator would correctly re-issue monitors but would not
modify the monitors map accordingly. The SAC coordinator would not then
be notified of the corresponding down messages.

References rabbitmq/rabbitmq-stream-dotnet-client#447

Author: acogoluegnes

deps/rabbit/src/rabbit_stream_coordinator.erl

@@ -88,6 +88,7 @@
 -define(REPLICA_FRESHNESS_LIMIT_MS, 10 * 1000). %% 10s
 -define(V2_OR_MORE(Vsn), (Vsn >= 2)).
 -define(V5_OR_MORE(Vsn), (Vsn >= 5)).
+-define(V7_OR_MORE(Vsn), (Vsn >= 7)). %% SAC monitors no longer in monitors map
 -define(SAC_V4, rabbit_stream_sac_coordinator_v4).
 -define(SAC_CURRENT, rabbit_stream_sac_coordinator).
 
@@ -543,7 +544,7 @@ reachable_coord_members() ->
     Nodes = rabbit_nodes:list_reachable(),
     [{?MODULE, Node} || Node <- Nodes].
 
-version() -> 6.
+version() -> 7.
 
 which_module(_) ->
     ?MODULE.
@@ -588,14 +589,21 @@ apply(#{index := _Idx, machine_version := MachineVersion} = Meta0,
         {reply, Reply} ->
             return(Meta, State0, Reply, [])
     end;
-apply(Meta, {sac, SacCommand}, #?MODULE{single_active_consumer = SacState0,
-                                        monitors = Monitors0} = State0) ->
+apply(#{machine_version := Vsn} = Meta, {sac, SacCommand},
+      #?MODULE{single_active_consumer = SacState0,
+               monitors = Monitors0} = State0) ->
     Mod = sac_module(Meta),
     {SacState1, Reply, Effects0} = Mod:apply(SacCommand, SacState0),
     {SacState2, Monitors1, Effects1} =
          Mod:ensure_monitors(SacCommand, SacState1, Monitors0, Effects0),
+    Monitors2 = case ?V7_OR_MORE(Vsn) of
+                    true ->
+                        Monitors0;
+                    false ->
+                        Monitors1
+                end,
     return(Meta, State0#?MODULE{single_active_consumer = SacState2,
-                                monitors = Monitors1}, Reply, Effects1);
+                                monitors = Monitors2}, Reply, Effects1);
 apply(#{machine_version := Vsn} = Meta, {down, Pid, Reason} = Cmd,
       #?MODULE{streams = Streams0,
                monitors = Monitors0,
@@ -607,6 +615,18 @@ apply(#{machine_version := Vsn} = Meta, {down, Pid, Reason} = Cmd,
                    _ ->
                        []
                end,
+    {SacState1, SacEffects} =
+        case Vsn >= 7 of
+            true ->
+                %% all down PIDs are submitted to SAC
+                %% it filters out if not interested
+                sac_handle_connection_down(Meta, SacState0,
+                                           Pid, Reason, Vsn);
+            false ->
+                {SacState0, []}
+        end,
+    Effects1 = Effects0 ++ SacEffects,
+    State1 = State#?MODULE{single_active_consumer = SacState1},
     case maps:take(Pid, Monitors0) of
         {{StreamId, listener}, Monitors} when Vsn < 2 ->
             Listeners = case maps:take(StreamId, StateListeners0) of
@@ -620,8 +640,8 @@ apply(#{machine_version := Vsn} = Meta, {down, Pid, Reason} = Cmd,
                                         Listeners1#{StreamId => Pids}
                                 end
                         end,
-            return(Meta, State#?MODULE{listeners = Listeners,
-                                       monitors = Monitors}, ok, Effects0);
+            return(Meta, State1#?MODULE{listeners = Listeners,
+                                        monitors = Monitors}, ok, Effects1);
         {{PidStreams, listener}, Monitors} when ?V2_OR_MORE(Vsn) ->
             Streams = maps:fold(
                 fun(StreamId, _, Acc) ->
@@ -638,31 +658,34 @@ apply(#{machine_version := Vsn} = Meta, {down, Pid, Reason} = Cmd,
                             Acc
                     end
                 end, Streams0, PidStreams),
-            return(Meta, State#?MODULE{streams = Streams,
-                                       monitors = Monitors}, ok, Effects0);
+            return(Meta, State1#?MODULE{streams = Streams,
+                                        monitors = Monitors}, ok, Effects1);
         {{StreamId, member}, Monitors1} ->
             case Streams0 of
                 #{StreamId := Stream0} ->
                     Stream1 = update_stream(Meta, Cmd, Stream0),
-                    {Stream, Effects} = evaluate_stream(Meta, Stream1, Effects0),
+                    {Stream, Effects} = evaluate_stream(Meta, Stream1, Effects1),
                     Streams = Streams0#{StreamId => Stream},
-                    return(Meta, State#?MODULE{streams = Streams,
-                                               monitors = Monitors1}, ok,
+                    return(Meta, State1#?MODULE{streams = Streams,
+                                                monitors = Monitors1}, ok,
                            Effects);
                 _ ->
                     %% stream not found, can happen if "late" downs are
                     %% received
-                    return(Meta, State#?MODULE{streams = Streams0,
-                                               monitors = Monitors1}, ok, Effects0)
+                    return(Meta, State1#?MODULE{streams = Streams0,
+                                                monitors = Monitors1}, ok, Effects1)
             end;
-        {sac, Monitors1} ->
-            {SacState1, SacEffects} = sac_handle_connection_down(Meta, SacState0,
-                                                                 Pid, Reason, Vsn),
-            return(Meta, State#?MODULE{single_active_consumer = SacState1,
+        {sac, Monitors1} when Vsn < 7 ->
+            #?MODULE{single_active_consumer = SacSt0} = State1,
+            {SacSt1, SacEfts} = sac_handle_connection_down(Meta, SacSt0,
+                                                           Pid, Reason, Vsn),
+            return(Meta, State#?MODULE{single_active_consumer = SacSt1,
                                        monitors = Monitors1},
-                   ok, [Effects0 ++ SacEffects]);
+                   ok, [Effects1 ++ SacEfts]);
         error ->
-            return(Meta, State, ok, Effects0)
+            return(Meta, State1, ok, Effects1);
+        _ ->
+            return(Meta, State1, ok, Effects1)
     end;
 apply(#{machine_version := MachineVersion} = Meta,
       {register_listener, #{pid := Pid,
@@ -2304,6 +2327,9 @@ machine_version(4 = From, 5, #?MODULE{single_active_consumer = Sac0} = State) ->
     SacExport = rabbit_stream_sac_coordinator_v4:state_to_map(Sac0),
     Sac1 = rabbit_stream_sac_coordinator:import_state(From, SacExport),
     {State#?MODULE{single_active_consumer = Sac1}, []};
+machine_version(6, 7, #?MODULE{monitors = Monitors0} = State) ->
+    Monitors = maps:filter(fun(_Key, Value) -> Value =/= sac end, Monitors0),
+    {State#?MODULE{monitors = Monitors}, []};
 machine_version(From, To, State) ->
     ?LOG_INFO("Stream coordinator machine version changes from ~tp to ~tp, no state changes required.",
                     [From, To]),

deps/rabbit/src/rabbit_stream_coordinator.hrl

@@ -63,7 +63,8 @@
          monitors = #{} :: #{pid() => {stream_id() | %% v0 & v1
                                        #{stream_id() => ok}, %% v2
                                        monitor_role()} |
-                             sac},
+                                       sac %% before v7
+                            },
          %% not used as of v2
          listeners = #{} :: undefined | #{stream_id() =>
                                           #{pid() := queue_ref()}},

deps/rabbit/src/rabbit_stream_sac_coordinator.erl

@@ -955,7 +955,8 @@ list_nodes(#?MODULE{groups = Groups}) ->
 
 -spec state_enter(ra_server:ra_state(), state() | term()) ->
     ra_machine:effects().
-state_enter(leader, #?MODULE{groups = Groups} = State)
+state_enter(leader, #?MODULE{groups = Groups,
+                             pids_groups = PidsGroups} = State)
   when ?IS_STATE_REC(State) ->
     %% becoming leader, we re-issue monitors and timers for connections with
     %% disconnected consumers
@@ -978,8 +979,9 @@ state_enter(leader, #?MODULE{groups = Groups} = State)
                                   end, Acc, Cs)
               end, {#{}, #{}}, Groups),
     DisTimeout = disconnected_timeout(State),
-    %% monitor involved nodes
+    %% monitor connections and involved nodes
     %% reset a timer for disconnected connections
+    [{monitor, process, P} || P <- lists:sort(maps:keys(PidsGroups))] ++
     [{monitor, node, N} || N <- lists:sort(maps:keys(Nodes))] ++
     [begin
          Time = case ts() - Ts of

deps/rabbit/test/rabbit_stream_sac_coordinator_SUITE.erl

@@ -1870,21 +1870,21 @@ state_enter_test(_) ->
 
     assertEmpty(?MOD:state_enter(follower, #{})),
 
-    ?assertEqual(mon_node_eff([N0, N1, N2]),
+    ?assertEqual(mon_node_eff([N0, N1, N2]) ++ mon_proc_eff([P0, P1, P2]),
                  state_enter_leader(#{Id0 => grp([csr(P0), csr(P0), csr(P0)]),
                                       Id1 => grp([csr(P1), csr(P1), csr(P1)]),
                                       Id2 => grp([csr(P2), csr(P2), csr(P2)])})),
 
-    ?assertEqual(mon_node_eff([N0, N1]),
+    ?assertEqual(mon_node_eff([N0, N1]) ++ mon_proc_eff([P0, P1]),
                  state_enter_leader(#{Id0 => grp([csr(P0), csr(P0), csr(P0)]),
                                       Id1 => grp([csr(P1), csr(P1), csr(P1)]),
                                       Id2 => grp([csr(P0), csr(P1), csr(P1)])})),
 
-    ?assertEqual(lists:sort(mon_node_eff([N0, N1]) ++ [timer_eff(P1)]),
+    ?assertEqual(lists:sort(mon_node_eff([N0, N1]) ++ mon_proc_eff([P0, P1]) ++ [timer_eff(P1)]),
                  state_enter_leader(#{Id0 => grp([csr(P0), csr(P1, {disconnected, waiting})]),
                                       Id2 => grp([csr(P0)])})),
 
-    ?assertEqual(lists:sort(mon_node_eff([N0, N1, N2]) ++ timer_eff([P1, P2])),
+    ?assertEqual(lists:sort(mon_node_eff([N0, N1, N2]) ++ mon_proc_eff([P0, P1, P2]) ++ timer_eff([P1, P2])),
                  state_enter_leader(#{Id0 => grp([csr(P0), csr(P1, {disconnected, waiting})]),
                                       Id1 => grp([csr(P0), csr(P2, {disconnected, waiting})]),
                                       Id2 => grp([csr(P0), csr(P1, {disconnected, waiting})])})),
@@ -1898,6 +1898,12 @@ mon_node_eff(Nodes) when is_list(Nodes) ->
 mon_node_eff(N) ->
     {monitor, node, N}.
 
+mon_proc_eff(Pids) when is_list(Pids) ->
+    lists:sort([mon_proc_eff(P) || P <- Pids]);
+mon_proc_eff(Pid) ->
+    {monitor, process, Pid}.
+
+
 timer_eff(Pids) when is_list(Pids) ->
     lists:sort([timer_eff(Pid) || Pid <- Pids]);
 timer_eff(Pid) ->