Introduce stream SAC status instead of active flag

deps/rabbit/Makefile

@@ -268,7 +268,7 @@ PARALLEL_CT_SET_2_B = clustering_recovery crashing_queues deprecated_features di
 PARALLEL_CT_SET_2_C = disk_monitor dynamic_qq unit_disk_monitor unit_file_handle_cache unit_log_management unit_operator_policy
 PARALLEL_CT_SET_2_D = queue_length_limits queue_parallel quorum_queue_member_reconciliation rabbit_fifo rabbit_fifo_dlx rabbit_stream_coordinator
 
-PARALLEL_CT_SET_3_A = definition_import per_user_connection_channel_limit_partitions per_vhost_connection_limit_partitions policy priority_queue_recovery rabbit_fifo_prop rabbit_fifo_v0 rabbit_stream_sac_coordinator unit_credit_flow unit_queue_consumers unit_queue_location unit_quorum_queue
+PARALLEL_CT_SET_3_A = definition_import per_user_connection_channel_limit_partitions per_vhost_connection_limit_partitions policy priority_queue_recovery rabbit_fifo_prop rabbit_fifo_v0 rabbit_stream_sac_coordinator_v4 rabbit_stream_sac_coordinator unit_credit_flow unit_queue_consumers unit_queue_location unit_quorum_queue
 PARALLEL_CT_SET_3_B = cluster_upgrade list_consumers_sanity_check list_queues_online_and_offline logging lqueue maintenance_mode rabbit_fifo_q
 PARALLEL_CT_SET_3_C = cli_forget_cluster_node feature_flags_v2 mc_unit message_containers_deaths_v2 message_size_limit metadata_store_migration
 PARALLEL_CT_SET_3_D = metadata_store_phase1 metrics mirrored_supervisor peer_discovery_classic_config proxy_protocol runtime_parameters unit_stats_and_metrics unit_supervisor2 unit_vm_memory_monitor

deps/rabbit/ct.test.spec

@@ -117,6 +117,7 @@
 ,   rabbit_local_random_exchange_SUITE
 ,   rabbit_msg_interceptor_SUITE
 ,   rabbit_stream_coordinator_SUITE
+,   rabbit_stream_sac_coordinator_v4_SUITE
 ,   rabbit_stream_sac_coordinator_SUITE
 ,   rabbitmq_4_0_deprecations_SUITE
 ,   rabbitmq_queues_cli_integration_SUITE

deps/rabbit/src/rabbit_stream_coordinator.hrl

@@ -68,6 +68,7 @@
          listeners = #{} :: undefined | #{stream_id() =>
                                           #{pid() := queue_ref()}},
          single_active_consumer = undefined :: undefined |
+                                               rabbit_stream_sac_coordinator_v4:state() |
                                                rabbit_stream_sac_coordinator:state(),
          %% future extensibility
          reserved_2}).

deps/rabbit/src/rabbit_stream_sac_coordinator.hrl

@@ -22,22 +22,28 @@
 -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.
 
 -record(consumer,
         {pid :: pid(),
          subscription_id :: subscription_id(),
          owner :: owner(), %% just a label
-         active :: boolean()}).
+         status :: {consumer_connectivity(), consumer_status()}}).
 -record(group,
         {consumers :: [#consumer{}], partition_index :: integer()}).
 -record(rabbit_stream_sac_coordinator,
-        {groups :: #{group_id() => #group{}},
-         pids_groups ::
-             #{connection_pid() =>
-                   #{group_id() => true}}, %% inner map acts as a set
+        {groups :: groups(),
+         pids_groups :: pids_groups(),
          %% future extensibility
          reserved_1,
          reserved_2}).
+
+-type group() :: #group{}.
+-type groups() :: #{group_id() => group()}.
+%% inner map acts as a set
+-type pids_groups() :: #{connection_pid() => #{group_id() => true}}.
+
 %% commands
 -record(command_register_consumer,
         {vhost :: vhost(),
@@ -56,3 +62,7 @@
 -record(command_activate_consumer,
         {vhost :: vhost(), stream :: stream(),
          consumer_name :: consumer_name()}).
+-record(command_connection_reconnected,
+        {pid :: connection_pid()}).
+-record(command_purge_nodes,
+        {nodes :: [node()]}).