SC: fail active mnesia update actions on leader change

Else it is possible that any mnesia update actions that are in
progress when a stream coordinator leader change occurs get stuck
and never restart.

Exclude test from mixed versions

The assertion fixes a bug and requires the new code to be running.

2 files changed, 136 insertions, 9 deletions

diff --git a/deps/rabbit/src/rabbit_stream_coordinator.erl b/deps/rabbit/src/rabbit_stream_coordinator.erl
index 4371c2de81..f6aaa9d2c4 100644
--- a/deps/rabbit/src/rabbit_stream_coordinator.erl
+++ b/deps/rabbit/src/rabbit_stream_coordinator.erl
@@ -22,6 +22,7 @@
          overview/1]).
 
 -export([recover/0,
+         stop/0,
          add_replica/2,
          delete_replica/2,
          register_listener/1,
@@ -33,8 +34,10 @@
 -export([policy_changed/1]).
 
 -export([local_pid/1,
+         writer_pid/1,
          members/1]).
 -export([query_local_pid/3,
+         query_writer_pid/2,
          query_members/2]).
 
 
@@ -113,6 +116,16 @@ recover() ->
             ok
     end.
 
+%% stop the stream coordinator on the local node
+stop() ->
+    case erlang:whereis(?MODULE) of
+        undefined ->
+            ok;
+        _Pid ->
+            ra:stop_server(?RA_SYSTEM, {?MODULE, node()})
+    end.
+
+
 %% new api
 
 new_stream(Q, LeaderNode)
@@ -189,11 +202,18 @@ state() ->
             Any
     end.
 
+writer_pid(StreamId) when is_list(StreamId) ->
+    MFA = {?MODULE, query_writer_pid, [StreamId]},
+    query_pid(StreamId, MFA).
+
 local_pid(StreamId) when is_list(StreamId) ->
     MFA = {?MODULE, query_local_pid, [StreamId, node()]},
+    query_pid(StreamId, MFA).
+
+query_pid(StreamId, MFA) when is_list(StreamId) ->
     case ra:local_query({?MODULE, node()}, MFA) of
         {ok, {_, {ok, Pid}}, _} ->
-            case is_process_alive(Pid) of
+            case erpc:call(node(Pid), erlang, is_process_alive, [Pid]) of
                 true ->
                     {ok, Pid};
                 false ->
@@ -264,6 +284,20 @@ query_local_pid(StreamId, Node, #?MODULE{streams = Streams}) ->
             {error, not_found}
     end.
 
+query_writer_pid(StreamId, #?MODULE{streams = Streams}) ->
+    case Streams of
+        #{StreamId := #stream{members = Members}} ->
+            maps:fold(
+              fun (_Node, #member{role = {writer, _},
+                                  state = {running, _, Pid}}, _Acc) ->
+                      {ok, Pid};
+                  (_, _, Acc) ->
+                      Acc
+              end, {error, writer_not_found}, Members);
+        _ ->
+            {error, stream_not_found}
+    end.
+
 -spec register_listener(amqqueue:amqqueue()) ->
     {error, term()} | {ok, ok | stream_not_found, atom() | {atom(), atom()}}.
 register_listener(Q) when ?is_amqqueue(Q)->
@@ -711,11 +745,14 @@ handle_aux(leader, _, {delete_member, StreamId, #{node := Node} = Args, Conf},
     ActionFun = phase_delete_member(StreamId, Args, Conf),
     run_action(delete_member, StreamId, Args, ActionFun, Aux, LogState);
 handle_aux(leader, _, fail_active_actions,
-           #aux{actions = Monitors} = Aux, LogState,
+           #aux{actions = Actions} = Aux, LogState,
            #?MODULE{streams = Streams}) ->
+    %% this bit of code just creates an exclude map of currently running
+    %% tasks to avoid failing them, this could only really happen during
+    %% a leader flipflap
     Exclude = maps:from_list([{S, ok}
-                              || {P, {S, _, _}} <- maps:to_list(Monitors),
-                             not is_process_alive(P)]),
+                              || {P, {S, _, _}} <- maps:to_list(Actions),
+                             is_process_alive(P)]),
     rabbit_log:debug("~ts: failing actions: ~w", [?MODULE, Exclude]),
     fail_active_actions(Streams, Exclude),
     {no_reply, Aux, LogState, []};
@@ -1707,11 +1744,27 @@ eval_replica(_Meta, #member{node = Node} = Replica, _LeaderState, _Stream,
 
 fail_active_actions(Streams, Exclude) ->
     maps:map(
-      fun (_,  #stream{id = Id, members = Members})
+      fun (_,  #stream{id = Id,
+                       members = Members,
+                       mnesia = Mnesia})
             when not is_map_key(Id, Exclude)  ->
               _ = maps:map(fun(_, M) ->
                                    fail_action(Id, M)
-                           end, Members)
+                           end, Members),
+              case Mnesia of
+                  {updating, E} ->
+                      rabbit_log:debug("~ts: failing active action for ~ts node
+                                       ~w Action ~w",
+                                       [?MODULE, Id, node(), updating_mnesia]),
+                      send_self_command({action_failed, Id,
+                                         #{action => updating_mnesia,
+                                           index => 0,
+                                           node => node(),
+                                           epoch => E}});
+                  _ ->
+                      ok
+              end,
+              ok
       end, Streams),
 
     ok.
diff --git a/deps/rabbit/test/rabbit_stream_queue_SUITE.erl b/deps/rabbit/test/rabbit_stream_queue_SUITE.erl
index 1f197dd655..02fc475d51 100644
--- a/deps/rabbit/test/rabbit_stream_queue_SUITE.erl
+++ b/deps/rabbit/test/rabbit_stream_queue_SUITE.erl
@@ -54,7 +54,8 @@ groups() ->
            queue_size_on_declare,
            leader_locator_balanced,
            leader_locator_balanced_maintenance,
-           select_nodes_with_least_replicas
+           select_nodes_with_least_replicas,
+           recover_after_leader_and_coordinator_kill
           ]},
      {cluster_size_3_1, [], [shrink_coordinator_cluster]},
      {cluster_size_3_2, [], [recover,
@@ -217,7 +218,8 @@ init_per_testcase(TestCase, Config)
 init_per_testcase(TestCase, Config)
   when TestCase == replica_recovery
        orelse TestCase == leader_failover
-       orelse TestCase == leader_failover_dedupe ->
+       orelse TestCase == leader_failover_dedupe
+       orelse TestCase == recover_after_leader_and_coordinator_kill ->
     case rabbit_ct_helpers:is_mixed_versions() of
         true ->
             %% not supported because of machine version difference
@@ -587,7 +589,8 @@ grow_coordinator_cluster(Config) ->
     %% and adds a new member on the new node
     rabbit_ct_helpers:await_condition(
       fun() ->
-              case rpc:call(Server0, ra, members, [{rabbit_stream_coordinator, Server0}]) of
+              case rpc:call(Server0, ra, members,
+                            [{rabbit_stream_coordinator, Server0}]) of
                   {_, Members, _} ->
                       Nodes = lists:sort([N || {_, N} <- Members]),
                       lists:sort([Server0, Server1]) == Nodes;
@@ -1129,6 +1132,45 @@ receive_basic_cancel_on_queue_deletion(Config) ->
             exit(timeout)
     end.
 
+recover_after_leader_and_coordinator_kill(Config) ->
+    [Server1 | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    Ch1 = rabbit_ct_client_helpers:open_channel(Config, Server1),
+    % Ch2 = rabbit_ct_client_helpers:open_channel(Config, Server1),
+    Q = ?config(queue_name, Config),
+    ?assertEqual({'queue.declare_ok', Q, 0, 0},
+                 declare(Ch1, Q, [{<<"x-queue-type">>, longstr, <<"stream">>}])),
+
+    publish_confirm(Ch1, Q, [<<"msg 1">>]),
+
+    QName = rabbit_misc:r(<<"/">>, queue, Q),
+    [begin
+         Sleep = rand:uniform(10),
+         ct:pal("Attempt ~b Sleep ~b", [Num, Sleep]),
+         {ok, {_LeaderNode, LeaderPid}} = leader_info(Config),
+         {_, CoordNode} = get_stream_coordinator_leader(Config),
+         kill_stream_leader_then_coordinator_leader(Config, CoordNode,
+                                                    LeaderPid, Sleep),
+         publish_confirm(Ch1, Q, [<<"msg">>]),
+         recover_coordinator(Config, CoordNode),
+         timer:sleep(Sleep),
+         rabbit_ct_helpers:await_condition(
+           fun () ->
+                   rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE,
+                                                validate_writer_pid, [QName])
+           end)
+     end || Num <- lists:seq(1, 10)],
+
+    {_, Node} = get_stream_coordinator_leader(Config),
+
+    CState = rabbit_ct_broker_helpers:rpc(Config, Node, sys,
+                                          get_state,
+                                          [rabbit_stream_coordinator]),
+
+
+    ct:pal("sys state ~p", [CState]),
+
+    ok.
+
 keep_consuming_on_leader_restart(Config) ->
     [Server1 | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
 
@@ -1172,6 +1214,13 @@ leader_info(Config) ->
     rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, get_leader_info,
                                  [QName]).
 
+validate_writer_pid(QName) ->
+    {ok, Q} = rabbit_amqqueue:lookup(QName),
+    MnesiaPid = amqqueue:get_pid(Q),
+    QState = amqqueue:get_type_state(Q),
+    #{name := StreamId} = QState,
+    {ok, MnesiaPid} == rabbit_stream_coordinator:writer_pid(StreamId).
+
 get_leader_info(QName) ->
     {ok, Q} = rabbit_amqqueue:lookup(QName),
     QState = amqqueue:get_type_state(Q),
@@ -1195,6 +1244,31 @@ kill_process(Config, Node, Pid) ->
 do_kill_process(Pid) ->
     exit(Pid, kill).
 
+do_stop_start_coordinator() ->
+    rabbit_stream_coordinator:stop(),
+    timer:sleep(10),
+    rabbit_stream_coordinator:recover().
+
+recover_coordinator(Config, Node) ->
+    rabbit_ct_broker_helpers:rpc(Config, Node, rabbit_stream_coordinator,
+                                 recover, []).
+
+get_stream_coordinator_leader(Config) ->
+    Node = hd(rabbit_ct_broker_helpers:get_node_configs(Config, nodename)),
+    rabbit_ct_broker_helpers:rpc(Config, Node, ra_leaderboard,
+                                 lookup_leader, [rabbit_stream_coordinator]).
+
+kill_stream_leader_then_coordinator_leader(Config, CoordLeaderNode,
+                                           StreamLeaderPid, Sleep) ->
+    rabbit_ct_broker_helpers:rpc(Config, CoordLeaderNode, ?MODULE,
+                                 do_stop_kill, [StreamLeaderPid, Sleep]).
+
+do_stop_kill(Pid, Sleep) ->
+    exit(Pid, kill),
+    timer:sleep(Sleep),
+    rabbit_stream_coordinator:stop().
+
+
 filter_consumers(Config, Server, CTag) ->
     CInfo = rabbit_ct_broker_helpers:rpc(Config, Server, ets, tab2list, [consumer_created]),
     lists:foldl(fun(Tuple, Acc) ->