#!/usr/bin/env python3
"""Gate D5: a peer that goes away is reported, and not on a wall clock.
python3 -B test/test_peer_down.py [--bootstrap host:port] [--seconds N]
Two daemons meet on a topic; one is KILLED; the survivor's client must be told.
⛔ THE DEADLINE IS DERIVED FROM THE CONSTANT THAT GOVERNS IT, not picked.
UdxTransport sends a zero-length encrypted keepalive to an idle peer every
PEER_KEEPALIVE_INTERVAL_MS (5s) and disconnects one that has said nothing for
PEER_DEAD_MS (20s) — so a correct implementation reports inside ~20 seconds and
this allows 30. Set DOWN_DEADLINE_S to probe a different bound.
⛔ SIGKILL, NOT SIGTERM. A clean shutdown is the easy case and not the one that
hurts: a process that was killed never gets to say goodbye.
⛔ IT ASSERTS ITS OWN BASELINE. If the two daemons never meet, "no PEER-DOWN"
means nothing, so a swarm that never meshed exits INCONCLUSIVE rather than FAIL.
⚠ WHAT THIS GATE COST, recorded because the shape recurs (hyper-games #75):
nothing detected a dead peer at all, and three layers each looked like they did.
DhtConnector's DHT_CONN_CONNECTED keepalive branch is unreachable — the
connection is freed the instant it reaches CONNECTED. libudx's keepalive was
never switched on, and switching it on is not liveness either: send_probe writes
a bare heartbeat OUTSIDE the packet and RTO machinery, and libudx's own TODO at
that line says the packet that would fix it is not written. The fix had to be at
our layer, and its wire form is the reference's — @hyperswarm/secret-stream
sends an empty encrypted frame and eats the ones it receives.
"""
import os, re, signal, subprocess, sys, tempfile, time
HERE = os.path.dirname(os.path.realpath(__file__))
ROOT = os.path.dirname(HERE)
SWARMD = os.path.join(ROOT, "swarmd")
PROBE = os.path.join(ROOT, "tools", "swarmd-probe")
EXIT_PASS, EXIT_FAIL, EXIT_INCONCLUSIVE = 0, 1, 2
TOPIC = "d5" * 32
PROTOCOL = "swarmd-peerdown/v1"
# PEER_DEAD_MS is 20s (src/udx_transport.h), so a working implementation reports
# inside that. Thirty allows for the keepalive phase and a slow testnet.
DOWN_DEADLINE_S = int(os.environ.get("DOWN_DEADLINE_S", "30"))
def arg(name, default=None):
return sys.argv[sys.argv.index(name) + 1] if name in sys.argv else default
def read(path):
try:
with open(path, "rb") as fh:
return fh.read().decode("utf-8", "replace")
except OSError:
return ""
def wait_for(path, needle, timeout, proc=None):
end = time.time() + timeout
while time.time() < end:
if proc is not None and proc.poll() is not None:
return False
if needle in read(path):
return True
time.sleep(0.2)
return False
def main():
bootstrap = arg("--bootstrap")
window = int(arg("--seconds", "90" if bootstrap else "240"))
for path in (SWARMD, PROBE):
if not os.path.isfile(path):
print("missing %s — run ./build.sh" % path)
return EXIT_INCONCLUSIVE
tmp = tempfile.mkdtemp(prefix="swarmd-peerdown-")
procs, logs = [], {}
def spawn(argv, tag):
logs[tag] = os.path.join(tmp, tag + ".log")
fh = open(logs[tag], "wb")
p = subprocess.Popen(argv, stdout=fh, stderr=subprocess.STDOUT)
procs.append(p)
return p
try:
daemons = {}
for tag in ("a", "b"):
argv = [SWARMD, "--socket", os.path.join(tmp, tag + ".sock")]
if bootstrap:
argv += ["--bootstrap", bootstrap]
daemons[tag] = spawn(argv, tag)
for tag in ("a", "b"):
if not wait_for(logs[tag], "dht=up", 90, daemons[tag]):
print("\nINCONCLUSIVE: daemon %s never reached dht=up." % tag)
return EXIT_INCONCLUSIVE
print(" both daemons are on the DHT")
watchers = {}
for tag in ("a", "b"):
watchers[tag] = spawn([PROBE, "--socket", os.path.join(tmp, tag + ".sock"),
"--topic-hex", TOPIC, "--protocol", PROTOCOL,
"--watch-peers", "--seconds", str(window)],
"watch-" + tag)
# ---- baseline: they have to meet before a departure means anything ----
if not wait_for(logs["watch-a"], "PEER-UP", window, watchers["a"]):
print("\nINCONCLUSIVE: daemon A never saw a peer, so its silence "
"afterwards says nothing.")
return EXIT_INCONCLUSIVE
print(" A saw the peer arrive")
# D7's other half, read from a client rather than from a log line.
status = re.search(r"STATUS dht_state=(\d+) bootstrapped=(\d+)", read(logs["watch-a"]))
if status:
print(" A's STATUS: dht_state=%s bootstrapped=%s"
% (status.group(1), status.group(2)))
# ---- the departure ----
killed_at = time.time()
daemons["b"].send_signal(signal.SIGKILL)
daemons["b"].wait(timeout=10)
print(" B killed with SIGKILL")
seen = wait_for(logs["watch-a"], "PEER-DOWN", DOWN_DEADLINE_S, watchers["a"])
elapsed = time.time() - killed_at
if not seen:
print("\nFAIL: A was not told its peer had gone within %ds, and "
"PEER_DEAD_MS is 20s — so either the keepalive is not being "
"sent, or the sweep is not disconnecting on it."
% DOWN_DEADLINE_S)
print("--- A ---\n%s" % read(logs["watch-a"])[-1500:])
return EXIT_FAIL
print("\nPASS: PEER-DOWN reached the client %.1fs after the peer was "
"killed (PEER_DEAD_MS 20s, deadline %ds)." % (elapsed, DOWN_DEADLINE_S))
if status and status.group(1) != "2":
print(" ⚠ but STATUS reported dht_state=%s while the daemon's own log "
"said dht=up — those disagree." % status.group(1))
return EXIT_FAIL
return EXIT_PASS
finally:
for p in procs:
if p.poll() is None:
p.terminate()
try: p.wait(timeout=5)
except subprocess.TimeoutExpired: p.kill()
print("(logs: %s)" % tmp)
if __name__ == "__main__":
sys.exit(main())