Support PID alias in the send_msg effect

src/ra_server_proc.erl

@@ -1579,26 +1579,46 @@ handle_effect(_, {next_event, Evt}, EvtType, State, Actions) ->
 handle_effect(_, {next_event, _, _} = Next, _, State, Actions) ->
     {State, [Next | Actions]};
 handle_effect(leader, {send_msg, To, Msg}, _, State, Actions) ->
-    %% default is to send without any wrapping
-    ToNode = get_node(To),
-    ?ASYNC_DIST(ToNode, _ = send(To, Msg, State#state.conf)),
+    case is_reference(To) of
+        false ->
+            %% default is to send without any wrapping
+            ToNode = get_node(To),
+            ?ASYNC_DIST(ToNode, _ = send(To, Msg, State#state.conf));
+        true ->
+            %% When a PID alias is passed, send from the leader. We have no way
+            %% to determine what the PID and thus its node are.
+            _ = send(To, Msg, State#state.conf)
+    end,
     {State, Actions};
 handle_effect(RaftState, {send_msg, To, _Msg, Options} = Eff,
               _, State, Actions) ->
-    ToNode = get_node(To),
-    case has_local_opt(Options) of
-        true ->
-            case can_execute_locally(RaftState, ToNode, State) of
+    case is_reference(To) of
+        false ->
+            ToNode = get_node(To),
+            case has_local_opt(Options) of
                 true ->
+                    case can_execute_locally(RaftState, ToNode, State) of
+                        true ->
+                            ?ASYNC_DIST(ToNode, send_msg(Eff, State));
+                        false ->
+                            ok
+                    end;
+                false when RaftState == leader ->
+                    %% the effect got here so we can execute
                     ?ASYNC_DIST(ToNode, send_msg(Eff, State));
                 false ->
                     ok
             end;
-        false when RaftState == leader ->
-            %% the effect got here so we can execute
-            ?ASYNC_DIST(ToNode, send_msg(Eff, State));
-        false ->
-            ok
+        true ->
+            %% When a PID alias is passed, send from the leader. We have no way
+            %% to determine what the PID and thus its node are. Therefore, we
+            %% can't honor the `local' option.
+            case RaftState of
+                leader ->
+                    send_msg(Eff, State);
+                _ ->
+                    ok
+            end
     end,
     {State, Actions};
 handle_effect(leader, {append, Cmd}, _EvtType, State, Actions) ->

test/ra_machine_int_SUITE.erl

@@ -29,6 +29,7 @@ all() ->
 all_tests() ->
     [
      send_msg_without_options,
+     send_msg_with_alias,
      send_msg_with_ra_event_option,
      send_msg_with_cast_option,
      send_msg_with_ra_event_and_cast_options,
@@ -120,6 +121,25 @@ send_msg_without_options(Config) ->
     end,
     ok.
 
+send_msg_with_alias(Config) ->
+    Mod = ?config(modname, Config),
+    meck:new(Mod, [non_strict]),
+    meck:expect(Mod, init, fun (_) -> the_state end),
+    meck:expect(Mod, apply, fun (_, {echo, Alias, Msg}, State) ->
+                                    {State, ok, {send_msg, Alias, Msg}}
+                            end),
+    ClusterName = ?config(cluster_name, Config),
+    ServerId = ?config(server_id, Config),
+    ok = start_cluster(ClusterName, {module, Mod, #{}}, [ServerId]),
+    Alias = erlang:alias([reply]),
+    {ok, ok, _} = ra:process_command(ServerId, {echo, Alias, ?FUNCTION_NAME}),
+    receive ?FUNCTION_NAME -> ok
+    after 250 ->
+              flush(),
+              exit(receive_msg_timeout)
+    end,
+    ok.
+
 send_msg_with_ra_event_option(Config) ->
     Mod = ?config(modname, Config),
     meck:new(Mod, [non_strict]),