phase 5: Node.call() gen_server protocol

Node wraps EPMD lookup + handshake + Channel into a single client
object. Constructor eagerly opens the dist connection; call() runs the
synchronous $gen_call protocol against a registered name on the peer:

  caller -> {'$gen_call', {FromPid, Ref}, Request}    (REG_SEND)
  server -> {Ref, Reply}                              (SEND)

Synthesized FromPid and a Node-lifetime Ref counter route replies back
to us; mismatched Ref or unexpected control op raises CallProtocolError.
Reply timeout raises CallTimeout (also covers Erlang's silent-drop case
when the registered name doesn't exist).

Tests against an erl peer running a $gen_call-aware loop:
  - {ping, X} -> {pong, X}
  - {add, A, B} -> {ok, A + B}
  - five sequential calls with monotonically increasing Refs
  - server error response surfaces as Python tuple
  - slow responder triggers CallTimeout
  - unknown registered name surfaces as CallTimeout
  - ref uniqueness across 100 synthesized refs

111 tests green across 10 consecutive runs, lint clean.

docs/ROADMAP.md

@@ -62,11 +62,22 @@ if we hit a peer that requires it.
     process: EPMD → handshake → REG_SEND → recv reply, payload matches
   - Survives an outbound tick before the reply
 
-## Phase 5 — gen_call convenience layer
+## Phase 5 — gen_call convenience layer ✅
 
-- `Node.call(name_or_pid, request, timeout=5.0)` → reply term
-- Wraps the `$gen_call` protocol used by `:gen_server`
-- Idempotency-key handling lives in the caller; we just pass the term
+- `Node(our_name, peer_name, cookie)` lazy-connects: EPMD lookup +
+  TCP connect + v6 handshake on construction
+- `Node.call(target_name, request, timeout=5.0)` implements
+  the `$gen_call` protocol:
+  ```
+  caller -> {'$gen_call', {FromPid, Ref}, Request}    (REG_SEND)
+  server -> {Ref, Reply}                              (SEND)
+  ```
+- Unique Ref per call, matched on receive — protocol violations
+  surface as `CallProtocolError`, timeouts as `CallTimeout`
+- Tests: `{ping, X} -> {pong, X}`, `{add, A, B} -> {ok, A+B}`,
+  sequential calls without crosstalk, timeout against a slow
+  responder, drop-on-the-floor against an unknown registered name,
+  Ref uniqueness across the Node lifetime
 
 ## Phase 6 — TLS dist
 

erldistpy/__init__.py

@@ -12,9 +12,12 @@ from erldistpy.etf import (
     encode,
 )
 from erldistpy.handshake import HandshakeError, HandshakeResult, handshake
+from erldistpy.node import CallProtocolError, CallTimeout, Node, NodeError
 
 __all__ = [
     "Atom",
+    "CallProtocolError",
+    "CallTimeout",
     "Channel",
     "ChannelError",
     "EpmdError",
@@ -22,6 +25,8 @@ __all__ = [
     "HandshakeError",
     "HandshakeResult",
     "IncomingMessage",
+    "Node",
+    "NodeError",
     "Pid",
     "Reference",
     "__version__",

erldistpy/node.py

@@ -0,0 +1,160 @@
+"""High-level node client — gen_server ``call`` over native dist.
+
+Wraps EPMD lookup + handshake + Channel into a single object that
+implements the OTP ``$gen_call`` protocol:
+
+    caller -> {'$gen_call', {FromPid, Ref}, Request}  (REG_SEND)
+    server -> {Ref, Reply}                            (SEND)
+
+Matching on ``Ref`` lets us correlate replies with calls and detect
+out-of-band messages (which would indicate a protocol violation in
+single-threaded sync use).
+
+Thread safety: synchronous, one in-flight call at a time. For concurrent
+callers wrap with an external lock or use one Node per worker.
+"""
+
+from __future__ import annotations
+
+import os
+import socket
+import struct
+
+from erldistpy.channel import (
+    OP_SEND,
+    OP_SEND_SENDER,
+    Channel,
+    ChannelError,
+)
+from erldistpy.epmd import lookup
+from erldistpy.etf import Atom, Pid, Reference
+from erldistpy.flags import DEFAULT_FLAGS
+from erldistpy.handshake import HandshakeResult, handshake
+
+
+class NodeError(RuntimeError):
+    """Generic node error."""
+
+
+class CallTimeout(NodeError):
+    """gen_server reply did not arrive within the call timeout."""
+
+
+class CallProtocolError(NodeError):
+    """Peer sent a message we couldn't route — out-of-band reply, wrong
+    Ref, unexpected op code. Likely indicates a server-side bug."""
+
+
+GEN_CALL_TAG = Atom("$gen_call")
+
+
+class Node:
+    """One client-side dist connection to one peer node."""
+
+    def __init__(
+        self,
+        *,
+        our_name: str,
+        peer_name: str,
+        cookie: str,
+        peer_host: str = "localhost",
+        connect_timeout: float = 5.0,
+    ):
+        if "@" not in our_name:
+            raise ValueError(f"our_name must look like 'name@host', got {our_name!r}")
+        self.our_name = our_name
+        self.peer_name = peer_name
+        self.peer_host = peer_host
+        self._cookie = cookie  # never logged
+        self._our_atom = Atom(our_name)
+
+        info = lookup(peer_name, host=peer_host, timeout=connect_timeout)
+        if info is None:
+            raise NodeError(f"peer node {peer_name!r} not registered with EPMD on {peer_host}")
+        sock = socket.create_connection((peer_host, info.port), timeout=connect_timeout)
+        try:
+            result: HandshakeResult = handshake(
+                sock,
+                our_name=our_name,
+                cookie=cookie,
+                our_flags=DEFAULT_FLAGS,
+                timeout=connect_timeout,
+            )
+        except Exception:
+            sock.close()
+            raise
+        self._sock = sock
+        self._channel = Channel(sock, recv_timeout=None)
+        self.our_creation = result.our_creation
+        self.peer_full_name = result.peer_name
+        self.peer_creation = result.peer_creation
+
+        # Unique Ref/Pid counters per Node lifetime
+        self._ref_counter = 0
+        self._our_pid = Pid(node=self._our_atom, id=1, serial=0, creation=self.our_creation)
+
+    # ----------------------------------------------------------------- public
+    def call(
+        self,
+        target_name: str,
+        request: object,
+        timeout: float = 5.0,
+    ) -> object:
+        """Synchronously call a registered gen_server on the peer node.
+
+        ``target_name`` is the registered atom name (e.g. ``"wallet_rpc"``).
+        ``request`` is any ETF-encodable term.
+
+        Returns the ``Reply`` half of the server's ``{Ref, Reply}`` tuple.
+        Raises :class:`CallTimeout` if the server doesn't reply in time.
+        Raises :class:`CallProtocolError` if the server replies with a
+        Ref we don't recognize or sends an unrelated message.
+        """
+        ref = self._next_ref()
+        payload = (GEN_CALL_TAG, (self._our_pid, ref), request)
+        self._channel.send_reg_send(self._our_pid, Atom(target_name), payload)
+
+        self._sock.settimeout(timeout)
+        try:
+            msg = self._channel.recv_message()
+        except ChannelError as e:
+            if "timed out" in str(e):
+                raise CallTimeout(f"no reply from {target_name!r} in {timeout}s") from e
+            raise
+        if msg is None:
+            raise CallProtocolError("peer closed the channel mid-call")
+        if msg.op not in (OP_SEND, OP_SEND_SENDER):
+            raise CallProtocolError(
+                f"expected SEND reply, got control op {msg.op}: {msg.control!r}"
+            )
+        if not isinstance(msg.payload, tuple) or len(msg.payload) != 2:
+            raise CallProtocolError(f"reply payload is not a {{Ref, Reply}} 2-tuple: {msg.payload!r}")
+        reply_ref, reply = msg.payload
+        if reply_ref != ref:
+            raise CallProtocolError(
+                f"reply ref mismatch: expected {ref!r}, got {reply_ref!r}"
+            )
+        return reply
+
+    def close(self) -> None:
+        self._channel.close()
+
+    def __enter__(self) -> Node:
+        return self
+
+    def __exit__(self, *_a) -> None:
+        self.close()
+
+    # ----------------------------------------------------------------- internals
+    def _next_ref(self) -> Reference:
+        self._ref_counter += 1
+        return Reference(
+            node=self._our_atom,
+            creation=self.our_creation,
+            ids=(self._ref_counter, 0, 0),
+        )
+
+
+def _random_creation() -> int:
+    n = struct.unpack(">I", os.urandom(4))[0]
+    return n if n else 1

tests/test_node.py

@@ -0,0 +1,203 @@
+"""High-level Node tests.
+
+Pure tests cover ref/pid generation. Live tests spawn an erl node with
+a process that implements the gen_server reply protocol by hand
+(no module compilation needed, just a receive on $gen_call) and
+exercise Node.call() end-to-end.
+"""
+
+from __future__ import annotations
+
+import os
+import shutil
+import socket
+import subprocess
+import time
+
+import pytest
+
+from erldistpy.etf import Atom, Reference
+from erldistpy.node import CallProtocolError, CallTimeout, Node, NodeError
+
+# Boot script registers `gen_target` which mimics gen_server's reply
+# protocol. Two operations supported:
+#   {ping, X}        -> {pong, X}
+#   {add, A, B}      -> {ok, A + B}
+#   slow             -> sleeps 2s before replying — used to force timeout
+#
+# The Y-combinator trick lets the receive loop be expressed inline.
+READY_FLAG = "/tmp/erldistpy_test_node_ready"
+ERL_GEN_BOOT = (
+    "Handle = fun "
+    "  ({ping, X}) -> {pong, X}; "
+    "  ({add, A, B}) -> {ok, A + B}; "
+    "  (slow) -> timer:sleep(2000), {ok, slow}; "
+    "  (Other) -> {error, {bad_request, Other}} "
+    "end, "
+    "Loop = fun(F) -> "
+    "  receive "
+    "    {'$gen_call', {From, Ref}, Request} -> "
+    "      From ! {Ref, Handle(Request)}, F(F); "
+    "    Other -> "
+    "      io:format(\"unexpected: ~p~n\", [Other]), F(F) "
+    "  end "
+    "end, "
+    "Pid = spawn(fun() -> Loop(Loop) end), "
+    "register(gen_target, Pid), "
+    f'file:write_file("{READY_FLAG}", "1"), '
+    "timer:sleep(infinity)."
+)
+
+
+def _epmd_running() -> bool:
+    try:
+        with socket.create_connection(("localhost", 4369), timeout=0.5):
+            return True
+    except OSError:
+        return False
+
+
+@pytest.fixture(scope="module")
+def gen_peer():
+    if not shutil.which("erl"):
+        pytest.skip("erl not installed")
+    if not _epmd_running():
+        pytest.skip("EPMD not running on localhost")
+
+    if os.path.exists(READY_FLAG):
+        os.remove(READY_FLAG)
+
+    sname = "erldistpy_gen"
+    cookie = "ERLDISTPY_GEN_COOKIE"
+    proc = subprocess.Popen(
+        ["erl", "-sname", sname, "-setcookie", cookie, "-noshell", "-eval", ERL_GEN_BOOT],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+
+    deadline = time.monotonic() + 6.0
+    from erldistpy.epmd import lookup
+    ready = False
+    while time.monotonic() < deadline:
+        time.sleep(0.1)
+        if lookup(sname, timeout=0.5) is None:
+            continue
+        if os.path.exists(READY_FLAG):
+            ready = True
+            break
+    if not ready:
+        proc.terminate()
+        proc.wait(timeout=2)
+        pytest.skip("erl gen_target did not become ready")
+
+    yield sname, cookie
+
+    proc.terminate()
+    try:
+        proc.wait(timeout=2)
+    except subprocess.TimeoutExpired:
+        proc.kill()
+    if os.path.exists(READY_FLAG):
+        os.remove(READY_FLAG)
+
+
+# --------------------------------------------------------------------------
+# Pure validation
+# --------------------------------------------------------------------------
+
+
+def test_node_requires_qualified_our_name():
+    with pytest.raises(ValueError, match="name@host"):
+        Node(our_name="bare", peer_name="x", cookie="X")
+
+
+def test_node_unknown_peer_raises():
+    with pytest.raises(NodeError, match="not registered"):
+        Node(
+            our_name="erldistpy_test@localhost",
+            peer_name="definitely_not_a_real_node_qzxw",
+            cookie="X",
+            connect_timeout=1.0,
+        )
+
+
+# --------------------------------------------------------------------------
+# Live call() against a process implementing $gen_call
+# --------------------------------------------------------------------------
+
+
+def _client_name() -> str:
+    return f"erldistpy_node@{socket.gethostname()}"
+
+
+def test_call_ping(gen_peer):
+    sname, cookie = gen_peer
+    with Node(our_name=_client_name(), peer_name=sname, cookie=cookie) as node:
+        reply = node.call("gen_target", (Atom("ping"), 42))
+        assert reply == (Atom("pong"), 42)
+
+
+def test_call_add(gen_peer):
+    sname, cookie = gen_peer
+    with Node(our_name=_client_name(), peer_name=sname, cookie=cookie) as node:
+        reply = node.call("gen_target", (Atom("add"), 7, 35))
+        assert reply == (Atom("ok"), 42)
+
+
+def test_call_multiple_in_sequence(gen_peer):
+    """Each call must increment Ref and not crosstalk with prior calls."""
+    sname, cookie = gen_peer
+    with Node(our_name=_client_name(), peer_name=sname, cookie=cookie) as node:
+        for i in range(5):
+            reply = node.call("gen_target", (Atom("add"), i, 1))
+            assert reply == (Atom("ok"), i + 1)
+
+
+def test_call_bad_request_surfaces_server_error(gen_peer):
+    sname, cookie = gen_peer
+    with Node(our_name=_client_name(), peer_name=sname, cookie=cookie) as node:
+        reply = node.call("gen_target", Atom("nonsense"))
+        # Server returns {error, {bad_request, Other}} per our handle fun
+        assert isinstance(reply, tuple)
+        assert reply[0] == Atom("error")
+
+
+def test_call_timeout(gen_peer):
+    """Server sleeps 2s; our timeout is 0.3s — must raise CallTimeout."""
+    sname, cookie = gen_peer
+    with (
+        Node(our_name=_client_name(), peer_name=sname, cookie=cookie) as node,
+        pytest.raises(CallTimeout),
+    ):
+        node.call("gen_target", Atom("slow"), timeout=0.3)
+
+
+def test_call_unknown_registered_name_times_out(gen_peer):
+    """Sending to a non-existent registered name is silently dropped by
+    Erlang — we must surface this as a timeout, not hang forever."""
+    sname, cookie = gen_peer
+    with (
+        Node(our_name=_client_name(), peer_name=sname, cookie=cookie) as node,
+        pytest.raises(CallTimeout),
+    ):
+        node.call("does_not_exist", (Atom("ping"), 1), timeout=0.4)
+
+
+def test_ref_counter_is_per_node_lifetime():
+    """Synthesized references must be unique within a Node instance."""
+    # We don't need a live peer; just exercise _next_ref via a dummy bypass.
+    # Construct a Node without going through __init__ to avoid the network.
+    node = Node.__new__(Node)
+    node._our_atom = Atom("x@y")
+    node.our_creation = 1
+    node._ref_counter = 0
+    refs = [node._next_ref() for _ in range(100)]
+    assert len(set((r.creation, r.ids) for r in refs)) == 100
+    assert all(isinstance(r, Reference) for r in refs)
+
+
+def test_protocol_error_path_uses_callprotocolerror():
+    """CallProtocolError exists as a distinct exception so callers can
+    distinguish 'server misbehaved' from 'server timed out'."""
+    assert issubclass(CallTimeout, NodeError)
+    assert issubclass(CallProtocolError, NodeError)