Replace "forgotten" by "presumed_down"

Author: acogoluegnes

deps/rabbit/src/rabbit_stream_coordinator.erl

@@ -759,7 +759,7 @@ apply(Meta, {machine_version, From, To}, State0) ->
 apply(Meta, {timeout, {sac, node_disconnected, #{connection_pid := Pid}}},
       #?MODULE{single_active_consumer = SacState0} = State0) ->
     Mod = sac_module(Meta),
-    {SacState1, Effects} = Mod:forget_connection(Pid, SacState0),
+    {SacState1, Effects} = Mod:presume_connection_down(Pid, SacState0),
     return(Meta, State0#?MODULE{single_active_consumer = SacState1}, ok,
            Effects);
 apply(Meta, UnkCmd, State) ->
@@ -2437,11 +2437,8 @@ 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) ->

deps/rabbit/src/rabbit_stream_sac_coordinator.erl

@@ -43,7 +43,7 @@
          ensure_monitors/4,
          handle_connection_down/3,
          handle_node_reconnected/3,
-         forget_connection/2,
+         presume_connection_down/2,
          consumer_groups/3,
          group_consumers/5,
          overview/1,
@@ -66,12 +66,12 @@
 
 -define(CONNECTED, connected).
 -define(DISCONNECTED, disconnected).
--define(FORGOTTTEN, forgotten).
+-define(PDOWN, presumed_down).
 
 -define(CONN_ACT, {?CONNECTED, ?ACTIVE}).
 -define(CONN_WAIT, {?CONNECTED, ?WAITING}).
 -define(DISCONN_ACT, {?DISCONNECTED, ?ACTIVE}).
--define(FORG_ACT, {?FORGOTTTEN, ?ACTIVE}).
+-define(PDOWN_ACT, {?PDOWN, ?ACTIVE}).
 
 -define(DISCONNECTED_TIMEOUT_APP_KEY, stream_sac_disconnected_timeout).
 -define(DISCONNECTED_TIMEOUT_CONF_KEY, disconnected_timeout).
@@ -389,7 +389,7 @@ handle_forgotten_active_reconnected(Pid,
             lists:foldr(fun(#consumer{status = St,
                                       pid = P,
                                       subscription_id = SID} = C, {Cs, Eff})
-                              when P =:= Pid andalso St =:= ?FORG_ACT ->
+                              when P =:= Pid andalso St =:= ?PDOWN_ACT ->
                                 {[csr_status(C, ?CONN_WAIT) | Cs],
                                  [notify_consumer_effect(Pid, SID, S,
                                                          Name, false, true) | Eff]};
@@ -400,13 +400,13 @@ handle_forgotten_active_reconnected(Pid,
             lists:foldr(fun(#consumer{status = St,
                                       pid = P,
                                       subscription_id = SID} = C, {Cs, Eff})
-                              when P =:= Pid andalso St =:= ?FORG_ACT ->
+                              when P =:= Pid andalso St =:= ?PDOWN_ACT ->
                                 %% update forgotten active
                                 %% tell it to step down
                                 {[csr_status(C, ?CONN_WAIT) | Cs],
                                  [notify_consumer_effect(P, SID, S,
                                                          Name, false, true) | Eff]};
-                           (#consumer{status = {?FORGOTTTEN, _},
+                           (#consumer{status = {?PDOWN, _},
                                       pid = P} = C, {Cs, Eff})
                               when P =:= Pid ->
                                 %% update forgotten
@@ -428,7 +428,7 @@ handle_forgotten_active_reconnected(Pid,
     {S0#?MODULE{groups = Groups1}, Eff1}.
 
 has_forgotten_active(#group{consumers = Consumers}, Pid) ->
-    case lists:search(fun(#consumer{status = ?FORG_ACT,
+    case lists:search(fun(#consumer{status = ?PDOWN_ACT,
                                     pid = P}) when P =:= Pid ->
                               true;
                          (_) -> false
@@ -453,8 +453,6 @@ has_consumer_with_status(#group{consumers = Consumers}, Status) ->
             true
     end.
 
-
-
 maybe_rebalance_group(#group{partition_index = -1, consumers = Consumers0} = G0,
                       {_VH, S, Name}) ->
     case lookup_active_consumer(G0) of
@@ -611,7 +609,7 @@ group_consumers(VirtualHost,
             {error, not_found}
     end.
 
-cli_consumer_status_label({?FORGOTTTEN, _}) ->
+cli_consumer_status_label({?PDOWN, _}) ->
    inactive;
 cli_consumer_status_label({_, ?ACTIVE}) ->
    active;
@@ -757,17 +755,17 @@ handle_node_reconnected(Node,
 
     {State0#?MODULE{pids_groups = PidsGroups1}, Effects1}.
 
--spec forget_connection(connection_pid(), state()) ->
+-spec presume_connection_down(connection_pid(), state()) ->
     {state(), ra_machine:effects()}.
-forget_connection(Pid, #?MODULE{groups = Groups} = State0) ->
+presume_connection_down(Pid, #?MODULE{groups = Groups} = State0) ->
     {State1, Eff} =
-        maps:fold(fun(G, _, {St, Eff}) ->
-                          handle_group_forget_connection(Pid, St, Eff, G)
-                  end, {State0, []}, Groups),
+    maps:fold(fun(G, _, {St, Eff}) ->
+                      handle_group_connection_presumed_down(Pid, St, Eff, G)
+              end, {State0, []}, Groups),
     {State1, Eff}.
 
-handle_group_forget_connection(Pid, #?MODULE{groups = Groups0} = S0,
-                               Eff0, {VH, S, Name} = K) ->
+handle_group_connection_presumed_down(Pid, #?MODULE{groups = Groups0} = S0,
+                                      Eff0, {VH, S, Name} = K) ->
     case lookup_group(VH, S, Name, Groups0) of
         undefined ->
             {S0, Eff0};
@@ -776,7 +774,7 @@ handle_group_forget_connection(Pid, #?MODULE{groups = Groups0} = S0,
                 lists:foldr(
                   fun(#consumer{pid = P, status = {?DISCONNECTED, St}} = C, {L, _})
                         when P == Pid ->
-                          {[csr_status(C, {?FORGOTTTEN, St}) | L], true};
+                          {[csr_status(C, {?PDOWN, St}) | L], true};
                      (C, {L, UpdatedFlag}) ->
                           {[C | L], UpdatedFlag or false}
                   end, {[], false}, Consumers0),
@@ -939,7 +937,7 @@ active_to_status(true) ->
 active_to_status(false) ->
     {?CONNECTED, ?WAITING}.
 
-is_active({?FORGOTTTEN, _}) ->
+is_active({?PDOWN, _}) ->
     false;
 is_active({_, ?ACTIVE}) ->
     true;
@@ -1128,7 +1126,7 @@ has_consumers_from_pid(#group{consumers = Consumers}, Pid) ->
 
 compute_active_consumer(#group{partition_index = -1,
                                consumers = Crs} = Group)
-    when length(Crs) == 0 ->
+  when length(Crs) == 0 ->
     Group;
 compute_active_consumer(#group{partition_index = -1,
                                consumers = Consumers} = G) ->
@@ -1140,7 +1138,7 @@ compute_active_consumer(#group{partition_index = -1,
         false ->
             case evaluate_active_consumer(G) of
                 undefined ->
-                    ok;
+                    G;
                 #consumer{pid = Pid, subscription_id = SubId} ->
                     Consumers1 =
                         lists:foldr(
@@ -1173,17 +1171,17 @@ evaluate_active_consumer(#group{consumers = Consumers} = G) ->
     end.
 
 do_evaluate_active_consumer(#group{consumers = Consumers})
-    when length(Consumers) == 0 ->
+  when length(Consumers) == 0 ->
     undefined;
 do_evaluate_active_consumer(#group{partition_index = -1,
-                                consumers = [Consumer]}) ->
+                                   consumers = [Consumer]}) ->
     Consumer;
 do_evaluate_active_consumer(#group{partition_index = -1,
-                                consumers = [Consumer | _]}) ->
+                                   consumers = [Consumer | _]}) ->
     Consumer;
 do_evaluate_active_consumer(#group{partition_index = PartitionIndex,
                                    consumers = Consumers})
-    when PartitionIndex >= 0 ->
+  when PartitionIndex >= 0 ->
     ActiveConsumerIndex = PartitionIndex rem length(Consumers),
     lists:nth(ActiveConsumerIndex + 1, Consumers).
 
@@ -1284,8 +1282,7 @@ compute_node_pid_group_dependencies(Node, Groups) ->
                                   end, Acc, Consumers)
               end, #{}, Groups).
 
--spec csr(pid(), subscription_id(), owner(),
-          {consumer_connectivity(), consumer_status()}) ->
+-spec csr(pid(), subscription_id(), owner(), consumer_status()) ->
     consumer().
 csr(Pid, Id, Owner, Status) ->
     #consumer{pid = Pid,
@@ -1294,8 +1291,7 @@ csr(Pid, Id, Owner, Status) ->
               status = Status,
               ts = ts()}.
 
--spec csr_status(consumer(), {consumer_connectivity(), consumer_status()}) ->
-    consumer().
+-spec csr_status(consumer(), consumer_status()) -> consumer().
 csr_status(C, Status) ->
     C#consumer{status = Status, ts = ts()}.
 

deps/rabbit/src/rabbit_stream_sac_coordinator.hrl

@@ -22,16 +22,17 @@
 -type subscription_id() :: byte().
 -type group_id() :: {vhost(), stream(), consumer_name()}.
 -type owner() :: binary().
--type consumer_status() :: active | waiting | deactivating.
--type consumer_connectivity() :: connected | disconnected | forgotten.
+-type consumer_activity() :: active | waiting | deactivating.
+-type consumer_connectivity() :: connected | disconnected | presumed_down.
+-type consumer_status() :: {consumer_connectivity(), consumer_activity()}.
 -type conf() :: map().
 -type timestamp() :: integer().
 
 -record(consumer,
         {pid :: pid(),
          subscription_id :: subscription_id(),
          owner :: owner(), %% just a label
-         status :: {consumer_connectivity(), consumer_status()},
+         status :: consumer_status(),
          ts :: timestamp()}).
 -record(group,
         {consumers :: [#consumer{}], partition_index :: integer()}).