fix(emqx_conf): avoid crash/deadlock depending on node startup order

apps/emqx_conf/src/emqx_cluster_rpc.erl

@@ -42,6 +42,8 @@
     code_change/3
 ]).
 
+-export([get_tables_status/0]).
+
 -export_type([txn_id/0, succeed_num/0, multicall_return/1, multicall_return/0]).
 
 -ifdef(TEST).
@@ -172,6 +174,29 @@ get_node_tnx_id(Node) ->
         [#cluster_rpc_commit{tnx_id = TnxId}] -> TnxId
     end.
 
+%% Checks whether the Mnesia tables used by this module are waiting to
+%% be loaded and from where.
+-spec get_tables_status() -> #{atom() => {waiting, [node()]} | {disc | network, node()}}.
+get_tables_status() ->
+    maps:from_list([
+        {Tab, do_get_tables_status(Tab)}
+     || Tab <- [?CLUSTER_COMMIT, ?CLUSTER_MFA]
+    ]).
+
+do_get_tables_status(Tab) ->
+    Props = mnesia:table_info(Tab, all),
+    TabNodes = proplists:get_value(all_nodes, Props),
+    KnownDown = mnesia_recover:get_mnesia_downs(),
+    LocalNode = node(),
+    case proplists:get_value(load_node, Props) of
+        unknown ->
+            {waiting, TabNodes -- [LocalNode | KnownDown]};
+        LocalNode ->
+            {disc, LocalNode};
+        Node ->
+            {network, Node}
+    end.
+
 %% Regardless of what MFA is returned, consider it a success),
 %% then move to the next tnxId.
 %% if the next TnxId failed, need call the function again to skip.

apps/emqx_conf/src/emqx_conf_app.erl

@@ -24,26 +24,57 @@
 -include_lib("emqx/include/logger.hrl").
 -include("emqx_conf.hrl").
 
+-define(DEFAULT_INIT_TXN_ID, -1).
+
 start(_StartType, _StartArgs) ->
     init_conf(),
     emqx_conf_sup:start_link().
 
 stop(_State) ->
     ok.
 
-%% internal functions
+get_override_config_file() ->
+    Node = node(),
+    case emqx_app:get_init_config_load_done() of
+        false ->
+            {error, #{node => Node, msg => "init_conf_load_not_done"}};
+        true ->
+            case erlang:whereis(emqx_config_handler) of
+                undefined ->
+                    {error, #{node => Node, msg => "emqx_config_handler_not_ready"}};
+                _ ->
+                    Fun = fun() ->
+                        TnxId = emqx_cluster_rpc:get_node_tnx_id(Node),
+                        WallClock = erlang:statistics(wall_clock),
+                        Conf = emqx_config_handler:get_raw_cluster_override_conf(),
+                        #{wall_clock => WallClock, conf => Conf, tnx_id => TnxId, node => Node}
+                    end,
+                    case mria:ro_transaction(?CLUSTER_RPC_SHARD, Fun) of
+                        {atomic, Res} -> {ok, Res};
+                        {aborted, Reason} -> {error, #{node => Node, msg => Reason}}
+                    end
+            end
+    end.
+
+%% ------------------------------------------------------------------------------
+%% Internal functions
+%% ------------------------------------------------------------------------------
+
 init_conf() ->
     {ok, TnxId} = copy_override_conf_from_core_node(),
     emqx_app:set_init_tnx_id(TnxId),
     emqx_config:init_load(emqx_conf:schema_module()),
     emqx_app:set_init_config_load_done().
 
+cluster_nodes() ->
+    maps:get(running_nodes, ekka_cluster:info()) -- [node()].
+
 copy_override_conf_from_core_node() ->
-    case mria_mnesia:running_nodes() -- [node()] of
+    case cluster_nodes() of
         %% The first core nodes is self.
         [] ->
             ?SLOG(debug, #{msg => "skip_copy_overide_conf_from_core_node"}),
-            {ok, -1};
+            {ok, ?DEFAULT_INIT_TXN_ID};
         Nodes ->
             {Results, Failed} = emqx_conf_proto_v1:get_override_config_file(Nodes),
             {Ready, NotReady0} = lists:partition(fun(Res) -> element(1, Res) =:= ok end, Results),
@@ -64,12 +95,39 @@ copy_override_conf_from_core_node() ->
                 [] ->
                     %% Other core nodes running but no one replicated it successfully.
                     ?SLOG(error, #{
-                        msg => "copy_overide_conf_from_core_node_failed",
+                        msg => "copy_override_conf_from_core_node_failed",
                         nodes => Nodes,
                         failed => Failed,
                         not_ready => NotReady
                     }),
-                    {error, "core node not ready"};
+
+                    case should_proceed_with_boot() of
+                        true ->
+                            %% Act as if this node is alone, so it can
+                            %% finish the boot sequence and load the
+                            %% config for other nodes to copy it.
+                            ?SLOG(info, #{
+                                msg => "skip_copy_overide_conf_from_core_node",
+                                loading_from_disk => true,
+                                nodes => Nodes,
+                                failed => Failed,
+                                not_ready => NotReady
+                            }),
+                            {ok, ?DEFAULT_INIT_TXN_ID};
+                        false ->
+                            %% retry in some time
+                            Jitter = rand:uniform(2_000),
+                            Timeout = 10_000 + Jitter,
+                            ?SLOG(info, #{
+                                msg => "copy_overide_conf_from_core_node_retry",
+                                timeout => Timeout,
+                                nodes => Nodes,
+                                failed => Failed,
+                                not_ready => NotReady
+                            }),
+                            timer:sleep(Timeout),
+                            copy_override_conf_from_core_node()
+                    end;
                 _ ->
                     SortFun = fun(
                         {ok, #{wall_clock := W1}},
@@ -79,7 +137,10 @@ copy_override_conf_from_core_node() ->
                     end,
                     [{ok, Info} | _] = lists:sort(SortFun, Ready),
                     #{node := Node, conf := RawOverrideConf, tnx_id := TnxId} = Info,
-                    Msg = #{msg => "copy_overide_conf_from_core_node_success", node => Node},
+                    Msg = #{
+                        msg => "copy_overide_conf_from_core_node_success",
+                        node => Node
+                    },
                     ?SLOG(debug, Msg),
                     ok = emqx_config:save_to_override_conf(
                         RawOverrideConf,
@@ -89,28 +150,16 @@ copy_override_conf_from_core_node() ->
             end
     end.
 
-get_override_config_file() ->
-    Node = node(),
-    Role = mria_rlog:role(),
-    case emqx_app:get_init_config_load_done() of
-        false ->
-            {error, #{node => Node, msg => "init_conf_load_not_done"}};
-        true when Role =:= core ->
-            case erlang:whereis(emqx_config_handler) of
-                undefined ->
-                    {error, #{node => Node, msg => "emqx_config_handler_not_ready"}};
-                _ ->
-                    Fun = fun() ->
-                        TnxId = emqx_cluster_rpc:get_node_tnx_id(Node),
-                        WallClock = erlang:statistics(wall_clock),
-                        Conf = emqx_config_handler:get_raw_cluster_override_conf(),
-                        #{wall_clock => WallClock, conf => Conf, tnx_id => TnxId, node => Node}
-                    end,
-                    case mria:ro_transaction(?CLUSTER_RPC_SHARD, Fun) of
-                        {atomic, Res} -> {ok, Res};
-                        {aborted, Reason} -> {error, #{node => Node, msg => Reason}}
-                    end
-            end;
-        true when Role =:= replicant ->
-            {ignore, #{node => Node}}
+should_proceed_with_boot() ->
+    TablesStatus = emqx_cluster_rpc:get_tables_status(),
+    LocalNode = node(),
+    case maps:get(?CLUSTER_COMMIT, TablesStatus) of
+        {disc, LocalNode} ->
+            %% Loading locally; let this node finish its boot sequence
+            %% so others can copy the config from this one.
+            true;
+        _ ->
+            %% Loading from another node or still waiting for nodes to
+            %% be up.  Try again.
+            false
     end.