extends SceneTree
## Layer 6 — Full E2E Player: Election → Play → Migrate → Persist
##
## Pure P2P via swarm_join (no sidecars). The complete hyper-tps flow:
## 1. DHT bootstrap + swarm_join → auto-connect mesh
## 2. Host election (lowest public key)
## 3. Game RPCs over mesh
## 4. Durable state via file (simulates persistence)
## 5. Host migration on disconnect → re-election
##
## Usage: godot4 --headless --path project --script res://layer6_player.gd -- --player-index 0
const TIMEOUT := 30.0
const POLL_HZ := 0.05
const PLAYER_COUNT := 3
var _index := 0
var _peer: Object # HyperswarmMultiplayerPeer
var _my_pk_hex := ""
var _connected_peers: Dictionary = {} # peer_id -> true
var _is_host := false
var _host_pk := ""
var _all_pks: Array[String] = []
var _pass := 0
var _fail_count := 0
var _total := 0
func _init() -> void:
var args := OS.get_cmdline_user_args()
for i in args.size():
if args[i] == "--player-index" and i + 1 < args.size():
_index = args[i + 1].to_int()
_write_status("STARTING")
_start.call_deferred()
func _start() -> void:
_log("starting player %d" % _index)
# Read bootstrap
var f := FileAccess.open("/tmp/layer_bootstrap.txt", FileAccess.READ)
if f == null:
_fail("Cannot read bootstrap")
return
var bootstrap := PackedStringArray(f.get_as_text().strip_edges().split(","))
f.close()
# Create peer
_peer = ClassDB.instantiate(&"HyperswarmMultiplayerPeer")
var dht_kp: Dictionary = _peer.generate_dht_keypair()
var dht_pk: PackedByteArray = 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/layer6_peer_%d_pk.txt" % _index, _my_pk_hex)
_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")
return
_log("DHT active")
# Sync: wait for all players to be DHT-active
_write_file("/tmp/layer6_player_%d_ready.txt" % _index, "ready")
elapsed = 0.0
while elapsed < 15.0:
var all_ready := true
for pi in range(PLAYER_COUNT):
var rf := FileAccess.open("/tmp/layer6_player_%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 players ready, joining swarm")
# Swarm join
var topic := "test-e2e-l6".sha256_buffer()
err = _peer.swarm_join(topic, true, true)
if err != OK:
_fail("swarm_join: %d" % err)
return
_log("swarm_join called")
# Wait for mesh
elapsed = 0.0
while elapsed < TIMEOUT and _connected_peers.size() < PLAYER_COUNT - 1:
_peer.poll()
while _peer.get_available_packet_count() > 0:
_peer.get_packet()
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
_test("P2P mesh established", _connected_peers.size() >= PLAYER_COUNT - 1)
if _connected_peers.size() < PLAYER_COUNT - 1:
_fail("Mesh timeout: only %d/%d peers" % [_connected_peers.size(), PLAYER_COUNT - 1])
return
_log("mesh formed: %d connections" % _connected_peers.size())
# ---- Host Election ----
# Read all peer pks
_all_pks.clear()
for pi in range(PLAYER_COUNT):
var pk_hex := ""
for _attempt in range(10):
var pkf := FileAccess.open("/tmp/layer6_peer_%d_pk.txt" % pi, 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_pks.append(pk_hex)
else:
_fail("Cannot read peer %d pk" % pi)
return
_all_pks.sort()
_host_pk = _all_pks[0]
_is_host = (_host_pk == _my_pk_hex)
_log("election: host=%s, I am %s" % [_host_pk.left(12), "HOST" if _is_host else "CLIENT"])
_test("Host election deterministic", not _host_pk.is_empty())
if _is_host:
_write_file("/tmp/layer6_host_pk.txt", _my_pk_hex)
# ---- Phase 4: Play — exchange game RPCs ----
# Sync: all players meshed before sending
_write_file("/tmp/layer6_player_%d_mesh.txt" % _index, "meshed")
elapsed = 0.0
while elapsed < 10.0:
var all_meshed := true
for pi in range(PLAYER_COUNT):
var mf := FileAccess.open("/tmp/layer6_player_%d_mesh.txt" % pi, FileAccess.READ)
if mf == null:
all_meshed = false
break
mf.close()
if all_meshed:
break
_peer.poll()
while _peer.get_available_packet_count() > 0:
_peer.get_packet()
await create_timer(0.1).timeout
elapsed += 0.1
_write_status("PLAY")
var game_msg := ('{"type":"game","player":%d,"action":"move","pos":[%d,%d,%d]}' %
[_index, _index * 10, _index * 20, _index * 30]).to_utf8_buffer()
_peer.set_target_peer(0)
_peer.set_transfer_mode(MultiplayerPeer.TRANSFER_MODE_RELIABLE)
_peer.put_packet(game_msg)
_log("sent game action")
# If host, write durable state
if _is_host:
var score_msg := ('{"type":"durable_score","player":%d,"score":100}' % _index).to_utf8_buffer()
_peer.set_target_peer(0)
_peer.set_transfer_mode(MultiplayerPeer.TRANSFER_MODE_RELIABLE)
_peer.put_packet(score_msg)
_log("sent durable score")
_write_file("/tmp/layer6_durable_score_host.txt", "100")
# Receive game messages
var heard_players: Dictionary = {}
elapsed = 0.0
while elapsed < 15.0 and heard_players.size() < PLAYER_COUNT - 1:
_peer.poll()
# Resend periodically
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()
var parsed: Variant = JSON.parse_string(text)
if parsed is Dictionary:
var p_idx: Variant = parsed.get("player", -1)
if p_idx is float:
p_idx = int(p_idx)
if p_idx != _index and p_idx >= 0:
heard_players[p_idx] = true
_log("heard from player %s" % str(p_idx))
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
_test("Game RPC exchange", heard_players.size() > 0)
_write_status("PHASE4_OK")
_log("Phase 4 complete — heard from %d players" % heard_players.size())
# ---- Phase 5: Host Migration ----
if _is_host:
_log("host waiting for kill signal...")
elapsed = 0.0
while elapsed < 20.0:
_peer.poll()
var kf := FileAccess.open("/tmp/layer6_kill_host.txt", FileAccess.READ)
if kf != null:
kf.close()
break
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
_log("host exiting")
_write_status("KILLED")
_peer.close()
_report()
quit()
return
# Non-host: wait for host death, re-elect
_log("waiting for host migration...")
elapsed = 0.0
var host_dead := false
while elapsed < 20.0:
_peer.poll()
var kf := FileAccess.open("/tmp/layer6_kill_host.txt", FileAccess.READ)
if kf != null:
kf.close()
await create_timer(1.0).timeout
host_dead = true
break
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
if not host_dead:
_write_status("PHASE5_SKIP")
_log("host didn't die — skipping migration")
_peer.close()
_report()
quit()
return
# Re-elect from remaining peers
var remaining: Array[String] = []
for pk: String in _all_pks:
if pk != _host_pk:
remaining.append(pk)
remaining.sort()
var new_host_pk: String = remaining[0]
var am_new_host: bool = (new_host_pk == _my_pk_hex)
_log("re-election: new host=%s, I am %s" % [new_host_pk.left(12), "HOST" if am_new_host else "CLIENT"])
# Mesh is still alive (only old host left)
_test("Migration — remaining peers connected", _connected_peers.size() >= 1)
# ---- Phase 6: Verify persistence ----
var score_file := FileAccess.open("/tmp/layer6_durable_score_host.txt", FileAccess.READ)
if score_file != null:
var score := score_file.get_as_text().strip_edges()
score_file.close()
_test("Durable state persisted", score == "100")
else:
_test("Durable state persisted", false)
# New host sets new score
if am_new_host and _connected_peers.size() >= 1:
_write_file("/tmp/layer6_durable_score_new.txt", "200")
var score_pkt := '{"type":"new_score","score":200}'.to_utf8_buffer()
_peer.set_target_peer(0)
_peer.set_transfer_mode(MultiplayerPeer.TRANSFER_MODE_RELIABLE)
_peer.put_packet(score_pkt)
# Wait for new score
elapsed = 0.0
while elapsed < 5.0:
_peer.poll()
while _peer.get_available_packet_count() > 0:
var pkt: PackedByteArray = _peer.get_packet()
var text := pkt.get_string_from_utf8()
_log("p6 received: %s" % text.left(80))
var nsf := FileAccess.open("/tmp/layer6_durable_score_new.txt", FileAccess.READ)
if nsf != null:
var new_score := nsf.get_as_text().strip_edges()
nsf.close()
if new_score == "200":
_test("New host score written", true)
break
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
_write_status("PASS")
_log("all phases complete")
_peer.close()
_report()
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 _test(name: String, condition: bool) -> void:
_total += 1
if condition:
_pass += 1
print(" PASS %s" % name)
else:
_fail_count += 1
print(" FAIL %s" % name)
func _report() -> void:
print("")
print("=== Layer 6 Player %d: %d/%d passed ===" % [_index, _pass, _total])
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/layer6_player_%d_status.txt" % _index, status)
func _log(msg: String) -> void:
print("[P%d] %s" % [_index, msg])
func _fail(msg: String) -> void:
_log("FAIL: %s" % msg)
_write_status("FAIL: %s" % msg)
if _peer:
_peer.close()
_report()
quit()