more fixes for disteral to work correctly

Signed-off-by: Alejandro M. Ramallo <alejandro.ramallo@leapsight.com>

Author: aramallo

src/partisan.erl

@@ -182,6 +182,7 @@
 -export([nodes/0]).
 -export([nodes/1]).
 -export([nodestring/0]).
+-export([remote_ref_to_disterl/1]).
 -export([self/1]).
 
 
@@ -429,8 +430,18 @@ monitor(process, Term, Opts) when erlang:is_pid(Term) orelse is_atom(Term) ->
     erlang:monitor(process, Term, to_erl_monitor_opts(Opts));
 
 monitor(process, RemoteRef, Opts) ->
-    %% eqwalizer:ignore RemoteRef
-    partisan_monitor:monitor(RemoteRef, Opts);
+    %% When `connect_disterl' is true the test harness explicitly opted into
+    %% disterl-only routing — decode the remote-ref and use `erlang:monitor'
+    %% so DOWN messages fire promptly when the peer drops. Otherwise use the
+    %% partisan transport's own monitor (the default behaviour).
+    case partisan_config:get(connect_disterl, false)
+            andalso remote_ref_to_disterl(RemoteRef) of
+        {ok, {Name, _Node} = NN} when is_atom(Name) ->
+            erlang:monitor(process, NN, to_erl_monitor_opts(Opts));
+        _ ->
+            %% eqwalizer:ignore RemoteRef
+            partisan_monitor:monitor(RemoteRef, Opts)
+    end;
 
 monitor(Type, Term, Opts) when Type == port orelse Type == time_offset ->
     erlang:monitor(Type, Term, to_erl_monitor_opts(Opts)).
@@ -1762,4 +1773,39 @@ process_exit_reason(_) ->
     noproc.
 
 
+%% @private
+%% True if the target lives on a node we already have a disterl connection
+%% to (or is local). Caller uses this to prefer disterl-based monitoring,
+%% which detects noconnection promptly.
+is_disterl_connected(Pid) when erlang:is_pid(Pid) ->
+    Node = erlang:node(Pid),
+    Node =:= erlang:node()
+        orelse lists:member(Node, erlang:nodes());
+is_disterl_connected({Name, Node}) when is_atom(Name), is_atom(Node) ->
+    Node =:= erlang:node()
+        orelse lists:member(Node, erlang:nodes());
+is_disterl_connected(_) ->
+    false.
+
+%% @private
+%% When `connect_disterl' is set, convert a partisan_remote_ref to a
+%% disterl-usable form: a foreign pid (via `list_to_pid/1') for encoded pids,
+%% or `{Name, Node}' for encoded names. Returns `error' for refs we don't
+%% know how to convert (the caller should fall back to partisan_monitor).
+remote_ref_to_disterl(Ref) ->
+    try
+        Node = partisan_remote_ref:node(Ref),
+        case partisan_remote_ref:target(Ref) of
+            {encoded_pid, Str} ->
+                {ok, list_to_pid(Str)};
+            {encoded_name, Str} ->
+                {ok, {list_to_existing_atom(Str), Node}};
+            _ ->
+                error
+        end
+    catch
+        _:_ -> error
+    end.
+
+
 

src/partisan_otp_patches.erl

@@ -184,7 +184,11 @@ do_send_request_source() ->
     "            do_send_request(LocalProcess, Label, Request);\n"
     "        false ->\n"
     "            Opts = partisan_gen:get_opts(),\n"
-    "            Mref = partisan:monitor(process, Process, Opts),\n"
+    "            %% Same `{alias, demonitor}' requirement as the {Name, Node}\n"
+    "            %% clause above — the receiver replies via `[alias|Mref]'.\n"
+    "            Mref = partisan:monitor(\n"
+    "                process, Process, [{alias, demonitor} | Opts]\n"
+    "            ),\n"
     "            Message = {Label, {partisan:self(), [alias|Mref]}, Request},\n"
     "            partisan:forward_message(Process, Message, Opts),\n"
     "            Mref\n"

src/partisan_pluggable_peer_service_manager.erl

@@ -479,9 +479,26 @@ forward_message(RemoteRef, Message, Opts) ->
         orelse error(badarg),
 
     Node = partisan_remote_ref:node(RemoteRef),
-    Target = partisan_remote_ref:target(RemoteRef),
-
-    forward_message(Node, Target, Message, Opts).
+    %% When `connect_disterl' is true, prefer disterl: send directly to the
+    %% native pid/name. Otherwise route through the encoded target so the
+    %% partisan transport handles it.
+    case partisan_config:get(connect_disterl, false) of
+        true ->
+            case partisan:remote_ref_to_disterl(RemoteRef) of
+                {ok, Pid} when is_pid(Pid) ->
+                    _ = (catch erlang:send(Pid, Message, [noconnect])),
+                    ok;
+                {ok, {Name, _Node} = NN} when is_atom(Name) ->
+                    _ = (catch erlang:send(NN, Message, [noconnect])),
+                    ok;
+                _ ->
+                    Target = partisan_remote_ref:target(RemoteRef),
+                    forward_message(Node, Target, Message, Opts)
+            end;
+        false ->
+            Target = partisan_remote_ref:target(RemoteRef),
+            forward_message(Node, Target, Message, Opts)
+    end.
 
 
 %% -----------------------------------------------------------------------------

test/partisan_gen_server_SUITE.erl

@@ -952,7 +952,8 @@ hibernate(Config) when is_list(Config) ->
         erlang:process_info(Pid, current_function)),
     receive
     {result,R} ->
-        {current_function,{erlang,hibernate,3}} = R
+        {current_function, MFA0} = R,
+        true = is_hibernating_mfa(MFA0)
     end,
 
     true = partisan_gen_server:call(my_test_name_hibernate, hibernate),
@@ -1036,10 +1037,10 @@ is_in_erlang_hibernate_1(0, Pid) ->
     ct:fail(not_in_erlang_hibernate_3);
 is_in_erlang_hibernate_1(N, Pid) ->
     {current_function,MFA} = erlang:process_info(Pid, current_function),
-    case MFA of
-    {erlang,hibernate,_Arity} ->
+    case is_hibernating_mfa(MFA) of
+    true ->
         ok;
-    _ ->
+    false ->
         receive after 10 -> ok end,
         is_in_erlang_hibernate_1(N-1, Pid)
     end.
@@ -1053,14 +1054,22 @@ is_not_in_erlang_hibernate_1(0, Pid) ->
     ct:fail(not_in_erlang_hibernate_3);
 is_not_in_erlang_hibernate_1(N, Pid) ->
     {current_function,MFA} = erlang:process_info(Pid, current_function),
-    case MFA of
-        {erlang,hibernate,_Arity} ->
+    case is_hibernating_mfa(MFA) of
+        true ->
             receive after 10 -> ok end,
             is_not_in_erlang_hibernate_1(N-1, Pid);
-        _ ->
+        false ->
             ok
     end.
 
+%% Recognize a hibernating-process current_function. OTP 28's gen_server
+%% hibernation parks the process inside its own `loop_hibernate/4' rather than
+%% inside `erlang:hibernate/_' (which the test was originally written for).
+is_hibernating_mfa({erlang, hibernate, _}) -> true;
+is_hibernating_mfa({partisan_gen_server, loop_hibernate, _}) -> true;
+is_hibernating_mfa({gen_server, loop_hibernate, _}) -> true;
+is_hibernating_mfa(_) -> false.
+
 %% --------------------------------------
 %% Test partisan_gen_server:abcast and handle_cast.
 %% Test all different return values from
@@ -1609,20 +1618,21 @@ undef_terminate2(Config) when is_list(Config) ->
     ok = partisan_gen_server:stop(Server, {error, test}, infinity),
     ok = verify_down_reason(MRef, Server, {error, test}).
 
-%% Start should return an undef error if init isn't implemented
+%% Start should return an undef error if init isn't implemented.
+%% OTP 28 catches the undef inside start_link and returns the error
+%% synchronously without ever fully spawning the server, so no `{'EXIT', _, _}'
+%% message is delivered to the trap_exit caller.
 undef_init(_Config) ->
     {error, {undef, [{oc_init_server, init, [_], _}|_]}} =
         partisan_gen_server:start(oc_init_server, [], []),
     process_flag(trap_exit, true),
     {error, {undef, [{oc_init_server, init, [_], _}|_]}} =
         (catch partisan_gen_server:start_link(oc_init_server, [], [])),
     receive
-        {'EXIT', Server,
-         {undef, [{oc_init_server, init, [_], _}|_]}} when is_pid(Server) ->
-            partisan:is_pid(Server) orelse ct:fail({bad_pid, Server}),
-            ok
-    after 1000 ->
-        ct:fail(expected_exit_msg)
+        Msg ->
+            ct:fail({unexpected_msg, Msg})
+    after 500 ->
+        ok
     end.
 
 %% The upgrade should fail if code_change is expected in the callback module

test/partisan_gen_statem_SUITE.erl

@@ -1644,10 +1644,10 @@ wait_erlang_hibernate_1(0, Pid) ->
     ct:fail(should_be_in_erlang_hibernate_3);
 wait_erlang_hibernate_1(N, Pid) ->
     {current_function,MFA} = erlang:process_info(Pid, current_function),
-    case MFA of
-    {erlang,hibernate,_Arity} ->
+    case is_hibernating_mfa(MFA) of
+    true ->
         ok;
-    _ ->
+    false ->
         receive after 10 -> ok end,
         wait_erlang_hibernate_1(N-1, Pid)
     end.
@@ -1660,14 +1660,22 @@ is_not_in_erlang_hibernate_1(0, _Pid) ->
     ct:fail(should_not_be_in_erlang_hibernate_3);
 is_not_in_erlang_hibernate_1(N, Pid) ->
     {current_function,MFA} = erlang:process_info(Pid, current_function),
-    case MFA of
-    {erlang,hibernate,_Arity} ->
+    case is_hibernating_mfa(MFA) of
+    true ->
         receive after 10 -> ok end,
         is_not_in_erlang_hibernate_1(N-1, Pid);
-    _ ->
+    false ->
         ok
     end.
 
+%% OTP 28's gen_statem (and the rewritten partisan_gen_statem) parks a
+%% hibernating process inside its own `loop_hibernate/4' rather than inside
+%% `erlang:hibernate/_'. Match either form.
+is_hibernating_mfa({erlang, hibernate, _}) -> true;
+is_hibernating_mfa({partisan_gen_statem, loop_hibernate, _}) -> true;
+is_hibernating_mfa({gen_statem, loop_hibernate, _}) -> true;
+is_hibernating_mfa(_) -> false.
+
 
 enter_loop(_Config) ->
     OldFlag = process_flag(trap_exit, true),

test/partisan_support.erl

@@ -597,6 +597,12 @@ start_ct_node(Name, Opts) ->
     end.
 
 stop_ct_node(Node) ->
+    %% Ask the peer to leave the partisan cluster gracefully so the local
+    %% partisan_monitor sees the disconnect (and fires `noconnection' DOWN
+    %% messages on outstanding monitors). Without this, killing the peer via
+    %% `peer:stop' alone makes the runner wait for partisan's heartbeat to
+    %% time out, which exceeds CT's per-test budget.
+    _ = (catch rpc:call(Node, partisan_peer_service, leave, [], 2000)),
     Key = {?MODULE, peer, Node},
     case persistent_term:get(Key, undefined) of
         undefined ->