OverviewFilesCommitsIssuesDocsReleasesPeers

hyperswarm-peer commits / f826718

swarmd: the two-daemon crossing, proven on the public DHT
#72's blocking item was the responder. #73's was this: channels were wired and the crossing was never shown. It is shown now — two daemons, a swarmd-probe on each, one frame over a Protomux channel, on the real DHT with a 240s window. The gate no longer writes any wire. It drives tools/swarmd-probe, which uses the generated encoders, and reads its words; the previous version hand-rolled compact-encoding in Python and the daemon answered "undecodable message" to every frame it sent — the test was wrong, not the daemon. It keeps the discipline it was written with and gains a second baseline: it still exits INCONCLUSIVE when the daemons never meet, and it now reports separately whether the daemon ever ACCEPTED the frame, so "they met but no channel opened" and "it was accepted and vanished" are different verdicts instead of one shrug. The sender starts five seconds BEFORE the receiver on purpose, so the refusal path is exercised on every run rather than only when the DHT is slow. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
f826718c0960ef48503829faea29891449f9d820
by plankatron <fleeky@gmail.com> · 2026-09-10T18:46:46.000Z

Changed files (2) +128 -112

swarmd/README.md +12 -4
……@@ -33,6 +33,7 @@ encodings the same way, with `schema/build.js` and the resulting
3333 ./build.sh # gcc/g++ only. No npm.
3434 ./test/test_uds_backpressure
3535 ./test/test_control
36 ./test/test_send_truth
3637
3738## Status
3839
……@@ -48,7 +49,14 @@ encodings the same way, with `schema/build.js` and the resulting
4849 [swarmd] dht=connecting peers=0 local=0
4950 [swarmd] dht=up peers=0 local=0
5051
51⏳ Channels are wired but the two-daemon crossing is NOT yet proven: the gate
52exists (`test/test_channel_e2e.py`) and currently exits INCONCLUSIVE because the
53daemons had not met inside its window. Content is not built. See
54`../docs/SWARMD-DESIGN.md` §9 for the order and §8 for the remaining gates.
52✅ The two-daemon crossing is proven, on the PUBLIC DHT — two daemons, a
53swarmd-probe on each, one frame across a Protomux channel:
54
55 python3 -B test/test_channel_e2e.py --seconds 240 # public DHT
56 python3 -B test/test_channel_e2e.py --bootstrap host:port --seconds 60
57
58The gate still exits INCONCLUSIVE rather than FAIL when the daemons never meet,
59because a message result from a swarm that never meshed says nothing.
60
61⏳ Content is not built. See `../docs/SWARMD-DESIGN.md` §9 for the order and §8
62for the remaining gates.
swarmd/test/test_channel_e2e.py +116 -108
……@@ -1,11 +1,18 @@
11#!/usr/bin/env python3
22"""Gate D1/D4: a frame crosses between two swarmd daemons over a real channel.
33
4 python3 -B test/test_channel_e2e.py [--bootstrap host:port]
4 python3 -B test/test_channel_e2e.py [--bootstrap host:port] [--seconds N]
55
6Two daemons join one topic, each local client opens the SAME protocol channel,
7and a frame sent by one arrives at the other. That is the whole product in one
8assertion: everything before it is plumbing.
6Two daemons join one topic, a swarmd-probe client on each opens the SAME
7protocol channel, and a frame sent by one arrives at the other. That is the
8whole product in one assertion: everything before it is plumbing.
9
10⛔ NO WIRE IS WRITTEN HERE. The first version hand-rolled compact-encoding in
11Python and the daemon answered "undecodable message" to every frame — the test
12was wrong, not the daemon. Hand-writing the wire in a client is exactly what
13generating it was supposed to stop, and a test client is still a client. So this
14drives tools/swarmd-probe, which uses the generated encoders, and reads its
15words. The only thing Python parses here is stdout.
916
1017⛔ IT ASSERTS ITS OWN BASELINE. If the two daemons never see each other the
1118message result says nothing, so a swarm that never meshes exits INCONCLUSIVE
……@@ -13,146 +20,147 @@ rather than FAIL — the mistake the first long-lived gate made, reporting "the
1320responder has gone silent" from a run in which nothing had ever connected.
1421"""
1522
16import os, socket, struct, subprocess, sys, tempfile, time
23import os, re, subprocess, sys, tempfile, time
1724
1825HERE = os.path.dirname(os.path.realpath(__file__))
19SWARMD = os.path.join(os.path.dirname(HERE), "swarmd")
26ROOT = os.path.dirname(HERE)
27SWARMD = os.path.join(ROOT, "swarmd")
28PROBE = os.path.join(ROOT, "tools", "swarmd-probe")
2029EXIT_PASS, EXIT_FAIL, EXIT_INCONCLUSIVE = 0, 1, 2
2130
22# Command ids are the append-only ids from schema/hrpc.json.
23import json
24HRPC = json.load(open(os.path.join(os.path.dirname(HERE), "schema", "hrpc.json")))
25CMD = {h["name"].split("/", 1)[1]: h["id"] for h in HRPC["schema"]}
26
27
28def frame(payload):
29 return struct.pack("<I", len(payload)) + payload
31TOPIC = "5a" * 32
32PROTOCOL = "swarmd-gate/v1"
33PAYLOAD = "hello-over-the-swarm"
3034
35# The sender goes first ON PURPOSE. It re-offers the frame until the daemon
36# accepts it, so starting it before the receiver exists exercises the refusal
37# path as well as the crossing — the case the fire-once-on-a-timer probe passed
38# by luck on a fast testnet.
39RECEIVER_DELAY = 5
3140
32def uint(n):
33 # compact-encoding uint
34 if n < 0xfc: return bytes([n])
35 if n <= 0xffff: return b"\xfd" + struct.pack("<H", n)
36 if n <= 0xffffffff: return b"\xfe" + struct.pack("<I", n)
37 return b"\xff" + struct.pack("<Q", n)
3841
42def arg(name, default=None):
43 return sys.argv[sys.argv.index(name) + 1] if name in sys.argv else default
3944
40def rpc_request(cmd_id, req_id, body):
41 # rpc_message: type=1 (request), id, command, stream=0, then data
42 return uint(1) + uint(req_id) + uint(cmd_id) + uint(0) + uint(len(body)) + body
4345
44
45def spawn(sock, bootstrap, log):
46 argv = [SWARMD, "--socket", sock]
47 if bootstrap:
48 argv += ["--bootstrap", bootstrap]
49 return subprocess.Popen(argv, stdout=log, stderr=subprocess.STDOUT)
46def read(path):
47 try:
48 with open(path, "rb") as fh:
49 return fh.read().decode("utf-8", "replace")
50 except OSError:
51 return ""
5052
5153
52def wait_log(path, needle, timeout, proc):
54def wait_for(path, needle, timeout, proc):
5355 end = time.time() + timeout
5456 while time.time() < end:
55 if proc.poll() is not None:
57 if proc is not None and proc.poll() is not None:
5658 return False
57 try:
58 with open(path, "rb") as fh:
59 if needle in fh.read():
60 return True
61 except OSError:
62 pass
59 if needle in read(path):
60 return True
6361 time.sleep(0.3)
6462 return False
6563
6664
65def peers_seen(text):
66 """The highest peer count this daemon ever reported."""
67 return max([int(m) for m in re.findall(r"peers=(\d+)", text)] or [0])
68
69
6770def main():
68 bootstrap = None
69 if "--bootstrap" in sys.argv:
70 bootstrap = sys.argv[sys.argv.index("--bootstrap") + 1]
71 bootstrap = arg("--bootstrap")
72 window = int(arg("--seconds", "60" if bootstrap else "180"))
73
74 for path in (SWARMD, PROBE):
75 if not os.path.isfile(path):
76 print("missing %s — run ./build.sh" % path)
77 return EXIT_INCONCLUSIVE
7178
72 if not os.path.isfile(SWARMD):
73 print("missing %s — run ./build.sh" % SWARMD)
74 return EXIT_INCONCLUSIVE
79 where = "bootstrap %s" % bootstrap if bootstrap else "the PUBLIC DHT"
80 print("two daemons on %s, %ds window" % (where, window))
7581
7682 tmp = tempfile.mkdtemp(prefix="swarmd-chan-")
77 topic = bytes([0x5a] * 32)
78 protocol = b"swarmd-gate/v1"
83 procs, socks, logs = [], [], {}
84
85 def spawn(argv, logpath):
86 fh = open(logpath, "wb")
87 p = subprocess.Popen(argv, stdout=fh, stderr=subprocess.STDOUT)
88 procs.append(p)
89 return p
7990
80 procs, socks, logs = [], [], []
8191 try:
92 daemons = {}
8293 for tag in ("a", "b"):
8394 sp = os.path.join(tmp, tag + ".sock")
84 lp = os.path.join(tmp, tag + ".log")
85 log = open(lp, "wb")
86 p = spawn(sp, bootstrap, log)
87 procs.append(p); socks.append(sp); logs.append(lp)
88
89 for i, lp in enumerate(logs):
90 if not wait_log(lp, b"dht=up", 60, procs[i]):
91 print("\nINCONCLUSIVE: daemon %d never reached dht=up" % i)
95 logs[tag] = os.path.join(tmp, tag + ".log")
96 argv = [SWARMD, "--socket", sp]
97 if bootstrap:
98 argv += ["--bootstrap", bootstrap]
99 daemons[tag] = spawn(argv, logs[tag])
100 socks.append(sp)
101
102 # ---- baseline 1: both daemons are actually on a DHT ----
103 for tag in ("a", "b"):
104 if not wait_for(logs[tag], "dht=up", 90, daemons[tag]):
105 print("\nINCONCLUSIVE: daemon %s never reached dht=up." % tag)
92106 return EXIT_INCONCLUSIVE
93107 print(" both daemons are on the DHT")
94108
95 conns = []
96 for sp in socks:
97 c = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
98 c.connect(sp)
99 conns.append(c)
100
101 # join the topic, open the channel
102 for c in conns:
103 body = topic + b"\x01\x01" # fixed32 + bool + bool
104 c.sendall(frame(rpc_request(CMD["swarm-join"], 1, body)))
105 ob = uint(len(protocol)) + protocol + uint(0) + uint(2) # protocol, id(empty), slots
106 c.sendall(frame(rpc_request(CMD["channel-open"], 2, ob)))
107 print(" joined and opened the channel on both")
108
109 # Give the swarm time to mesh and pair the channel.
110 deadline = time.time() + 90
111 meshed = False
112 while time.time() < deadline and not meshed:
113 for lp in logs:
114 with open(lp, "rb") as fh:
115 if b"peers=1" in fh.read():
116 meshed = True
117 time.sleep(1)
109 def probe(tag, sock_tag, extra, seconds):
110 logs[tag] = os.path.join(tmp, tag + ".log")
111 argv = [PROBE, "--socket", os.path.join(tmp, sock_tag + ".sock"),
112 "--topic-hex", TOPIC, "--protocol", PROTOCOL,
113 "--seconds", str(seconds)] + extra
114 return spawn(argv, logs[tag])
115
116 sender = probe("send", "a", ["--send", PAYLOAD], window)
117 time.sleep(RECEIVER_DELAY)
118 receiver = probe("recv", "b", [], window - RECEIVER_DELAY)
119 print(" probes attached; waiting up to %ds" % window)
118120
119 if not meshed:
120 print("\nINCONCLUSIVE: the two daemons never met, so a message result "
121 "would say nothing about the channel.")
121 end = time.time() + window + 15
122 while time.time() < end and receiver.poll() is None:
123 time.sleep(1)
124 for p in (sender, receiver):
125 if p.poll() is None:
126 p.terminate()
127 try: p.wait(timeout=10)
128 except subprocess.TimeoutExpired: p.kill()
129
130 sender_out, receiver_out = read(logs["send"]), read(logs["recv"])
131 met = min(peers_seen(read(logs["a"])), peers_seen(read(logs["b"])))
132 accepted = "accepted after" in sender_out
133
134 print(" daemons met: %s | frame accepted by the daemon: %s"
135 % ("yes" if met else "no", "yes" if accepted else "no"))
136
137 # ---- baseline 2: if they never met, nothing downstream means anything ----
138 if not met:
139 print("\nINCONCLUSIVE: the two daemons never saw each other, so the "
140 "message result says nothing about the channel.")
141 print("(logs: %s)" % tmp)
122142 return EXIT_INCONCLUSIVE
123 print(" the daemons meshed")
124
125 # a -> b
126 payload = b"hello-over-the-swarm"
127 sb = uint(1) + uint(0) + uint(0) + uint(len(payload)) + payload # channel, slot, target, bytes
128 conns[0].sendall(frame(rpc_request(CMD["send"], 3, sb)))
129
130 conns[1].settimeout(30)
131 got = b""
132 end = time.time() + 30
133 while time.time() < end:
134 try:
135 chunk = conns[1].recv(65536)
136 except socket.timeout:
137 break
138 if not chunk:
139 break
140 got += chunk
141 if payload in got:
142 break
143
144 if payload in got:
143
144 if ("RECEIVED: " + PAYLOAD) in receiver_out:
145145 print("\nPASS: the frame crossed the swarm and arrived on the other daemon.")
146146 return EXIT_PASS
147 print("\nFAIL: the daemons meshed but the frame did not arrive.")
147
148 if not accepted:
149 print("\nFAIL: the daemons met but neither channel ever opened — the "
150 "daemon refused the frame for the whole window.")
151 else:
152 print("\nFAIL: the daemon accepted the frame and it never arrived.")
153 print("--- sender ---\n%s--- receiver ---\n%s" % (sender_out, receiver_out))
148154 return EXIT_FAIL
149155 finally:
150 for c in socks:
151 pass
152156 for p in procs:
153 p.terminate()
154 try: p.wait(timeout=5)
155 except subprocess.TimeoutExpired: p.kill()
157 if p.poll() is None:
158 p.terminate()
159 try: p.wait(timeout=5)
160 except subprocess.TimeoutExpired: p.kill()
161 for sp in socks:
162 try: os.unlink(sp)
163 except OSError: pass
156164 print("(logs: %s)" % tmp)
157165
158166