Merge pull request #188 from emqx/dev/fix-rpc-target2

ieQu1 · web-flow · commit d6123c790f0e · 2025-03-14T11:05:51.000+01:00
fix(rpc): Follow-up to the previous fix
diff --git a/src/mria.erl b/src/mria.erl
@@ -595,7 +595,7 @@ should_retry_rpc(_) ->
 find_upstream_node(Shard) ->
     ?tp_span(find_upstream_node, #{shard => Shard},
              begin
-                 {ok, Node} = mria_status:upstream_node(Shard, infinity),
+                 {ok, Node} = mria_status:rpc_target(Shard, infinity),
                  Node
              end).
 
@@ -688,7 +688,7 @@ db_nodes_maybe_rpc() ->
         replicant ->
             case mria_status:shards_up() of
                 [Shard|_] ->
-                    {ok, CoreNode} = mria_status:upstream_node(Shard, 5_000),
+                    {ok, CoreNode} = mria_status:rpc_target(Shard, 5_000),
                     case mria_lib:rpc_call_nothrow(CoreNode, mnesia, system_info, [db_nodes]) of
                         {badrpc, _} -> [];
                         {badtcp, _} -> [];
diff --git a/src/mria_rlog_replica.erl b/src/mria_rlog_replica.erl
@@ -1,5 +1,5 @@
 %%--------------------------------------------------------------------
-%% Copyright (c) 2021-2023 EMQ Technologies Co., Ltd. All Rights Reserved.
+%% Copyright (c) 2021-2025 EMQ Technologies Co., Ltd. All Rights Reserved.
 %%
 %% Licensed under the Apache License, Version 2.0 (the "License");
 %% you may not use this file except in compliance with the License.
@@ -226,6 +226,7 @@ initiate_bootstrap(D) ->
       , remote_core_node = Remote
       , parent_sup       = ParentSup
       } = D,
+    mria_status:notify_rpc_target_up(Shard, Remote),
     _Pid = mria_replicant_shard_sup:start_bootstrap_client(ParentSup, Shard, Remote, self()),
     ReplayqMemOnly = application:get_env(mria, rlog_replayq_mem_only, true),
     ReplayqBaseDir = application:get_env(mria, rlog_replayq_dir, "/tmp/rlog"),
@@ -256,6 +257,7 @@ handle_agent_down(State, Reason, D) ->
         #{ reason => Reason
          , repl_state => State
          }),
+    mria_status:notify_rpc_target_down(D#d.shard),
     case State of
         ?normal ->
             {next_state, ?disconnected, D#d{agent = undefined}};
@@ -378,7 +380,7 @@ handle_reconnect(D0 = #d{shard = Shard, checkpoint = Checkpoint, parent_sup = Pa
 try_connect(Shard, Checkpoint) ->
     Timeout = 4_000, % Don't block FSM forever, allow it to process other messages.
     %% Get the best node according to the LB
-    Nodes = case mria_status:get_core_node(Shard, Timeout) of
+    Nodes = case mria_status:replica_get_core_node(Shard, Timeout) of
                 {ok, N} -> [N];
                 timeout -> []
             end,
diff --git a/src/mria_status.erl b/src/mria_status.erl
@@ -24,11 +24,12 @@
 
 %% API:
 -export([start_link/0,
+         notify_rpc_target_up/2, notify_rpc_target_down/1, rpc_target/2,
          notify_shard_up/2, notify_shard_down/1, wait_for_shards/2,
-         notify_core_node_up/2, notify_core_node_down/1, get_core_node/2,
+         notify_core_node_up/2, notify_core_node_down/1, replica_get_core_node/2,
          notify_core_intercept_trans/2,
 
-         upstream/1, upstream_node/2,
+         upstream/1, upstream_node/1,
          shards_status/0, shards_up/0, shards_syncing/0, shards_down/0,
          get_shard_stats/1, agents/0, agents/1, replicants/0, get_shard_lag/1,
 
@@ -57,11 +58,13 @@
 -include("mria_rlog.hrl").
 -include_lib("snabbkaffe/include/trace.hrl").
 
-%% Tables and table keys:
+%% Optvars:
 -define(optvar(KEY), {mria, KEY}).
 -define(upstream_pid, upstream_pid).
 -define(core_node, core_node).
+-define(rpc_target, rpc_target).
 
+%% Tables and table keys:
 -define(stats_tab, mria_rlog_stats_tab).
 -define(core_intercept, core_intercept).
 -define(replicant_state, replicant_state).
@@ -80,29 +83,64 @@
 start_link() ->
     gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
 
-%% @doc Return name of the core node that is _currently serving_ the
-%% downstream shard. If shard is currently down, wait until started
-%% for at most `Timeout' millisecond. Note the difference in behavior
-%% as compared with `get_core_node'.
--spec upstream_node(mria_rlog:shard(), timeout()) -> {ok, node()} | timeout.
-upstream_node(Shard, Timeout) ->
-    optvar:read(?optvar({?core_node, Shard}), Timeout).
+%% @doc Return name of the core node that can serve as the RPC target.
+%% It is the same node that serves the local replica, but this optvar
+%% is set before local replica goes up fully. WARNING: this `optvar'
+%% is set before local replica becomes consistent.
+-spec rpc_target(mria_rlog:shard(), timeout()) -> {ok, node()} | disconnected.
+rpc_target(Shard, Timeout) ->
+    case optvar:read(?optvar({?rpc_target, Shard}), Timeout) of
+        OK = {ok, _} ->
+            OK;
+        timeout ->
+            disconnected
+    end.
+
+-spec notify_rpc_target_up(mria_rlog:shard(), node()) -> ok.
+notify_rpc_target_up(Shard, Upstream) ->
+    do_notify_up(?rpc_target, Shard, Upstream).
+
+-spec notify_rpc_target_down(mria_rlog:shard()) -> ok.
+notify_rpc_target_down(Shard) ->
+    do_notify_down(?rpc_target, Shard).
+
+%% @doc Return name of the core node that is currently serving the
+%% downstream shard. In contrast with `rpc_target', this optvar is set
+%% when the shard reaches `normal' state and local reads become
+%% consistent.
+-spec upstream_node(mria_rlog:shard()) -> {ok, node()} | disconnected.
+upstream_node(Shard) ->
+    case upstream(Shard) of
+        {ok, Pid}    -> {ok, node(Pid)};
+        disconnected -> disconnected
+    end.
 
-%% @doc Return pid of the core node agent that serves us.
+%% @doc Return pid of the core node agent that serves us (when shard
+%% is in `normal' state).
 -spec upstream(mria_rlog:shard()) -> {ok, pid()} | disconnected.
 upstream(Shard) ->
     case optvar:peek(?optvar({?upstream_pid, Shard})) of
         {ok, Pid} -> {ok, Pid};
         undefined -> disconnected
     end.
 
-%% @deprecated Return a core node that _might_ be able to serve the
-%% specified shard. WARNING: use of this function leads to unbalanced
-%% load of the core nodes.
--spec get_core_node(mria_rlog:shard(), timeout()) -> {ok, node()} | timeout.
-get_core_node(Shard, Timeout) ->
+%% @doc WARNING: this optvar is used STRICTLY for interaction between
+%% `mria_lb' and `mria_replica' FSM. Its value is equal to core node
+%% that serves minimal number of replicants. As such, it must NOT be
+%% used for RPC targeting: all RPCs from the entire cluster will end
+%% up on a single node.
+-spec replica_get_core_node(mria_rlog:shard(), timeout()) -> {ok, node()} | timeout.
+replica_get_core_node(Shard, Timeout) ->
     optvar:read(?optvar({?core_node, Shard}), Timeout).
 
+-spec notify_core_node_up(mria_rlog:shard(), node()) -> ok.
+notify_core_node_up(Shard, Node) ->
+    do_notify_up(?core_node, Shard, Node).
+
+-spec notify_core_node_down(mria_rlog:shard()) -> ok.
+notify_core_node_down(Shard) ->
+    do_notify_down(?core_node, Shard).
+
 -spec notify_shard_up(mria_rlog:shard(), _AgentPid :: pid()) -> ok.
 notify_shard_up(Shard, Upstream) ->
     do_notify_up(?upstream_pid, Shard, Upstream).
@@ -120,14 +158,6 @@ notify_shard_down(Shard) ->
                    ?replicant_bootstrap_import
                   ]).
 
--spec notify_core_node_up(mria_rlog:shard(), node()) -> ok.
-notify_core_node_up(Shard, Node) ->
-    do_notify_up(?core_node, Shard, Node).
-
--spec notify_core_node_down(mria_rlog:shard()) -> ok.
-notify_core_node_down(Shard) ->
-    do_notify_down(?core_node, Shard).
-
 -spec notify_core_intercept_trans(mria_rlog:shard(), mria_rlog:seqno()) -> ok.
 notify_core_intercept_trans(Shard, SeqNo) ->
     set_stat(Shard, ?core_intercept, SeqNo).
@@ -171,10 +201,10 @@ replicants() ->
 
 -spec get_shard_lag(mria_rlog:shard()) -> non_neg_integer() | disconnected.
 get_shard_lag(Shard) ->
-    case {mria_config:role(), upstream_node(Shard, 0)} of
+    case {mria_config:role(), upstream_node(Shard)} of
         {core, _} ->
             0;
-        {replicant, timeout} ->
+        {replicant, disconnected} ->
             disconnected;
         {replicant, {ok, Upstream}} ->
             RemoteSeqNo = erpc:call(Upstream, ?MODULE, get_stat, [Shard, ?core_intercept], 1000),
@@ -242,7 +272,7 @@ get_shard_stats(Shard) ->
              , server_mql             => get_mql(Shard)
              };
         replicant ->
-            case upstream_node(Shard, 0) of
+            case upstream_node(Shard) of
                 {ok, Upstream} -> ok;
                 _ -> Upstream = undefined
             end,
diff --git a/test/concuerror_tests.erl b/test/concuerror_tests.erl
@@ -1,5 +1,5 @@
 %%--------------------------------------------------------------------
-%% Copyright (c) 2021-2023 EMQ Technologies Co., Ltd. All Rights Reserved.
+%% Copyright (c) 2021-2025 EMQ Technologies Co., Ltd. All Rights Reserved.
 %%
 %% Licensed under the Apache License, Version 2.0 (the "License");
 %% you may not use this file except in compliance with the License.
@@ -69,7 +69,7 @@ get_core_node_test() ->
         spawn(fun() ->
                       catch mria_status:notify_core_node_up(foo, Node)
               end),
-        ?assertMatch({ok, Node}, mria_status:get_core_node(foo, infinity)),
+        ?assertMatch({ok, Node}, mria_status:replica_get_core_node(foo, infinity)),
         ?assertMatch([], flush())
     after
         cleanup()
diff --git a/test/mria_SUITE.erl b/test/mria_SUITE.erl
@@ -1,5 +1,5 @@
 %%--------------------------------------------------------------------
-%% Copyright (c) 2019-2023 EMQ Technologies Co., Ltd. All Rights Reserved.
+%% Copyright (c) 2019-2025 EMQ Technologies Co., Ltd. All Rights Reserved.
 %%
 %% Licensed under the Apache License, Version 2.0 (the "License");
 %% you may not use this file except in compliance with the License.
@@ -365,36 +365,6 @@ t_transaction_on_replicant(_) ->
                mria_rlog_props:no_unexpected_events(Trace)
        end).
 
-%% EMQX-11006
-%%
-%% Test retry of aborted R/O transaction RPCs
-t_ro_trans_retry(_) ->
-    Cluster = mria_ct:cluster([core, replicant], mria_mnesia_test_util:common_env()),
-    ?check_trace(
-       #{timetrap => 10000},
-       try
-           Nodes = [N1, N2] = mria_ct:start_cluster(mria, Cluster),
-           mria_mnesia_test_util:wait_tables(Nodes),
-           %% Delay restart of mria on the core until the replicant
-           %% retries the transaction at least once:
-           ?force_ordering(
-              #{?snk_kind := mria_retry_rpc_to_core},
-              #{?snk_kind := "Starting mnesia", ?snk_meta := #{node := N1}}),
-           %% Restart mria on the core.
-           ?assertMatch(ok, rpc:call(N1, mria, stop, [])),
-           rpc:cast(N1, mria, start, []),
-           %% Issue a R/O transaction on the replicant:
-           TransFun = fun() -> 42 end,
-           ?assertMatch({atomic, 42},
-                        rpc:call(N2, mria, ro_transaction, [test_shard, TransFun]))
-       after
-           mria_ct:teardown_cluster(Cluster)
-       end,
-       fun(Trace) ->
-               %% Ensure that the replicant retried RPC:
-               ?assertMatch([_|_], ?of_kind(mria_retry_rpc_to_core, Trace))
-       end).
-
 t_sync_transaction_on_replicant(_) ->
     Cluster = mria_ct:cluster([core, replicant, replicant], mria_mnesia_test_util:common_env()),
     ?check_trace(
@@ -863,9 +833,6 @@ t_sum_verify(_) ->
     ?check_trace(
        #{timetrap => 30000},
        try
-           ?force_ordering( #{?snk_kind := verify_trans_step, n := N} when N =:= NTrans div 4
-                          , #{?snk_kind := state_change, to := bootstrap, shard := test_shard}
-                          ),
            ?force_ordering( #{?snk_kind := verify_trans_step, n := N} when N =:= 2 * NTrans div 4
                           , #{?snk_kind := state_change, to := local_replay, shard := test_shard}
                           ),
diff --git a/test/mria_lb_SUITE.erl b/test/mria_lb_SUITE.erl
@@ -1,5 +1,5 @@
 %%--------------------------------------------------------------------
-%% Copyright (c) 2019-2021, 2023-2024 EMQ Technologies Co., Ltd. All Rights Reserved.
+%% Copyright (c) 2019-2025 EMQ Technologies Co., Ltd. All Rights Reserved.
 %%
 %% Licensed under the Apache License, Version 2.0 (the "License");
 %% you may not use this file except in compliance with the License.
@@ -283,7 +283,7 @@ t_custom_compat_check(_Config) ->
            [_C1, _C2, C3, R1] = mria_ct:start_cluster(mria, Cluster),
            ?assertEqual({ok, C3},
                         erpc:call( R1
-                                 , mria_status, get_core_node, [?mria_meta_shard, infinity]
+                                 , mria_status, replica_get_core_node, [?mria_meta_shard, infinity]
                                  , infinity
                                  ))
        after
