Address code review comments for stream SAC

References #3753

5 files changed, 39 insertions, 50 deletions

diff --git a/deps/rabbit/src/rabbit_stream_coordinator.erl b/deps/rabbit/src/rabbit_stream_coordinator.erl
index 3c124c599b..9ea98e8c00 100644
--- a/deps/rabbit/src/rabbit_stream_coordinator.erl
+++ b/deps/rabbit/src/rabbit_stream_coordinator.erl
@@ -283,7 +283,7 @@ register_local_member_listener(Q) when ?is_amqqueue(Q) ->
 
 %% Single Active Consumer API
 -spec register_consumer(binary(), binary(), integer(), binary(), pid(), binary(), integer()) ->
-    {ok, boolean()}.
+    {ok, boolean()} | {error, feature_flag_disabled}.
 register_consumer(VirtualHost,
                   Stream,
                   PartitionIndex,
@@ -299,7 +299,7 @@ register_consumer(VirtualHost,
               Res
       end).
 
--spec unregister_consumer(binary(), binary(), binary(), pid(), integer()) -> ok.
+-spec unregister_consumer(binary(), binary(), binary(), pid(), integer()) -> ok | {error, feature_flag_disabled}.
 unregister_consumer(VirtualHost,
                     Stream,
                     ConsumerName,
@@ -312,7 +312,7 @@ unregister_consumer(VirtualHost,
               Res
       end).
 
--spec activate_consumer(binary(), binary(), binary()) -> ok.
+-spec activate_consumer(binary(), binary(), binary()) -> ok | {error, feature_flag_disabled}.
 activate_consumer(VirtualHost, Stream, ConsumerName) ->
     maybe_sac_execute(
       fun() ->
@@ -321,7 +321,7 @@ activate_consumer(VirtualHost, Stream, ConsumerName) ->
       end).
 
 %% return the current groups for a given virtual host
--spec consumer_groups(binary(), [atom()]) -> {ok, [term()] | {error, atom()}}.
+-spec consumer_groups(binary(), [atom()]) -> {ok, [term()] | {error, feature_flag_disabled | atom()}}.
 consumer_groups(VirtualHost, InfoKeys) ->
     maybe_sac_execute(
       fun() ->
@@ -344,7 +344,7 @@ consumer_groups(VirtualHost, InfoKeys) ->
 
 %% get the consumers of a given group in a given virtual host
 -spec group_consumers(binary(), binary(), binary(), [atom()]) ->
-    {ok, [term()]} | {error, atom()}.
+    {ok, [term()]} | {error, feature_flag_disabled | atom()}.
 group_consumers(VirtualHost, Stream, Reference, InfoKeys) ->
     maybe_sac_execute(
       fun() ->
@@ -548,7 +548,7 @@ apply(#{machine_version := MachineVersion} = Meta, {down, Pid, Reason} = Cmd,
                     return(Meta, State#?MODULE{streams = Streams0,
                                                monitors = Monitors1}, ok, Effects0)
             end;
-        {{Pid, sac}, Monitors1} ->
+        {sac, Monitors1} ->
             {SacState1, Effects} = rabbit_stream_sac_coordinator:handle_connection_down(Pid, SacState0),
             return(Meta, State#?MODULE{single_active_consumer = SacState1,
                                        monitors = Monitors1}, ok, Effects);
diff --git a/deps/rabbit/src/rabbit_stream_coordinator.hrl b/deps/rabbit/src/rabbit_stream_coordinator.hrl
index 5c29fa5aeb..65f44d6feb 100644
--- a/deps/rabbit/src/rabbit_stream_coordinator.hrl
+++ b/deps/rabbit/src/rabbit_stream_coordinator.hrl
@@ -60,7 +60,7 @@
                                     monitors = #{} :: #{pid() => {stream_id() | %% v0 & v1
                                                                   #{stream_id() => ok}, %% v2
                                                                   monitor_role()} |
-                                                                 {pid(), sac}},
+                                                        sac},
                                     %% not used as of v2
                                     listeners = #{} :: undefined | #{stream_id() =>
                                                                      #{pid() := queue_ref()}},
diff --git a/deps/rabbit/src/rabbit_stream_sac_coordinator.erl b/deps/rabbit/src/rabbit_stream_sac_coordinator.erl
index c220fa14d6..5f1e566bf9 100644
--- a/deps/rabbit/src/rabbit_stream_sac_coordinator.erl
+++ b/deps/rabbit/src/rabbit_stream_sac_coordinator.erl
@@ -165,7 +165,8 @@ consumer_groups(VirtualHost, InfoKeys, #?MODULE{groups = Groups}) ->
                                                       length(Consumers)}
                                                      | RecAcc];
                                                 (Unknown, RecAcc) ->
-                                                    [{Unknown, unknown_field} | RecAcc]
+                                                    [{Unknown, unknown_field}
+                                                     | RecAcc]
                                             end,
                                             [], InfoKeys),
                             [Record | Acc];
@@ -209,8 +210,10 @@ group_consumers(VirtualHost,
                                                     (state, RecAcc) ->
                                                         [{state, inactive}
                                                          | RecAcc];
-                                                (Unknown, RecAcc) ->
-                                                    [{Unknown, unknown_field} | RecAcc]
+                                                    (Unknown, RecAcc) ->
+                                                        [{Unknown,
+                                                          unknown_field}
+                                                         | RecAcc]
                                                 end,
                                                 [], InfoKeys),
                                 [Record | Acc]
@@ -238,8 +241,7 @@ ensure_monitors({register_consumer,
     Groups0 = maps:get(Pid, PidsGroups0, sets:new()),
     PidsGroups1 =
         maps:put(Pid, sets:add_element(GroupId, Groups0), PidsGroups0),
-    {State0#?MODULE{pids_groups = PidsGroups1},
-     Monitors0#{Pid => {Pid, sac}},
+    {State0#?MODULE{pids_groups = PidsGroups1}, Monitors0#{Pid => sac},
      [{monitor, process, Pid}, {monitor, node, node(Pid)} | Effects]};
 ensure_monitors({unregister_consumer,
                  VirtualHost,
@@ -331,7 +333,7 @@ handle_connection_down(Pid,
 
 do_register_consumer(VirtualHost,
                      Stream,
-                     -1,
+                     -1 = _PartitionIndex,
                      ConsumerName,
                      ConnectionPid,
                      Owner,
@@ -365,8 +367,8 @@ do_register_consumer(VirtualHost,
 
     #consumer{active = Active} = Consumer,
     Effects =
-        case Consumer of
-            #consumer{active = true} ->
+        case Active of
+            true ->
                 [notify_consumer_effect(ConnectionPid, SubscriptionId, Active)];
             _ ->
                 []
@@ -375,7 +377,7 @@ do_register_consumer(VirtualHost,
     {State#?MODULE{groups = StreamGroups1}, {ok, Active}, Effects};
 do_register_consumer(VirtualHost,
                      Stream,
-                     _,
+                     _PartitionIndex,
                      ConsumerName,
                      ConnectionPid,
                      Owner,
@@ -425,7 +427,7 @@ do_register_consumer(VirtualHost,
                                                          false,
                                                          true)]}
                         end;
-                    undefined ->
+                    false ->
                         %% no active consumer in the (non-empty) group, we are waiting for the reply of a former active
                         {G1, []}
                 end
@@ -541,9 +543,12 @@ has_consumers_from_pid(#group{consumers = Consumers}, Pid) ->
               end,
               Consumers).
 
-compute_active_consumer(#group{consumers = []} = Group) ->
+compute_active_consumer(#group{consumers = Crs,
+                               partition_index = -1} =
+                            Group)
+    when length(Crs) == 0 ->
     Group;
-compute_active_consumer(#group{partition_index = _,
+compute_active_consumer(#group{partition_index = -1,
                                consumers = [Consumer0]} =
                             Group0) ->
     Consumer1 = Consumer0#consumer{active = true},
@@ -553,26 +558,11 @@ compute_active_consumer(#group{partition_index = -1,
                             Group0) ->
     Consumer1 = Consumer0#consumer{active = true},
     Consumers = lists:map(fun(C) -> C#consumer{active = false} end, T),
-    Group0#group{consumers = [Consumer1] ++ Consumers};
-compute_active_consumer(#group{partition_index = PartitionIndex,
-                               consumers = Consumers0} =
-                            Group) ->
-    ActiveConsumerIndex = PartitionIndex rem length(Consumers0),
-    rabbit_log:debug("Active consumer index = ~p rem ~p = ~p",
-                     [PartitionIndex, length(Consumers0), ActiveConsumerIndex]),
-    {_, Consumers1} =
-        lists:foldr(fun (C0, {Index, Cs}) when Index == ActiveConsumerIndex ->
-                            C1 = C0#consumer{active = true},
-                            {Index - 1, [C1 | Cs]};
-                        (C0, {Index, Cs}) ->
-                            C1 = C0#consumer{active = false},
-                            {Index - 1, [C1 | Cs]}
-                    end,
-                    {length(Consumers0) - 1, []}, Consumers0),
-    Group#group{consumers = Consumers1}.
+    Group0#group{consumers = [Consumer1] ++ Consumers}.
 
 evaluate_active_consumer(#group{partition_index = PartitionIndex,
-                                consumers = Consumers}) ->
+                                consumers = Consumers})
+    when PartitionIndex >= 0 ->
     ActiveConsumerIndex = PartitionIndex rem length(Consumers),
     lists:nth(ActiveConsumerIndex + 1, Consumers).
 
@@ -606,15 +596,14 @@ update_consumer_state_in_group(#group{consumers = Consumers0} = G,
                                Pid,
                                SubId,
                                NewState) ->
-    CS1 = lists:foldr(fun(C0, Acc) ->
-                         case C0 of
-                             #consumer{pid = Pid, subscription_id = SubId} ->
-                                 C1 = C0#consumer{active = NewState},
-                                 [C1 | Acc];
-                             C -> [C | Acc]
-                         end
-                      end,
-                      [], Consumers0),
+    CS1 = lists:map(fun(C0) ->
+                       case C0 of
+                           #consumer{pid = Pid, subscription_id = SubId} ->
+                               C0#consumer{active = NewState};
+                           C -> C
+                       end
+                    end,
+                    Consumers0),
     G#group{consumers = CS1}.
 
 mod_call_effect(Pid, Msg) ->
diff --git a/deps/rabbit/test/rabbit_stream_sac_coordinator_SUITE.erl b/deps/rabbit/test/rabbit_stream_sac_coordinator_SUITE.erl
index 08a716539b..f855f1bab7 100644
--- a/deps/rabbit/test/rabbit_stream_sac_coordinator_SUITE.erl
+++ b/deps/rabbit/test/rabbit_stream_sac_coordinator_SUITE.erl
@@ -243,7 +243,7 @@ ensure_monitors_test(_) ->
                                                       []),
     assertSize(1, PidsGroups1),
     assertSize(1, maps:get(self(), PidsGroups1)),
-    ?assertEqual(#{self() => {self(), sac}}, Monitors1),
+    ?assertEqual(#{self() => sac}, Monitors1),
     ?assertEqual([{monitor, process, self()}, {monitor, node, node()}],
                  Effects1),
 
@@ -257,7 +257,7 @@ ensure_monitors_test(_) ->
                                                       []),
     assertSize(1, PidsGroups2),
     assertSize(1, maps:get(self(), PidsGroups2)),
-    ?assertEqual(#{self() => {self(), sac}}, Monitors2),
+    ?assertEqual(#{self() => sac}, Monitors2),
     ?assertEqual([{monitor, process, self()}, {monitor, node, node()}],
                  Effects2),
     Group2 = cgroup([consumer(self(), 1, true)]),
@@ -275,7 +275,7 @@ ensure_monitors_test(_) ->
                                                       []),
     assertSize(1, PidsGroups3),
     assertSize(1, maps:get(self(), PidsGroups3)),
-    ?assertEqual(#{self() => {self(), sac}}, Monitors3),
+    ?assertEqual(#{self() => sac}, Monitors3),
     ?assertEqual([], Effects3),
 
     Command3 =
diff --git a/deps/rabbitmq_stream/src/rabbit_stream_manager.erl b/deps/rabbitmq_stream/src/rabbit_stream_manager.erl
index 55c6292acc..19ca6c368a 100644
--- a/deps/rabbitmq_stream/src/rabbit_stream_manager.erl
+++ b/deps/rabbitmq_stream/src/rabbit_stream_manager.erl
@@ -425,7 +425,7 @@ handle_call({route, RoutingKey, VirtualHost, SuperStream}, _From,
           catch
               exit:Error ->
                   rabbit_log:error("Error while looking up exchange ~p, ~p",
-                                   [ExchangeName, Error]),
+                                   [rabbit_misc:rs(ExchangeName), Error]),
                   {error, stream_not_found}
           end,
     {reply, Res, State};