extends SceneTree
## Layer 7 — Swarm Peer: Auto-Connect + Auto-Reconnect + Explicit Peer
##
## Tests SwarmManager behavior:
## Phase 1: 3 peers swarm_join(topic) → auto-connect + message exchange
## Phase 2: Kill peer 0 → auto-reconnect via RetryTimer
## Phase 3: Explicit peer via swarm_join_peer()
## Phase 4: Verify get_swarm_status() diagnostics
##
## Usage: godot4 --headless --path project --script res://layer7_swarm_peer.gd -- --peer-index 0
const TIMEOUT := 30.0
const POLL_HZ := 0.05
const PEER_COUNT := 3
var _peer_index := 0
var _peer: Object
var _dht_kp: Dictionary
var _dht_pk: PackedByteArray
var _my_pk_hex := ""
var _bootstrap: PackedStringArray
var _connected_peers: Dictionary = {} # peer_id -> true
var _heard_from: Dictionary = {} # peer_index_str -> true
var _phase := "INIT"
func _init() -> void:
var args := OS.get_cmdline_user_args()
for i in args.size():
if args[i] == "--peer-index" and i + 1 < args.size():
_peer_index = args[i + 1].to_int()
_write_status("STARTING")
_start.call_deferred()
func _start() -> void:
_log("starting peer index %d" % _peer_index)
# Read bootstrap
var f := FileAccess.open("/tmp/layer_bootstrap.txt", FileAccess.READ)
if f == null:
_fail("Cannot read bootstrap")
return
_bootstrap = PackedStringArray(f.get_as_text().strip_edges().split(","))
f.close()
# Create peer
_peer = ClassDB.instantiate(&"HyperswarmMultiplayerPeer")
# Generate DHT keypair
_dht_kp = _peer.generate_dht_keypair()
_dht_pk = _dht_kp["public_key"]
_my_pk_hex = _dht_pk.hex_encode()
_log("dht pk: %s" % _my_pk_hex.left(12))
# Write pk for orchestrator
_write_file("/tmp/layer7_peer_%d_pk.txt" % _peer_index, _my_pk_hex)
# Connect signals
_peer.peer_connected.connect(_on_peer_connected)
_peer.peer_disconnected.connect(_on_peer_disconnected)
_peer.swarm_connection.connect(_on_swarm_connection)
# Join DHT
var err = _peer.join_dht(_dht_pk, _dht_kp["secret_key"], _bootstrap)
if err != OK:
_fail("join_dht: %d" % err)
return
# Wait for DHT active
var elapsed := 0.0
while elapsed < 10.0:
_peer.poll()
if _peer.get_dht_status().get("active", false):
break
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
if not _peer.get_dht_status().get("active", false):
_fail("DHT not active after 10s")
return
_log("DHT active")
# ---- Sync point: wait for all peers to be DHT-active before swarm_join ----
# Matches JS Hyperswarm test pattern: all swarms created before any join()
_write_file("/tmp/layer7_peer_%d_ready.txt" % _peer_index, "ready")
_log("waiting for all peers to be ready...")
elapsed = 0.0
while elapsed < 15.0:
var all_ready := true
for pi in range(PEER_COUNT):
var rf := FileAccess.open("/tmp/layer7_peer_%d_ready.txt" % pi, FileAccess.READ)
if rf == null:
all_ready = false
break
rf.close()
if all_ready:
break
_peer.poll()
await create_timer(0.1).timeout
elapsed += 0.1
_log("all peers ready, starting swarm_join")
# ---- Phase 1: Auto-connect via swarm_join ----
_phase = "PHASE1"
_write_status("PHASE1")
var topic := "test-swarm-l7".sha256_buffer()
# Key difference from L3: swarm_join handles everything automatically
err = _peer.swarm_join(topic, true, true)
if err != OK:
_fail("swarm_join: %d" % err)
return
_log("swarm_join called — auto-connect enabled")
# Wait for full mesh: connected_peers.size() >= PEER_COUNT - 1
elapsed = 0.0
while elapsed < TIMEOUT and _connected_peers.size() < PEER_COUNT - 1:
_peer.poll()
# Drain any packets that arrive during connection setup
while _peer.get_available_packet_count() > 0:
_peer.get_packet()
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
# Check swarm status
var ss: Dictionary = _peer.get_swarm_status()
_log("swarm status: peers_connected=%s peers_known=%s connecting=%s queue_size=%s" % [
str(ss.get("peers_connected", 0)), str(ss.get("peers_known", 0)),
str(ss.get("connecting", 0)), str(ss.get("queue_size", 0))])
if _connected_peers.size() < PEER_COUNT - 1:
_fail("Phase 1: only %d/%d peers connected" % [_connected_peers.size(), PEER_COUNT - 1])
return
_log("Phase 1: full mesh (%d connections)" % _connected_peers.size())
# ---- Sync point: wait for all peers to be connected before message exchange ----
_write_file("/tmp/layer7_peer_%d_phase1.txt" % _peer_index, "connected")
_log("waiting for all peers to reach Phase 1 sync...")
elapsed = 0.0
while elapsed < 10.0:
var all_p1 := true
for pi in range(PEER_COUNT):
var pf := FileAccess.open("/tmp/layer7_peer_%d_phase1.txt" % pi, FileAccess.READ)
if pf == null:
all_p1 = false
break
pf.close()
if all_p1:
break
_peer.poll()
while _peer.get_available_packet_count() > 0:
_peer.get_packet() # drain stale
await create_timer(0.1).timeout
elapsed += 0.1
_log("all peers connected, sending identity broadcasts")
# Send identity message
var msg := ("I am peer %d" % _peer_index).to_utf8_buffer()
_peer.set_target_peer(0)
_peer.set_transfer_mode(MultiplayerPeer.TRANSFER_MODE_RELIABLE)
_peer.put_packet(msg)
_log("sent identity broadcast")
# Wait for messages from all other peers
elapsed = 0.0
while elapsed < 15.0 and _heard_from.size() < PEER_COUNT - 1:
_peer.poll()
# Resend every 0.5s
if int(elapsed * 20) % 10 == 0 and elapsed > 0.1:
_peer.set_target_peer(0)
_peer.set_transfer_mode(MultiplayerPeer.TRANSFER_MODE_RELIABLE)
_peer.put_packet(msg)
while _peer.get_available_packet_count() > 0:
var pkt: PackedByteArray = _peer.get_packet()
var text := pkt.get_string_from_utf8()
_log("received: '%s'" % text)
if text.begins_with("I am peer "):
var idx := text.substr(10).strip_edges()
if not _heard_from.has(idx):
_heard_from[idx] = true
_log("heard from peer %s (total: %d/%d)" % [idx, _heard_from.size(), PEER_COUNT - 1])
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
if _heard_from.size() >= PEER_COUNT - 1:
_write_status("PHASE1_OK")
_log("Phase 1 PASS — auto-connect + messages working")
else:
_log("Phase 1: heard from %d/%d" % [_heard_from.size(), PEER_COUNT - 1])
_write_status("PHASE1_OK") # Partial OK
# ---- Phase 2: Auto-reconnect ----
_phase = "PHASE2"
if _peer_index == 0:
# Peer 0: wait to be killed by orchestrator
_log("peer 0 waiting to be killed...")
elapsed = 0.0
while elapsed < 30.0:
_peer.poll()
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
var kf := FileAccess.open("/tmp/layer7_kill_peer0.txt", FileAccess.READ)
if kf != null:
kf.close()
_log("killed by orchestrator")
_peer.close()
_write_status("KILLED")
quit()
return
_peer.close()
_write_status("PHASE1_OK")
quit()
return
# Non-peer-0: wait for peer 0 disconnect, then verify auto-reconnect
_log("waiting for peer 0 disconnect...")
elapsed = 0.0
var initial_connections: int = _connected_peers.size()
var disconnect_detected := false
while elapsed < 30.0:
_peer.poll()
if _connected_peers.size() < initial_connections and not disconnect_detected:
disconnect_detected = true
_log("peer 0 disconnected — waiting for auto-reconnect with remaining peers")
# After disconnect, check if remaining peers are still/re-connected
if disconnect_detected:
ss = _peer.get_swarm_status()
if ss.get("peers_connected", 0) >= 1:
_write_status("PHASE2_OK")
_log("Phase 2 PASS — still connected after peer 0 death")
break
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
if not disconnect_detected:
_write_status("PHASE2_SKIP")
_log("Phase 2: peer 0 did not disconnect")
# ---- Phase 3: Explicit peer ----
_phase = "PHASE3"
# Read other peer's pk from file and add via swarm_join_peer
var other_idx: int = 2 if _peer_index == 1 else 1
var other_pk_hex := ""
var opf := FileAccess.open("/tmp/layer7_peer_%d_pk.txt" % other_idx, FileAccess.READ)
if opf != null:
other_pk_hex = opf.get_as_text().strip_edges()
opf.close()
if other_pk_hex.length() == 64:
var other_pk := _hex_to_bytes(other_pk_hex)
err = _peer.swarm_join_peer(other_pk)
if err == OK:
_log("swarm_join_peer called for peer %d" % other_idx)
# Wait briefly for status to update
elapsed = 0.0
while elapsed < 5.0:
_peer.poll()
ss = _peer.get_swarm_status()
if ss.get("explicit_count", 0) >= 1:
_write_status("PHASE3_OK")
_log("Phase 3 PASS — explicit_count=%d" % ss.get("explicit_count", 0))
break
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
else:
_log("swarm_join_peer failed: %d" % err)
_write_status("PHASE3_SKIP")
else:
_log("Phase 3: could not read other peer pk")
_write_status("PHASE3_SKIP")
# ---- Phase 4: Status verification ----
_phase = "PHASE4"
ss = _peer.get_swarm_status()
_log("final status: peers_connected=%s connecting=%s queue_size=%s max_peers=%s max_parallel=%s" % [
str(ss.get("peers_connected", 0)), str(ss.get("connecting", 0)),
str(ss.get("queue_size", 0)), str(ss.get("max_peers", 0)),
str(ss.get("max_parallel", 0))])
var p4_ok := true
if ss.get("max_peers", 0) != 64:
_log("FAIL: max_peers != 64")
p4_ok = false
if ss.get("max_parallel", 0) != 3:
_log("FAIL: max_parallel != 3")
p4_ok = false
if p4_ok:
_write_status("PHASE4_OK")
_log("Phase 4 PASS — status verification")
else:
_write_status("PHASE4_FAIL")
_peer.close()
quit()
func _on_peer_connected(id: int) -> void:
_log("peer_connected: %d" % id)
_connected_peers[id] = true
func _on_peer_disconnected(id: int) -> void:
_log("peer_disconnected: %d" % id)
_connected_peers.erase(id)
func _on_swarm_connection(peer_id: int, public_key: PackedByteArray, is_client: bool) -> void:
_log("swarm_connection: peer_id=%d pk=%s is_client=%s" % [
peer_id, public_key.hex_encode().left(12),
"true" if is_client else "false"])
func _hex_to_bytes(hex: String) -> PackedByteArray:
var result := PackedByteArray()
for i in range(0, hex.length(), 2):
result.append(("0x" + hex.substr(i, 2)).hex_to_int())
return result
func _write_file(path: String, content: String) -> void:
var wf := FileAccess.open(path, FileAccess.WRITE)
wf.store_string(content)
wf.close()
func _write_status(status: String) -> void:
_write_file("/tmp/layer7_peer_%d_status.txt" % _peer_index, status)
func _log(msg: String) -> void:
print("[P%d] %s" % [_peer_index, msg])
func _fail(msg: String) -> void:
_log("FAIL: %s" % msg)
_write_status("FAIL: %s" % msg)
if _peer:
_peer.close()
quit()