From ac4636e00c6cf94ef68ad9c47c3fcbc7aa46df49 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Tue, 16 Jun 2026 11:33:28 -0400 Subject: [PATCH] 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 | 19 +++- erldistpy/__init__.py | 5 ++ erldistpy/node.py | 160 +++++++++++++++++++++++++++++++++ tests/test_node.py | 203 ++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 383 insertions(+), 4 deletions(-) create mode 100644 erldistpy/node.py create mode 100644 tests/test_node.py diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index d9fe3d1..a61650f 100644 --- a/docs/ROADMAP.md +++ b/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 diff --git a/erldistpy/__init__.py b/erldistpy/__init__.py index 07a6155..17812f7 100644 --- a/erldistpy/__init__.py +++ b/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__", diff --git a/erldistpy/node.py b/erldistpy/node.py new file mode 100644 index 0000000..ef32cb5 --- /dev/null +++ b/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 diff --git a/tests/test_node.py b/tests/test_node.py new file mode 100644 index 0000000..3b2426e --- /dev/null +++ b/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)