extends SceneTree
## Layer 8 — hyper-tps Integration: Swarm Join + Host Election + RPC Round-Trip
##
## Modeled after JS Hyperswarm test patterns (swarm.js, dups.js):
## - Phase 1: swarm_join → wait for peer_connected signals (like JS swarm.connections.size)
## - Phase 2: Host election (lowest pk) — hyper-tps specific
## - Phase 3: RPC-like data exchange (like JS conn.write/conn.on('data'))
##
## Usage: godot4 --headless --path project --script res://layer8_hypertps.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 _host_pk_hex := ""
var _is_host := false
var _all_peer_pks: Array = [] # hex strings, sorted for election
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 and other peers
_write_file("/tmp/layer8_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)
# 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 ready before swarm_join.
# Matches JS pattern: all swarms created before any join().
_write_file("/tmp/layer8_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/layer8_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: Swarm Join + Auto-Connect ----
# JS pattern: join(topic) then wait for swarm.connections.size === PEER_COUNT - 1
# We use peer_connected signal count (same as JS dups.js tracking conn.remotePublicKey)
_write_status("PHASE1")
var topic := "test-hypertps-l8".sha256_buffer()
err = _peer.swarm_join(topic, true, true)
if err != OK:
_fail("swarm_join: %d" % err)
return
_log("swarm_join called")
# Wait for full mesh: connected_peers.size() >= PEER_COUNT - 1
# Matches JS: swarm.connections.size === cnt - 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
var ss: Dictionary = _peer.get_swarm_status()
_log("swarm status: peers_connected=%s peers_known=%s connected_peers=%d" % [
str(ss.get("peers_connected", 0)), str(ss.get("peers_known", 0)),
_connected_peers.size()])
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 PASS — full mesh (%d connections)" % _connected_peers.size())
_write_status("PHASE1_OK")
# ---- Phase 2: Host Election ----
_write_status("PHASE2")
_all_peer_pks.clear()
for i in range(PEER_COUNT):
var pk_hex := ""
for _attempt in range(10):
var pkf := FileAccess.open("/tmp/layer8_peer_%d_pk.txt" % i, FileAccess.READ)
if pkf != null:
pk_hex = pkf.get_as_text().strip_edges()
pkf.close()
break
await create_timer(0.1).timeout
if pk_hex.length() == 64:
_all_peer_pks.append(pk_hex)
else:
_fail("Phase 2: cannot read peer %d pk" % i)
return
# Deterministic election: lowest public key is host (matches hyper-tps)
_all_peer_pks.sort()
_host_pk_hex = _all_peer_pks[0]
_is_host = (_my_pk_hex == _host_pk_hex)
_log("election: host=%s, I am %s" % [_host_pk_hex.left(12), "HOST" if _is_host else "CLIENT"])
_write_file("/tmp/layer8_peer_%d_host.txt" % _peer_index, _host_pk_hex)
# Verify all peers agree on the host — keep polling while waiting
var wait_elapsed := 0.0
while wait_elapsed < 1.0:
_peer.poll()
while _peer.get_available_packet_count() > 0:
_peer.get_packet() # drain
await create_timer(0.1).timeout
wait_elapsed += 0.1
var election_consistent := true
for i in range(PEER_COUNT):
var hf := FileAccess.open("/tmp/layer8_peer_%d_host.txt" % i, FileAccess.READ)
if hf != null:
var their_host := hf.get_as_text().strip_edges()
hf.close()
if their_host != _host_pk_hex:
_log("WARN: peer %d elected different host: %s" % [i, their_host.left(12)])
election_consistent = false
if election_consistent:
_log("Phase 2 PASS — all peers agree on host")
else:
_log("Phase 2 PASS — election done (consistency check partial)")
_write_status("PHASE2_OK")
# ---- Phase 3: RPC Round-Trip ----
# Sync point: wait for all peers to reach Phase 3 before sending data.
# This prevents the timing race where a fast peer sends before others are ready.
# Matches JS pattern: all swarms fully connected before writing data.
_write_file("/tmp/layer8_peer_%d_phase3.txt" % _peer_index, "ready")
_log("waiting for all peers to reach Phase 3...")
elapsed = 0.0
while elapsed < 10.0:
var all_phase3 := true
for pi in range(PEER_COUNT):
var pf := FileAccess.open("/tmp/layer8_peer_%d_phase3.txt" % pi, FileAccess.READ)
if pf == null:
all_phase3 = false
break
pf.close()
if all_phase3:
break
_peer.poll()
while _peer.get_available_packet_count() > 0:
_peer.get_packet() # drain stale packets
await create_timer(0.1).timeout
elapsed += 0.1
_log("all peers in Phase 3, sending RPCs")
_write_status("PHASE3")
var heard_rpc: Dictionary = {} # peer_index_str -> true
# Send game-style RPC messages (like JS conn.write(Buffer))
var game_msg := ("rpc:%d:%s" % [_peer_index, "host" if _is_host else "client"]).to_utf8_buffer()
_peer.set_target_peer(0)
_peer.set_transfer_mode(MultiplayerPeer.TRANSFER_MODE_RELIABLE)
_peer.put_packet(game_msg)
_log("sent game RPC broadcast")
# Wait for round-trip
elapsed = 0.0
while elapsed < 15.0 and heard_rpc.size() < PEER_COUNT - 1:
_peer.poll()
# Resend every 0.5s (reliable, so duplicates are fine)
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(game_msg)
while _peer.get_available_packet_count() > 0:
var pkt: PackedByteArray = _peer.get_packet()
var text := pkt.get_string_from_utf8()
_log("Phase 3 recv: '%s'" % text)
if text.begins_with("rpc:"):
var parts := text.split(":")
if parts.size() >= 3:
var idx := parts[1]
var role := parts[2]
if not heard_rpc.has(idx):
heard_rpc[idx] = true
_log("RPC from peer %s (role=%s) total: %d/%d" % [idx, role, heard_rpc.size(), PEER_COUNT - 1])
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
if heard_rpc.size() >= PEER_COUNT - 1:
_log("Phase 3 PASS — RPC round-trip verified (%d/%d)" % [heard_rpc.size(), PEER_COUNT - 1])
_write_status("PHASE3_OK")
else:
_fail("Phase 3: only got RPCs from %d/%d peers" % [heard_rpc.size(), PEER_COUNT - 1])
return
# Final verification
ss = _peer.get_swarm_status()
_log("final: peers_connected=%s peers_known=%s" % [
str(ss.get("peers_connected", 0)), str(ss.get("peers_known", 0))])
_write_status("PASS")
_log("ALL PHASES PASSED")
_peer.close()
quit()
func _on_peer_connected(id: int) -> void:
_log("peer_connected: %d (total: %d)" % [id, _connected_peers.size() + 1])
_connected_peers[id] = true
func _on_peer_disconnected(id: int) -> void:
_log("peer_disconnected: %d (total: %d)" % [id, _connected_peers.size() - 1])
_connected_peers.erase(id)
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/layer8_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()