fix(rpc): Introduce a separate optvar for RPC targeting

Author: ieQu1

src/mria.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.
@@ -595,7 +595,7 @@ should_retry_rpc(_) ->
 find_upstream_node(Shard) ->
     ?tp_span(find_upstream_node, #{shard => Shard},
              begin
-                 {ok, Node} = mria_status:get_core_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:get_core_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, _} -> [];

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,

src/mria_status.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.
@@ -24,8 +24,9 @@
 
 %% 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/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,31 +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. Note the difference in behavior as compared with
-%% `get_core_node'. Returns `disconnected' if the local replica of the
-%% shard is down.
+%% @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.
 
-%% @doc Return a core node that _might_ be able to serve the specified
-%% shard.
--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).
@@ -122,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).

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()

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.
@@ -863,9 +863,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}
                           ),

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