Single Active Consumer for streams

deps/rabbit/BUILD.bazel

@@ -794,6 +794,12 @@ suites = [
             "//deps/rabbit_common:erlang_app",
         ],
     ),
+    rabbitmq_suite(
+        name = "rabbit_stream_sac_coordinator_SUITE",
+        deps = [
+            "//deps/rabbit_common:erlang_app",
+        ],
+    ),
     rabbitmq_integration_suite(
         PACKAGE,
         name = "rabbit_stream_queue_SUITE",

deps/rabbit/rebar.config

@@ -0,0 +1,13 @@
+{plugins, [rebar3_format]}.
+
+{format, [
+    {files, ["src/rabbit_stream_sac_coordinator.*",
+             "test/rabbit_stream_sac_coordinator_SUITE.erl"]},
+    {formatter, default_formatter},
+    {options, #{
+                paper => 80,
+                ribbon => 70,
+                inline_attributes => {when_under, 1},
+                inline_items => {when_under, 4}
+               }}
+]}.

deps/rabbit/src/rabbit_core_ff.erl

@@ -13,7 +13,8 @@
          implicit_default_bindings_migration/3,
          virtual_host_metadata_migration/3,
          maintenance_mode_status_migration/3,
-         user_limits_migration/3]).
+         user_limits_migration/3,
+         stream_single_active_consumer_migration/3]).
 
 -rabbit_feature_flag(
    {classic_mirrored_queue_version,
@@ -68,6 +69,15 @@
        migration_fun => {?MODULE, user_limits_migration}
      }}).
 
+-rabbit_feature_flag(
+   {stream_single_active_consumer,
+    #{desc          => "Single active consumer for streams",
+      doc_url       => "https://www.rabbitmq.com/stream.html",
+      stability     => stable,
+      depends_on    => [stream_queue],
+      migration_fun => {?MODULE, stream_single_active_consumer_migration}
+     }}).
+
 classic_mirrored_queue_version_migration(_FeatureName, _FeatureProps, _Enable) ->
     ok.
 
@@ -188,3 +198,12 @@ user_limits_migration(_FeatureName, _FeatureProps, enable) ->
     end;
 user_limits_migration(_FeatureName, _FeatureProps, is_enabled) ->
     mnesia:table_info(rabbit_user, attributes) =:= internal_user:fields(internal_user_v2).
+
+%% -------------------------------------------------------------------
+%% Stream single active consumer.
+%% -------------------------------------------------------------------
+
+stream_single_active_consumer_migration(_FeatureName, _FeatureProps, enable) ->
+    ok;
+stream_single_active_consumer_migration(_FeatureName, _FeatureProps, is_enabled) ->
+    undefined.

deps/rabbit/src/rabbit_stream_coordinator.erl

@@ -40,6 +40,10 @@
 -export([log_overview/1]).
 -export([replay/1]).
 
+%% for SAC coordinator
+-export([process_command/1,
+         sac_state/1]).
+
 %% for testing and debugging
 -export([eval_listeners/3,
          state/0]).
@@ -88,6 +92,7 @@
                    {member_stopped, stream_id(), args()} |
                    {retention_updated, stream_id(), args()} |
                    {mnesia_updated, stream_id(), args()} |
+                   {sac, rabbit_stream_sac_coordinator:command()} |
                    ra_machine:effect().
 
 -export_type([command/0]).
@@ -171,6 +176,9 @@ policy_changed(Q) when ?is_amqqueue(Q) ->
     StreamId = maps:get(name, amqqueue:get_type_state(Q)),
     process_command({policy_changed, StreamId, #{queue => Q}}).
 
+sac_state(#?MODULE{single_active_consumer = SacState}) ->
+    SacState.
+
 %% for debugging
 state() ->
     case ra:local_query({?MODULE, node()}, fun(State) -> State end) of
@@ -338,15 +346,15 @@ all_coord_members() ->
     Nodes = rabbit_mnesia:cluster_nodes(running) -- [node()],
     [{?MODULE, Node} || Node <- [node() | Nodes]].
 
-version() -> 2.
+version() -> 3.
 
 which_module(_) ->
     ?MODULE.
 
 init(_Conf) ->
-    #?MODULE{}.
+    #?MODULE{single_active_consumer = rabbit_stream_sac_coordinator:init_state()}.
 
--spec apply(map(), command(), state()) ->
+-spec apply(ra_machine:command_meta_data(), command(), state()) ->
     {state(), term(), ra_machine:effects()}.
 apply(#{index := _Idx, machine_version := MachineVersion} = Meta0,
       {_CmdTag, StreamId, #{}} = Cmd,
@@ -380,11 +388,18 @@ apply(#{index := _Idx, machine_version := MachineVersion} = Meta0,
         Reply ->
             return(Meta, State0, Reply, [])
     end;
+apply(Meta, {sac, SacCommand}, #?MODULE{single_active_consumer = SacState0,
+                                        monitors = Monitors0} = State0) ->
+    {SacState1, Reply, Effects0} = rabbit_stream_sac_coordinator:apply(SacCommand, SacState0),
+    {SacState2, Monitors1, Effects1} =
+         rabbit_stream_sac_coordinator:ensure_monitors(SacCommand, SacState1, Monitors0, Effects0),
+    return(Meta, State0#?MODULE{single_active_consumer = SacState2,
+                                 monitors = Monitors1}, Reply, Effects1);
 apply(#{machine_version := MachineVersion} = Meta, {down, Pid, Reason} = Cmd,
       #?MODULE{streams = Streams0,
                monitors = Monitors0,
-               listeners = StateListeners0} = State) ->
-
+               listeners = StateListeners0,
+               single_active_consumer = SacState0 } = State) ->
     Effects0 = case Reason of
                    noconnection ->
                        [{monitor, node, node(Pid)}];
@@ -438,6 +453,10 @@ apply(#{machine_version := MachineVersion} = Meta, {down, Pid, Reason} = Cmd,
                     return(Meta, State#?MODULE{streams = Streams0,
                                                monitors = Monitors1}, ok, Effects0)
             end;
+        {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);
         error ->
             return(Meta, State, ok, Effects0)
     end;
@@ -1784,6 +1803,9 @@ machine_version(1, 2, State = #?MODULE{streams = Streams0,
     {State#?MODULE{streams = Streams1,
                    monitors = Monitors2,
                    listeners = undefined}, Effects};
+machine_version(2, 3, State) ->
+    rabbit_log:info("Stream coordinator machine version changes from 2 to 3, updating state."),
+    {State#?MODULE{single_active_consumer = rabbit_stream_sac_coordinator:init_state()}, []};
 machine_version(From, To, State) ->
     rabbit_log:info("Stream coordinator machine version changes from ~p to ~p, no state changes required.",
                     [From, To]),

deps/rabbit/src/rabbit_stream_coordinator.hrl

@@ -59,10 +59,11 @@
 -record(rabbit_stream_coordinator, {streams = #{} :: #{stream_id() => #stream{}},
                                     monitors = #{} :: #{pid() => {stream_id() | %% v0 & v1
                                                                   #{stream_id() => ok}, %% v2
-                                                                  monitor_role()}},
+                                                                  monitor_role()} |
+                                                        sac},
                                     %% not used as of v2
                                     listeners = #{} :: undefined | #{stream_id() =>
                                                                      #{pid() := queue_ref()}},
+                                    single_active_consumer = undefined :: undefined | rabbit_stream_sac_coordinator:state(),
                                     %% future extensibility
-                                    reserved_1,
                                     reserved_2}).