phase 4: distribution data channel

Channel wraps the post-handshake socket and carries 4-byte length-
prefixed distribution messages: pass-through byte ('p') + ETF control
tuple + optional payload term.

API surface:
  send_raw / recv_raw     -- raw 4-byte framed bytes, empty == net_tick
  send_tick               -- send keepalive frame
  send_control / recv_message  -- structured control + payload
  send_reg_send           -- helper for the REG_SEND case (FromPid,
                             registered name, payload)

recv_message() transparently skips inbound ticks; callers wanting tick
awareness use recv_raw().

etf.decode_term(data, offset) exposed as a streaming decoder so the
channel can read control + payload back-to-back from one frame body.

Tests:
  - pure encode/decode round-trips
  - socketpair tests for framing, ticks, helper signatures
  - live end-to-end against an erl node with a registered echo process:
    EPMD -> handshake -> REG_SEND -> recv reply, payload matches
  - boot script writes a /tmp ready-flag after registering `echo`;
    fixture waits for both EPMD registration AND the flag to dodge
    the race where EPMD registers the node before -eval runs

101 tests green, lint clean.

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
 

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__",

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)

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")

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