Handle disconnection in one function with reason parameter

Author: acogoluegnes

deps/rabbit/src/rabbit_stream_coordinator.erl

@@ -87,6 +87,8 @@
 -define(REPLICA_FRESHNESS_LIMIT_MS, 10 * 1000). %% 10s
 -define(V2_OR_MORE(Vsn), Vsn >= 2).
 -define(V5_OR_MORE(Vsn), Vsn >= 5).
+-define(SAC_V4, rabbit_stream_sac_coordinator_v4).
+-define(SAC_CURRENT, rabbit_stream_sac_coordinator).
 
 -type state() :: #?MODULE{}.
 -type args() :: #{index := ra:index(),
@@ -649,19 +651,12 @@ apply(#{machine_version := Vsn} = Meta, {down, Pid, Reason} = Cmd,
                     return(Meta, State#?MODULE{streams = Streams0,
                                                monitors = Monitors1}, ok, Effects0)
             end;
-        {sac, Monitors1} when Vsn < 5 orelse Reason =/= noconnection ->
-            %% A connection went down, v5+ treats noconnection differently but
-            %% v4- does not.
-            Mod = sac_module(Meta),
-            {SacState1, Effects} = Mod:handle_connection_down(Pid, SacState0),
+        {sac, Monitors1} ->
+            {SacState1, SacEffects} = sac_handle_connection_down(SacState0, Pid,
+                                                                 Reason, Vsn),
             return(Meta, State#?MODULE{single_active_consumer = SacState1,
-                                       monitors = Monitors1}, ok, [Effects0 ++ Effects]);
-        {sac, Monitors1} when Reason =:= noconnection ->
-            %% the node of a connection got disconnected
-            Mod = sac_module(Meta),
-            {SacState1, Effects} = Mod:handle_connection_node_disconnected(Pid, SacState0),
-            return(Meta, State#?MODULE{single_active_consumer = SacState1,
-                                       monitors = Monitors1}, ok, [Effects0 ++ Effects]);
+                                       monitors = Monitors1},
+                   ok, [Effects0 ++ SacEffects]);
         error ->
             return(Meta, State, ok, Effects0)
     end;
@@ -796,9 +791,9 @@ state_enter(_S, _) ->
     [].
 
 sac_module(#{machine_version := Vsn}) when ?V5_OR_MORE(Vsn) ->
-    rabbit_stream_sac_coordinator;
+    ?SAC_CURRENT;
 sac_module(_) ->
-    rabbit_stream_sac_coordinator_v4.
+    ?SAC_V4.
 
 all_member_nodes(Streams) ->
     maps:keys(
@@ -2441,6 +2436,14 @@ maps_to_list(M) ->
 ra_local_query(QueryFun) ->
     ra:local_query({?MODULE, node()}, QueryFun, infinity).
 
+sac_handle_connection_down(SacState, Pid, Reason, Vsn) when ?V5_OR_MORE(Vsn) ->
+    %% the node of a connection got disconnected
+    ?SAC_CURRENT:handle_connection_down(Pid, Reason, SacState);
+sac_handle_connection_down(SacState, Pid, _Reason, _Vsn) ->
+    %% A connection went down, v5+ treats noconnection differently but
+    %% v4- does not.
+    ?SAC_V4:handle_connection_down(Pid, SacState).
+
 sac_make_purge_nodes(Nodes) ->
     rabbit_stream_sac_coordinator:make_purge_nodes(Nodes).
 

deps/rabbit/src/rabbit_stream_sac_coordinator.erl

@@ -41,8 +41,7 @@
          init_state/0,
          send_message/2,
          ensure_monitors/4,
-         handle_connection_down/2,
-         handle_connection_node_disconnected/2,
+         handle_connection_down/3,
          handle_node_reconnected/3,
          forget_connection/2,
          consumer_groups/3,
@@ -704,9 +703,11 @@ ensure_monitors(#command_purge_nodes{},
 ensure_monitors(_, #?MODULE{} = State0, Monitors, Effects) ->
     {State0, Monitors, Effects}.
 
--spec handle_connection_down(connection_pid(), state()) ->
-                                {state(), ra_machine:effects()}.
-handle_connection_down(Pid,
+-spec handle_connection_down(connection_pid(), term(), state()) ->
+    {state(), ra_machine:effects()}.
+handle_connection_down(Pid, noconnection, State) ->
+    handle_connection_node_disconnected(Pid, State);
+handle_connection_down(Pid, _Reason,
                        #?MODULE{pids_groups = PidsGroups0} = State0) ->
     case maps:take(Pid, PidsGroups0) of
         error ->

deps/rabbit/src/rabbit_stream_sac_coordinator_v4.erl

@@ -409,7 +409,7 @@ ensure_monitors(_, #?STATE{} = State0, Monitors, Effects) ->
     {State0, Monitors, Effects}.
 
 -spec handle_connection_down(connection_pid(), state()) ->
-                                {state(), ra_machine:effects()}.
+    {state(), ra_machine:effects()}.
 handle_connection_down(Pid,
                        #?STATE{pids_groups = PidsGroups0} = State0) ->
     case maps:take(Pid, PidsGroups0) of

deps/rabbit/test/rabbit_stream_sac_coordinator_SUITE.erl

@@ -316,13 +316,13 @@ handle_connection_down_sac_should_get_activated_test(_) ->
     State0 = state(#{GroupId => Group}),
 
     {#?STATE{pids_groups = PidsGroups1, groups = Groups1} = State1,
-     Effects1} = ?MOD:handle_connection_down(Pid0, State0),
+     Effects1} = ?MOD:handle_connection_down(Pid0, normal, State0),
     assertSize(1, PidsGroups1),
     assertSize(1, maps:get(Pid1, PidsGroups1)),
     assertSendMessageActivateEffect(Pid1, 1, Stream, ConsumerName, true, Effects1),
     assertHasGroup(GroupId, grp([csr(Pid1, 1, active)]), Groups1),
     {#?STATE{pids_groups = PidsGroups2, groups = Groups2},
-     Effects2} = ?MOD:handle_connection_down(Pid1, State1),
+     Effects2} = ?MOD:handle_connection_down(Pid1, normal, State1),
     assertEmpty(PidsGroups2),
     assertEmpty(Effects2),
     assertEmpty(Groups2),
@@ -341,7 +341,7 @@ handle_connection_down_sac_active_does_not_change_test(_) ->
     State = state(#{GroupId => Group}),
 
     {#?STATE{pids_groups = PidsGroups, groups = Groups},
-     Effects} = ?MOD:handle_connection_down(Pid0, State),
+     Effects} = ?MOD:handle_connection_down(Pid0, normal, State),
     assertSize(1, PidsGroups),
     assertSize(1, maps:get(Pid1, PidsGroups)),
     assertEmpty(Effects),
@@ -358,7 +358,7 @@ handle_connection_down_sac_no_more_consumers_test(_) ->
     State = state(#{GroupId => Group}),
 
     {#?STATE{pids_groups = PidsGroups, groups = Groups},
-     Effects} = ?MOD:handle_connection_down(Pid0, State),
+     Effects} = ?MOD:handle_connection_down(Pid0, normal, State),
     assertEmpty(PidsGroups),
     assertEmpty(Groups),
     assertEmpty(Effects),
@@ -377,7 +377,7 @@ handle_connection_down_sac_no_consumers_in_down_connection_test(_) ->
                     Pid1 => maps:from_list([{GroupId, true}])}),
 
     {#?STATE{pids_groups = PidsGroups, groups = Groups},
-     Effects} = ?MOD:handle_connection_down(Pid0, State),
+     Effects} = ?MOD:handle_connection_down(Pid0, normal, State),
 
     assertSize(1, PidsGroups),
     assertSize(1, maps:get(Pid1, PidsGroups)),
@@ -400,7 +400,7 @@ handle_connection_down_super_stream_active_stays_test(_) ->
     State = state(#{GroupId => Group}),
 
     {#?STATE{pids_groups = PidsGroups, groups = Groups},
-     Effects} = ?MOD:handle_connection_down(Pid1, State),
+     Effects} = ?MOD:handle_connection_down(Pid1, normal, State),
     assertSize(1, PidsGroups),
     assertSize(1, maps:get(Pid0, PidsGroups)),
     assertEmpty(Effects),
@@ -424,7 +424,7 @@ handle_connection_down_super_stream_active_changes_test(_) ->
 
     {#?STATE{pids_groups = PidsGroups, groups = Groups},
      Effects} =
-    ?MOD:handle_connection_down(Pid0, State),
+    ?MOD:handle_connection_down(Pid0, normal, State),
     assertSize(1, PidsGroups),
     assertSize(1, maps:get(Pid1, PidsGroups)),
     assertSendMessageSteppingDownEffect(Pid1, 1, Stream, ConsumerName, Effects),
@@ -447,7 +447,7 @@ handle_connection_down_super_stream_activate_in_remaining_connection_test(_) ->
     State = state(#{GroupId => Group}),
 
     {#?STATE{pids_groups = PidsGroups, groups = Groups},
-     Effects} = ?MOD:handle_connection_down(Pid0, State),
+     Effects} = ?MOD:handle_connection_down(Pid0, normal, State),
     assertSize(1, PidsGroups),
     assertSize(1, maps:get(Pid1, PidsGroups)),
     assertSendMessageActivateEffect(Pid1, 3, Stream, ConsumerName, true, Effects),
@@ -472,7 +472,7 @@ handle_connection_down_super_stream_no_active_removed_or_present_test(_) ->
     State = state(#{GroupId => Group}),
 
     {#?STATE{pids_groups = PidsGroups, groups = Groups},
-     Effects} = ?MOD:handle_connection_down(Pid0, State),
+     Effects} = ?MOD:handle_connection_down(Pid0, normal, State),
     assertSize(1, PidsGroups),
     assertSize(1, maps:get(Pid1, PidsGroups)),
     assertEmpty(Effects),
@@ -509,7 +509,7 @@ handle_connection_down_consumers_from_dead_connection_should_be_filtered_out_tes
 
     {#?STATE{pids_groups = PidsGroups1, groups = Groups1} = State1,
      Effects1} =
-    ?MOD:handle_connection_down(Pid0, State0),
+    ?MOD:handle_connection_down(Pid0, normal, State0),
     assertSize(2, PidsGroups1),
     assertSize(1, maps:get(Pid1, PidsGroups1)),
     assertSize(1, maps:get(Pid2, PidsGroups1)),
@@ -520,7 +520,7 @@ handle_connection_down_consumers_from_dead_connection_should_be_filtered_out_tes
                    Groups1),
 
     {#?STATE{pids_groups = PidsGroups2, groups = Groups2},
-     Effects2} = ?MOD:handle_connection_down(Pid1, State1),
+     Effects2} = ?MOD:handle_connection_down(Pid1, normal, State1),
     assertSize(1, PidsGroups2),
     assertSize(1, maps:get(Pid2, PidsGroups2)),
     assertSendMessageActivateEffect(Pid2, 2, Stream, ConsumerName, true, Effects2),
@@ -599,7 +599,7 @@ handle_connection_node_disconnected_test(_) ->
 
     {#?STATE{pids_groups = PidsGroups1, groups = Groups1} = _State1,
      [Effect1]} =
-    ?MOD:handle_connection_node_disconnected(Pid1, State0),
+    ?MOD:handle_connection_down(Pid1, noconnection, State0),
     assertSize(2, PidsGroups1),
     assertSize(1, maps:get(Pid0, PidsGroups1)),
     assertSize(1, maps:get(Pid2, PidsGroups1)),
@@ -945,7 +945,8 @@ handle_connection_down_simple_disconn_active_block_rebalancing_test(_) ->
 
     Groups0 = #{GId => Group},
     State0 = state(Groups0),
-    {#?STATE{groups = Groups1}, Eff} = ?MOD:handle_connection_down(Pid2, State0),
+    {#?STATE{groups = Groups1}, Eff} = ?MOD:handle_connection_down(Pid2, normal,
+                                                                   State0),
     assertHasGroup(GId, grp([csr(Pid0, 0, {connected, waiting}),
                                 csr(Pid1, 0, {disconnected, active})]),
                    Groups1),
@@ -963,7 +964,8 @@ handle_connection_down_super_stream_disconn_active_block_rebalancing_test(_) ->
 
     Groups0 = #{GId => Group},
     State0 = state(Groups0),
-    {#?STATE{groups = Groups1}, Eff} = ?MOD:handle_connection_down(Pid0, State0),
+    {#?STATE{groups = Groups1}, Eff} = ?MOD:handle_connection_down(Pid0, normal,
+                                                                   State0),
     assertHasGroup(GId, grp(1, [csr(Pid1, 0, {disconnected, active}),
                                    csr(Pid2, 0, {connected, waiting})]),
                    Groups1),
@@ -982,7 +984,7 @@ handle_connection_node_disconnected_simple_disconn_active_block_rebalancing_test
     Groups0 = #{GId => Group},
     State0 = state(Groups0),
     {#?STATE{groups = Groups1}, Eff} =
-    ?MOD:handle_connection_node_disconnected(Pid2, State0),
+    ?MOD:handle_connection_down(Pid2, noconnection, State0),
     assertHasGroup(GId, grp([csr(Pid0, 0, {connected, waiting}),
                                 csr(Pid1, 0, {disconnected, active}),
                                 csr(Pid2, 0, {disconnected, waiting})]),
@@ -1002,7 +1004,7 @@ handle_connection_node_disconnected_super_stream_disconn_active_block_rebalancin
     Groups0 = #{GId => Group},
     State0 = state(Groups0),
     {#?STATE{groups = Groups1}, Eff} =
-    ?MOD:handle_connection_node_disconnected(Pid0, State0),
+    ?MOD:handle_connection_down(Pid0, noconnection, State0),
     assertHasGroup(GId, grp(1, [csr(Pid0, 0, {disconnected, waiting}),
                                    csr(Pid1, 0, {disconnected, active}),
                                    csr(Pid2, 0, {connected, waiting})]),
@@ -1317,9 +1319,9 @@ node_disconnected_and_reconnected_test(_) ->
 
     State0 = state(#{GId0 => G0, GId1 => G1, GId2 => G2}),
 
-    {State1, Eff1} = ?MOD:handle_connection_node_disconnected(N1P0, State0),
-    {State2, Eff2} = ?MOD:handle_connection_node_disconnected(N1P1, State1),
-    {State3, Eff3} = ?MOD:handle_connection_node_disconnected(N1P2, State2),
+    {State1, Eff1} = ?MOD:handle_connection_down(N1P0, noconnection, State0),
+    {State2, Eff2} = ?MOD:handle_connection_down(N1P1, noconnection, State1),
+    {State3, Eff3} = ?MOD:handle_connection_down(N1P2, noconnection, State2),
 
     assertNodeDisconnectedTimerEffect(N1P0, Eff1),
     assertNodeDisconnectedTimerEffect(N1P1, Eff2),
@@ -1449,7 +1451,7 @@ node_disconnected_reconnected_connection_down_test(_) ->
     S0 = state(#{GId => G0}),
 
     {#?STATE{groups = G1} = S1, Eff1} =
-        ?MOD:handle_connection_node_disconnected(P1, S0),
+        ?MOD:handle_connection_down(P1, noconnection, S0),
 
     assertHasGroup(GId,
                    grp(1, [csr(P0, {connected, waiting}),
@@ -1470,7 +1472,7 @@ node_disconnected_reconnected_connection_down_test(_) ->
 
     assertContainsCheckConnectionEffect(P1, Eff2),
 
-    {#?STATE{groups = G3}, Eff3} = ?MOD:handle_connection_down(P1, S2),
+    {#?STATE{groups = G3}, Eff3} = ?MOD:handle_connection_down(P1, normal, S2),
 
     assertHasGroup(GId,
                    grp(1, [csr(P0, {connected, waiting}),