Miscellaneous minor improvements in stream SAC coordinator

This commit handles edge cases in the stream SAC coordinator to make
sure it does not crash during execution. Most of these edge cases
consist in an inconsistent state, so there are very unlikely to happen.

This commit also makes sure there is no duplicate in the consumer list
of a group. Consumers are also now identified only by their connection
PID and their subscription ID, as now the timestamp they contain in
their state does not allow a field-by-field comparison.

Author: acogoluegnes

deps/rabbit/src/rabbit_stream_coordinator.erl

@@ -710,8 +710,7 @@ apply(#{machine_version := Vsn} = Meta,
         _ ->
             return(Meta, State0, stream_not_found, [])
     end;
-apply(#{machine_version := Vsn} = Meta,
-      {nodeup, Node} = Cmd,
+apply(Meta, {nodeup, Node} = Cmd,
       #?MODULE{monitors = Monitors0,
                streams = Streams0,
                single_active_consumer = Sac0} = State)  ->
@@ -735,14 +734,8 @@ apply(#{machine_version := Vsn} = Meta,
                           {Ss#{Id => S}, E}
                   end, {Streams0, Effects0}, Streams0),
 
-    {Sac1, Effects2} = case ?V5_OR_MORE(Vsn) of
-                           true ->
-                               SacMod = sac_module(Meta),
-                               SacMod:handle_node_reconnected(Node,
-                                                              Sac0, Effects1);
-                           false ->
-                               {Sac0, Effects1}
-                       end,
+
+    {Sac1, Effects2} = sac_handle_node_reconnected(Meta, Node, Sac0, Effects1),
     return(Meta, State#?MODULE{monitors = Monitors,
                                streams = Streams,
                                single_active_consumer = Sac1}, ok, Effects2);
@@ -2444,6 +2437,17 @@ sac_handle_connection_down(SacState, Pid, Reason, Vsn) when ?V5_OR_MORE(Vsn) ->
 sac_handle_connection_down(SacState, Pid, _Reason, _Vsn) ->
     ?SAC_V4:handle_connection_down(Pid, SacState).
 
+sac_handle_node_reconnected(#{machine_version := Vsn} = Meta, Node,
+                            Sac, Effects) ->
+    case ?V5_OR_MORE(Vsn) of
+        true ->
+            SacMod = sac_module(Meta),
+            SacMod:handle_node_reconnected(Node,
+                                           Sac, Effects);
+        false ->
+            {Sac, Effects}
+    end.
+
 sac_make_purge_nodes(Nodes) ->
     rabbit_stream_sac_coordinator:make_purge_nodes(Nodes).