diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 47b2c44..d9fe3d1 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -46,13 +46,21 @@ EPMD (Erlang Port Mapper Daemon) maps node names to TCP ports. Newer SHA-256 digest (DFLAG_MANDATORY_25_DIGEST) deferred — landed only if we hit a peer that requires it. -## Phase 4 — Distribution channel +## Phase 4 — Distribution channel ✅ -- After handshake, the socket carries control + payload messages framed - by a 4-byte length prefix -- Send `SEND_TT` / `REG_SEND` for outgoing messages -- Receive replies, route by ref -- Tick loop for keepalive (60s default per OTP) +- `Channel` wraps the post-handshake socket: 4-byte length frames, + pass-through byte (`'p'`), control tuple + optional payload term +- `send_reg_send(from_pid, to_name, payload)` helper for the common case +- `recv_message()` skips ticks transparently; `recv_raw()` exposes them + for callers that need tick awareness +- Tick keepalive via `send_tick()` — caller drives the timer for now + (background ticker lands in Phase 7 alongside unfeed integration) +- Tests: + - Pure encode/decode round-trips + - Socket-pair tests for send/recv framing, tick handling + - Live end-to-end against an `erl` node with a registered echo + process: EPMD → handshake → REG_SEND → recv reply, payload matches + - Survives an outbound tick before the reply ## Phase 5 — gen_call convenience layer diff --git a/erldistpy/__init__.py b/erldistpy/__init__.py index 221ab02..07a6155 100644 --- a/erldistpy/__init__.py +++ b/erldistpy/__init__.py @@ -2,6 +2,7 @@ __version__ = "0.0.1" +from erldistpy.channel import Channel, ChannelError, IncomingMessage from erldistpy.epmd import EpmdError, EpmdInfo, lookup from erldistpy.etf import ( Atom, @@ -14,10 +15,13 @@ from erldistpy.handshake import HandshakeError, HandshakeResult, handshake __all__ = [ "Atom", + "Channel", + "ChannelError", "EpmdError", "EpmdInfo", "HandshakeError", "HandshakeResult", + "IncomingMessage", "Pid", "Reference", "__version__", diff --git a/erldistpy/channel.py b/erldistpy/channel.py new file mode 100644 index 0000000..d0744b2 --- /dev/null +++ b/erldistpy/channel.py @@ -0,0 +1,187 @@ +"""Post-handshake distribution data channel. + +Once :func:`erldistpy.handshake.handshake` returns, the socket carries +4-byte length-prefixed distribution messages. Each non-empty message +starts with the pass-through byte ``'p'`` (112) followed by a +ControlMessage (an ETF tuple identifying the operation) and an optional +Payload (the actual term being sent). + +Empty 4-byte messages are net_tick keepalive frames. The peer expects +us to send one roughly every ``net_ticktime / 4`` seconds (default 15s +for the standard 60s ticktime) or it will close the connection. Tick +handling is exposed via :meth:`Channel.send_tick` — the caller drives +the timer (or a daemon thread does, see the unfeed integration phase). + +Spec: https://www.erlang.org/doc/apps/erts/erl_dist_protocol.html#protocol-between-connected-nodes +""" + +from __future__ import annotations + +import contextlib +import socket +import struct +from dataclasses import dataclass + +from erldistpy.etf import Atom, ETFError, Pid, decode_term, encode + +PASS_THROUGH = 0x70 # 'p' + +# Control operation codes (the first element of the ControlMessage tuple). +OP_LINK = 1 +OP_SEND = 2 +OP_EXIT = 3 +OP_UNLINK = 4 +OP_REG_SEND = 6 +OP_GROUP_LEADER = 7 +OP_EXIT2 = 8 +OP_SEND_TT = 12 +OP_EXIT_TT = 13 +OP_REG_SEND_TT = 16 +OP_EXIT2_TT = 18 +OP_MONITOR_P = 19 +OP_DEMONITOR_P = 20 +OP_MONITOR_P_EXIT = 21 +OP_SEND_SENDER = 22 +OP_SEND_SENDER_TT = 23 +OP_UNLINK_ID = 35 +OP_UNLINK_ID_ACK = 36 + + +class ChannelError(RuntimeError): + """Channel I/O or framing error.""" + + +@dataclass(frozen=True) +class IncomingMessage: + """A decoded distribution message.""" + + control: tuple + payload: object | None # None for control-only ops (LINK, MONITOR, etc.) + + @property + def op(self) -> int: + return self.control[0] if self.control else -1 + + +class Channel: + """Synchronous wrapper around a post-handshake distribution socket. + + Not thread-safe; serialize external access if multiple writers exist. + """ + + def __init__(self, sock: socket.socket, *, recv_timeout: float | None = 5.0): + self._sock = sock + if recv_timeout is not None: + self._sock.settimeout(recv_timeout) + + # ------------------------------------------------------------------ raw + def send_raw(self, body: bytes) -> None: + """Send a 4-byte length-prefixed frame. Empty body = net_tick.""" + try: + self._sock.sendall(struct.pack(">I", len(body)) + body) + except OSError as e: + raise ChannelError(f"send failed: {e}") from e + + def recv_raw(self) -> bytes: + """Read the next 4-byte length-prefixed frame body. + + Returns ``b""`` for net_tick keepalive frames. Raises on EOF. + """ + hdr = _recv_exact(self._sock, 4) + (length,) = struct.unpack(">I", hdr) + if length == 0: + return b"" + return _recv_exact(self._sock, length) + + def send_tick(self) -> None: + """Send a net_tick keepalive frame.""" + self.send_raw(b"") + + def close(self) -> None: + with contextlib.suppress(OSError): + self._sock.close() + + def __enter__(self) -> Channel: + return self + + def __exit__(self, *_a) -> None: + self.close() + + # ------------------------------------------------------------------ structured + def send_control(self, control: tuple, payload: object | None = None) -> None: + """Encode and send a distribution message.""" + self.send_raw(_encode_message(control, payload)) + + def recv_message(self) -> IncomingMessage | None: + """Block until the next non-tick message arrives. + + Returns ``None`` when the peer closes cleanly. Tick frames are + consumed silently — callers wanting tick awareness should drop + to :meth:`recv_raw`. + """ + while True: + body = self.recv_raw() + if body == b"": + # tick — keep waiting + continue + return _decode_message(body) + + def send_reg_send(self, from_pid: Pid, to_name: Atom, payload: object) -> None: + """Send ``payload`` to a registered process ``to_name`` on the peer. + + The peer routes replies back to ``from_pid`` over this channel. + ``from_pid`` may be synthetic — what matters is that it embeds our + node atom so the peer's `! / 2` reply gets routed to us. + """ + cookie = Atom("") # by spec, the cookie field is always '' + control = (OP_REG_SEND, from_pid, cookie, to_name) + self.send_control(control, payload) + + +# --------------------------------------------------------------------------- +# Pure helpers (exposed for tests) +# --------------------------------------------------------------------------- + + +def _encode_message(control: tuple, payload: object | None) -> bytes: + """Build a distribution message body (no length prefix).""" + body = bytes([PASS_THROUGH]) + encode(control) + if payload is not None: + body += encode(payload) + return body + + +def _decode_message(body: bytes) -> IncomingMessage: + if not body or body[0] != PASS_THROUGH: + raise ChannelError( + f"expected pass-through byte 0x70, got {body[:1].hex() or 'empty'}" + ) + try: + control, off = decode_term(body, 1) + if not isinstance(control, tuple): + raise ChannelError(f"control message is not a tuple: {type(control).__name__}") + payload: object | None + if off == len(body): + payload = None + else: + payload, off = decode_term(body, off) + if off != len(body): + raise ChannelError(f"trailing bytes after payload: {len(body) - off} unread") + except ETFError as e: + raise ChannelError(f"ETF decode failed: {e}") from e + return IncomingMessage(control=control, payload=payload) + + +def _recv_exact(sock: socket.socket, n: int) -> bytes: + chunks: list[bytes] = [] + remaining = n + while remaining > 0: + try: + chunk = sock.recv(remaining) + except OSError as e: + raise ChannelError(f"recv failed after {n - remaining}/{n}: {e}") from e + if not chunk: + raise ChannelError(f"peer closed after {n - remaining}/{n} bytes") + chunks.append(chunk) + remaining -= len(chunk) + return b"".join(chunks) diff --git a/erldistpy/etf.py b/erldistpy/etf.py index 65ecc78..9b502d8 100644 --- a/erldistpy/etf.py +++ b/erldistpy/etf.py @@ -168,14 +168,24 @@ def _encode_ref(r: Reference) -> bytes: def decode(data: bytes) -> object: """Decode ETF bytes (with magic version byte) into a Python value.""" - if not data or data[0] != MAGIC: - raise ETFError(f"bad magic: {data[:1]!r}") - term, off = _decode(data, 1) + term, off = decode_term(data, 0) if off != len(data): raise ETFError(f"trailing bytes after term: {len(data) - off} unread") return term +def decode_term(data: bytes, offset: int = 0) -> tuple[object, int]: + """Decode one ETF term starting at ``offset``. Returns ``(term, new_offset)``. + + Used by callers that need to read multiple terms back-to-back (e.g. + the distribution channel's pass-through messages, which carry a + control term and optionally a payload term in the same frame). + """ + if offset >= len(data) or data[offset] != MAGIC: + raise ETFError(f"bad magic at offset {offset}: {data[offset:offset + 1]!r}") + return _decode(data, offset + 1) + + def _decode(data: bytes, off: int) -> tuple[object, int]: if off >= len(data): raise ETFError("unexpected end of input") diff --git a/tests/test_channel.py b/tests/test_channel.py new file mode 100644 index 0000000..32ee601 --- /dev/null +++ b/tests/test_channel.py @@ -0,0 +1,283 @@ +"""Distribution channel tests. + +Unit tests cover the encode/decode of pass-through messages. +Live tests spawn an Erlang node with a registered echo process, +complete the full EPMD lookup + handshake + REG_SEND + recv cycle, +and verify the round-tripped payload matches. +""" + +from __future__ import annotations + +import shutil +import socket +import struct +import subprocess +import time + +import pytest + +from erldistpy.channel import ( + OP_REG_SEND, + OP_SEND, + OP_SEND_SENDER, + PASS_THROUGH, + Channel, + ChannelError, + IncomingMessage, + _decode_message, + _encode_message, +) +from erldistpy.epmd import lookup +from erldistpy.etf import Atom, Pid, encode +from erldistpy.handshake import handshake + +# -------------------------------------------------------------------------- +# Pure encode/decode +# -------------------------------------------------------------------------- + + +def test_encode_message_control_only(): + """LINK/UNLINK/etc. have no payload — just a control tuple.""" + control = (1, Pid(Atom("a@b"), 0, 0, 1), Pid(Atom("c@d"), 0, 0, 1)) + body = _encode_message(control, None) + assert body[0] == PASS_THROUGH + # Decode round-trip + msg = _decode_message(body) + assert msg.control == control + assert msg.payload is None + + +def test_encode_message_with_payload(): + control = (OP_REG_SEND, Pid(Atom("client@h"), 1, 0, 5), Atom(""), Atom("echo")) + payload = (Atom("hello"), b"world") + body = _encode_message(control, payload) + msg = _decode_message(body) + assert msg.control == control + assert msg.payload == payload + + +def test_decode_rejects_wrong_first_byte(): + with pytest.raises(ChannelError, match="pass-through byte"): + _decode_message(b"\x00" + encode((1,))) + + +def test_decode_rejects_non_tuple_control(): + body = bytes([PASS_THROUGH]) + encode(42) + with pytest.raises(ChannelError, match="not a tuple"): + _decode_message(body) + + +def test_decode_rejects_trailing_garbage(): + body = bytes([PASS_THROUGH]) + encode((1,)) + b"\xFF" + with pytest.raises(ChannelError): + _decode_message(body) + + +def test_incoming_message_op_property(): + msg = IncomingMessage(control=(OP_SEND, Atom(""), Pid(Atom("x@y"), 0, 0, 1)), payload=42) + assert msg.op == OP_SEND + + +# -------------------------------------------------------------------------- +# Raw socket framing (loopback with a paired socket) +# -------------------------------------------------------------------------- + + +def _socket_pair() -> tuple[socket.socket, socket.socket]: + return socket.socketpair() + + +def test_send_raw_recv_raw_round_trip(): + a, b = _socket_pair() + ch_a = Channel(a) + ch_b = Channel(b) + ch_a.send_raw(b"hello") + assert ch_b.recv_raw() == b"hello" + + +def test_tick_round_trip(): + a, b = _socket_pair() + ch_a = Channel(a) + ch_b = Channel(b) + ch_a.send_tick() + assert ch_b.recv_raw() == b"" + + +def test_send_control_recv_message(): + a, b = _socket_pair() + ch_a = Channel(a) + ch_b = Channel(b) + control = (OP_REG_SEND, Pid(Atom("a@h"), 1, 0, 7), Atom(""), Atom("echo")) + payload = (Atom("greet"), b"hi") + ch_a.send_control(control, payload) + msg = ch_b.recv_message() + assert msg is not None + assert msg.control == control + assert msg.payload == payload + + +def test_recv_message_skips_ticks(): + a, b = _socket_pair() + ch_a = Channel(a) + ch_b = Channel(b) + ch_a.send_tick() + ch_a.send_tick() + control = (OP_SEND, Atom(""), Pid(Atom("x@y"), 0, 0, 1)) + ch_a.send_control(control, b"payload") + msg = ch_b.recv_message() + assert msg is not None + assert msg.control == control + assert msg.payload == b"payload" + + +def test_send_reg_send_helper(): + a, b = _socket_pair() + ch_a = Channel(a) + ch_b = Channel(b) + from_pid = Pid(Atom("client@h"), 42, 0, 9) + ch_a.send_reg_send(from_pid, Atom("kernel"), (Atom("ping"), 1)) + msg = ch_b.recv_message() + assert msg is not None + assert msg.control == (OP_REG_SEND, from_pid, Atom(""), Atom("kernel")) + assert msg.payload == (Atom("ping"), 1) + + +def test_recv_after_peer_closes_raises(): + a, b = _socket_pair() + ch_a = Channel(a) + Channel(b).close() + with pytest.raises(ChannelError, match="peer closed"): + # The 4-byte length header will fail because peer closed + ch_a.recv_raw() + + +def test_recv_handles_truncated_length_header(): + """If only 2 of 4 length bytes arrive then peer dies, we must error + instead of hanging or returning garbage.""" + a, b = _socket_pair() + a.sendall(b"\x00\x00") # only 2 of 4 header bytes + a.close() + ch_b = Channel(b) + with pytest.raises(ChannelError, match="peer closed"): + ch_b.recv_raw() + + +# -------------------------------------------------------------------------- +# Live end-to-end — EPMD + handshake + REG_SEND + recv reply +# -------------------------------------------------------------------------- + + +# Spawn an erl node with a tiny echo process registered as `echo`. +# The Y-combinator trick lets us write a recursive fun inside -eval. +# A flag file is written after registration so the fixture can wait for +# the process to actually be reachable (EPMD registers the node *before* +# our -eval runs, so polling EPMD alone is racy). +READY_FLAG = "/tmp/erldistpy_test_channel_ready" +ERL_ECHO_BOOT = ( + "EchoLoop = fun(F) -> " + "receive {From, Msg} -> From ! {echoed, Msg}, F(F) end " + "end, " + "Pid = spawn(fun() -> EchoLoop(EchoLoop) end), " + "register(echo, 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 echo_node(): + if not shutil.which("erl"): + pytest.skip("erl not installed") + if not _epmd_running(): + pytest.skip("EPMD not running on localhost") + + import os + # Clear stale flag from a previous run + if os.path.exists(READY_FLAG): + os.remove(READY_FLAG) + + sname = "erldistpy_echo" + cookie = "ERLDISTPY_CH_COOKIE" + proc = subprocess.Popen( + ["erl", "-sname", sname, "-setcookie", cookie, "-noshell", "-eval", ERL_ECHO_BOOT], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + # Wait for BOTH: node registered with EPMD AND echo process up + deadline = time.monotonic() + 6.0 + 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 echo process did not become ready in time") + + 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) + + +def test_live_reg_send_echo(echo_node): + sname, cookie = echo_node + info = lookup(sname) + assert info is not None + sock = socket.create_connection(("localhost", info.port), timeout=2) + try: + client_name = f"erldistpy_ch@{socket.gethostname()}" + result = handshake(sock, our_name=client_name, cookie=cookie) + ch = Channel(sock, recv_timeout=3.0) + from_pid = Pid(Atom(client_name), 1, 0, result.our_creation) + message_body = b"hello over native dist" + ch.send_reg_send(from_pid, Atom("echo"), (from_pid, message_body)) + msg = ch.recv_message() + assert msg is not None + # Peer may use OP_SEND (legacy) or OP_SEND_SENDER (if it offered DFLAG_SEND_SENDER) + assert msg.op in (OP_SEND, OP_SEND_SENDER) + # Payload echoes our tuple form + assert msg.payload == (Atom("echoed"), message_body) + finally: + sock.close() + + +def test_live_reg_send_survives_tick(echo_node): + """An inbound tick frame before the reply must not break recv_message().""" + sname, cookie = echo_node + info = lookup(sname) + assert info is not None + sock = socket.create_connection(("localhost", info.port), timeout=2) + try: + client_name = f"erldistpy_ch_tick@{socket.gethostname()}" + result = handshake(sock, our_name=client_name, cookie=cookie) + ch = Channel(sock, recv_timeout=3.0) + ch.send_tick() # provoke nothing in particular — just exercise the wire + from_pid = Pid(Atom(client_name), 2, 0, result.our_creation) + ch.send_reg_send(from_pid, Atom("echo"), (from_pid, b"after tick")) + msg = ch.recv_message() + assert msg is not None + assert msg.payload == (Atom("echoed"), b"after tick") + finally: + sock.close() + + +# Mirror struct import only used implicitly by ETF; keep linters happy. +_ = struct