Merge pull request #184 from keynslug/fix/EMQX-13588/stricter-lock

fix: make global lock guarding join operations stricter

Author: keynslug

include/mria.hrl

@@ -17,4 +17,4 @@
 
 -type(member() :: #member{}).
 
--define(JOIN_LOCK_ID, {mria_sync_join, node()}).
+-define(JOIN_LOCK_ID(REQUESTER), {mria_sync_join, REQUESTER}).

rebar.config

@@ -2,8 +2,8 @@
 {minimum_otp_vsn, "21.0"}.
 
 {deps,
- [{snabbkaffe, {git, "https://github.com/kafka4beam/snabbkaffe", {tag, "1.0.7"}}},
-  {gen_rpc, {git, "https://github.com/emqx/gen_rpc", {tag, "3.3.0"}}},
+ [{snabbkaffe, {git, "https://github.com/kafka4beam/snabbkaffe", {tag, "1.0.10"}}},
+  {gen_rpc, {git, "https://github.com/emqx/gen_rpc", {tag, "3.4.1"}}},
   {replayq, {git, "https://github.com/emqx/replayq", {tag, "0.3.6"}}},
   {mnesia_rocksdb, {git, "https://github.com/emqx/mnesia_rocksdb", {tag, "0.1.16"}}},
   {optvar, {git, "https://github.com/emqx/optvar", {tag, "1.0.5"}}}

src/mria.erl

@@ -235,11 +235,22 @@ join(Node) ->
 join(Node, _) when Node =:= node() ->
     ignore;
 join(Node, Reason) when is_atom(Node) ->
+    %% NOTE
+    %%
     %% If two nodes are trying to join each other simultaneously,
     %% one of them must be blocked waiting for a lock.
     %% Once lock is released, it is expected to be already in the
     %% cluster (if the other node joined it successfully).
-    global:trans(?JOIN_LOCK_ID, fun() -> join1(Node, Reason) end, [node(), Node]).
+    %%
+    %% Additionally, avoid conducting concurrent join operations
+    %% by specifying current process PID as the lock requester.
+    %% Otherwise, concurrent joins can ruin each other's lives and
+    %% make any further cluster operations impossible.
+    %% This can happen, for example, when a concurrent join stops the
+    %% entire `mnesia` system while another join is running schema
+    %% transactions.
+    LockId = ?JOIN_LOCK_ID(self()),
+    global:trans(LockId, fun() -> join1(Node, Reason) end, [node(), Node]).
 
 %% @doc Leave the cluster
 -spec leave() -> ok | {error, term()}.

src/mria_mnesia.erl

@@ -130,6 +130,7 @@ ensure_stopped() ->
 %% @doc Cluster with node.
 -spec(connect(node()) -> ok | {error, any()}).
 connect(Node) ->
+    ?tp(mria_mnesia_connect, #{to => Node}),
     case mnesia:change_config(extra_db_nodes, [Node]) of
         {ok, [Node]} -> ok;
         {ok, []}     -> {error, {failed_to_connect_node, Node}};
@@ -237,6 +238,7 @@ is_node_in_cluster(Node) ->
 
 %% @doc Copy schema.
 copy_schema(Node) ->
+    ?tp(mria_mnesia_copy_schema, #{}),
     case mnesia:change_table_copy_type(schema, Node, disc_copies) of
         {atomic, ok} -> ok;
         {aborted, {already_exists, schema, Node, disc_copies}} ->

test/mria_SUITE.erl

@@ -1317,6 +1317,43 @@ t_join_another_node_simultaneously(_) ->
        end,
       []).
 
+t_join_many_nodes_simultaneously(_) ->
+    % Self = self(),
+    CommonEnv = mria_mnesia_test_util:common_env(),
+    Cluster = [maps:remove(join_to, Spec)
+               || Spec <- mria_ct:cluster([core, core, core, core], CommonEnv)],
+    ?check_trace(
+       #{timetrap => 15_000},
+       try
+           %% Spin the cluster up.
+           [N1, N2, N3, N4] = Nodes = mria_ct:start_cluster(mria, Cluster),
+           %% Connect only N2, N3, N4 together.
+           ok = rpc:call(N2, mria, join, [N4]),
+           ok = rpc:call(N3, mria, join, [N4]),
+           %% Subscribe to an event emitted right before schema transactions take place.
+           {ok, SRef} = snabbkaffe:subscribe(?match_event(#{?snk_kind := mria_mnesia_connect})),
+           %% Ask N1 to join the cluster (using N2 as a seed).
+           K1 = rpc:async_call(N1, mria, join, [N2]),
+           %% Wait for the event, and ask (concurrently) N1 to join the cluster (using
+           %% other 2 nodes as seeds).
+           {ok, _} = snabbkaffe:receive_events(SRef),
+           K2 = rpc:async_call(N1, mria, join, [N3]),
+           K3 = rpc:async_call(N1, mria, join, [N4]),
+           ?assertMatch([ok,
+                         {error, {already_in_cluster, _}},
+                         {error, {already_in_cluster, _}}],
+                        lists:sort([rpc:yield(K) || K <- [K1, K2, K3]])),
+           timer:sleep(3000),
+           ?assertEqual({[true, true, true, true], []},
+                        rpc:multicall(Nodes, mria_sup, is_running, [])),
+           {Results, []} = rpc:multicall(Nodes, mria_mnesia, running_nodes, []),
+           ?assertEqual([Nodes, Nodes, Nodes, Nodes],
+                        lists:map(fun lists:sort/1, Results))
+       after
+           ok = mria_ct:teardown_cluster(Cluster)
+       end,
+      []).
+
 cluster_benchmark(_) ->
     NReplicas = 6,
     Config = #{ trans_size => 10