extends SceneTree
## Layer 9 — Mesh Mode + Role-Based Authority Election
##
## Tests Step 6 changes:
## - Phase 1: swarm_join → full mesh, verify is_server() == false for ALL peers
## - Phase 2: Role exchange via RPC (roles read from file, not daemon)
## - Phase 3: Authority election — admin (peer 0) wins
## - Phase 4: Admin disconnects → re-election → mod (peer 1) wins
##
## Usage: godot4 --headless --path project --script res://layer9_mesh_authority.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 _my_role := ""
var _bootstrap: PackedStringArray
var _connected_peers: Dictionary = {} # peer_id -> true
var _peer_registry: Dictionary = {} # pk_hex -> {peer_id, role, role_priority}
var _authority_pk := ""
var _roles_received: Dictionary = {} # pk_hex -> role (from RPC)
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()
# Read assigned role
var rf := FileAccess.open("/tmp/layer9_peer_%d_role.txt" % _peer_index, FileAccess.READ)
if rf == null:
_fail("Cannot read role assignment")
return
_my_role = rf.get_as_text().strip_edges()
rf.close()
_log("assigned role: %s" % _my_role)
# 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/layer9_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 ready before swarm_join
_write_file("/tmp/layer9_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 readyf := FileAccess.open("/tmp/layer9_peer_%d_ready.txt" % pi, FileAccess.READ)
if readyf == null:
all_ready = false
break
readyf.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: Mesh Connect + is_server Check ----
_write_status("PHASE1")
var topic := "test-mesh-authority-l9".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
elapsed = 0.0
while elapsed < TIMEOUT and _connected_peers.size() < PEER_COUNT - 1:
_peer.poll()
_drain_and_handle_packets()
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
if _connected_peers.size() < PEER_COUNT - 1:
_fail("Phase 1: only %d/%d peers connected" % [_connected_peers.size(), PEER_COUNT - 1])
return
# CRITICAL Step 6 check: is_server must be false for ALL swarm peers
var stats: Dictionary = _peer.get_stats()
var is_server: bool = stats.get("is_server", false)
_write_file("/tmp/layer9_peer_%d_is_server.txt" % _peer_index, str(is_server).to_lower())
if is_server:
_fail("Phase 1: is_server = true (Step 6 mesh mode broken)")
return
_log("Phase 1 PASS — full mesh (%d connections), is_server=false" % _connected_peers.size())
_write_status("PHASE1_OK")
# ---- Phase 2: Role Exchange via RPC ----
_write_status("PHASE2")
# Register self in registry
_peer_registry[_my_pk_hex] = {
"peer_id": 1, # self
"role": _my_role,
"role_priority": _role_priority(_my_role)
}
_roles_received[_my_pk_hex] = _my_role
# Send role announcement to all connected peers
# Format: "role:<pk_hex>:<role>"
var role_msg := ("role:%s:%s" % [_my_pk_hex, _my_role]).to_utf8_buffer()
_peer.set_target_peer(0)
_peer.set_transfer_mode(MultiplayerPeer.TRANSFER_MODE_RELIABLE)
_peer.put_packet(role_msg)
_log("sent role announcement: %s=%s" % [_my_pk_hex.left(12), _my_role])
# Wait for role announcements from all other peers
elapsed = 0.0
while elapsed < TIMEOUT and _roles_received.size() < PEER_COUNT:
_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(role_msg)
while _peer.get_available_packet_count() > 0:
var pkt: PackedByteArray = _peer.get_packet()
var text: String = pkt.get_string_from_utf8()
if text.begins_with("role:"):
_handle_role_announcement(text)
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
if _roles_received.size() < PEER_COUNT:
_fail("Phase 2: only received roles from %d/%d peers" % [_roles_received.size(), PEER_COUNT])
return
# Write received roles for orchestrator verification
var roles_str := ""
for pk_hex in _roles_received:
roles_str += "%s=%s\n" % [pk_hex.left(12), _roles_received[pk_hex]]
_write_file("/tmp/layer9_peer_%d_roles.txt" % _peer_index, roles_str)
_log("Phase 2 PASS — all %d roles received" % _roles_received.size())
_write_status("PHASE2_OK")
# Phase 2 sync: wait for all peers to complete role exchange before proceeding
# Prevents fast peers from racing through Phase 3/4 while slow peers are still
# in Phase 2 (which would cause Phase 4 sync timeout → premature disconnect).
elapsed = 0.0
while elapsed < 10.0:
var all_phase2 := true
for pi in range(PEER_COUNT):
var status := ""
var sf := FileAccess.open("/tmp/layer9_peer_%d_status.txt" % pi, FileAccess.READ)
if sf != null:
status = sf.get_as_text().strip_edges()
sf.close()
if status != "PHASE2_OK" and status != "PHASE3" and status != "PHASE3_OK" and status != "PHASE4" and status != "PASS":
all_phase2 = false
break
if all_phase2:
break
_peer.poll()
_drain_and_handle_packets()
await create_timer(0.1).timeout
elapsed += 0.1
# ---- Phase 3: Authority Election ----
_write_status("PHASE3")
_elect_authority()
if _authority_pk.is_empty():
_fail("Phase 3: authority election returned empty")
return
_write_file("/tmp/layer9_peer_%d_authority.txt" % _peer_index, _authority_pk)
# Read peer 0's pk to verify admin won
var pk0 := ""
for _attempt in range(20):
var pk0f := FileAccess.open("/tmp/layer9_peer_0_pk.txt", FileAccess.READ)
if pk0f != null:
pk0 = pk0f.get_as_text().strip_edges()
pk0f.close()
break
await create_timer(0.1).timeout
if _authority_pk != pk0:
_fail("Phase 3: authority=%s, expected admin pk=%s" % [_authority_pk.left(12), pk0.left(12)])
return
_log("Phase 3 PASS — authority is admin (peer 0): %s" % _authority_pk.left(12))
_write_status("PHASE3_OK")
# ---- Phase 4: Authority Transfer ----
# Sync point: all peers must reach Phase 4 before admin disconnects
_write_file("/tmp/layer9_peer_%d_phase4.txt" % _peer_index, "ready")
_log("waiting for all peers to reach Phase 4...")
elapsed = 0.0
while elapsed < 10.0:
var all_phase4 := true
for pi in range(PEER_COUNT):
var pf := FileAccess.open("/tmp/layer9_peer_%d_phase4.txt" % pi, FileAccess.READ)
if pf == null:
all_phase4 = false
break
pf.close()
if all_phase4:
break
_peer.poll()
_drain_and_handle_packets()
await create_timer(0.1).timeout
elapsed += 0.1
_log("all peers in Phase 4")
_write_status("PHASE4")
if _peer_index == 0:
# Admin peer: disconnect to trigger re-election
_log("Phase 4: admin disconnecting...")
await create_timer(1.0).timeout # brief pause so peers are ready
_peer.close()
_write_file("/tmp/layer9_peer_0_authority_post.txt", "disconnected")
_write_status("PASS")
_log("ALL PHASES PASSED (admin — disconnected for Phase 4)")
quit()
return
# Non-admin peers: wait for admin to disconnect
_log("Phase 4: waiting for admin (peer 0) to disconnect...")
elapsed = 0.0
var admin_disconnected := false
while elapsed < 20.0:
_peer.poll()
_drain_and_handle_packets()
# Check if admin's peer_id is gone from connected_peers
# We detect this via the peer_disconnected signal handler
if _connected_peers.size() < PEER_COUNT - 1:
admin_disconnected = true
break
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
if not admin_disconnected:
_fail("Phase 4: admin did not disconnect within 20s")
return
_log("Phase 4: admin disconnected, re-electing...")
# Remove admin from registry (find by role)
var admin_pk := ""
for pk_hex in _peer_registry:
if _peer_registry[pk_hex]["role"] == "admin":
admin_pk = pk_hex
break
if not admin_pk.is_empty():
_peer_registry.erase(admin_pk)
_roles_received.erase(admin_pk)
# Re-elect authority
_elect_authority()
if _authority_pk.is_empty():
_fail("Phase 4: re-election returned empty")
return
_write_file("/tmp/layer9_peer_%d_authority_post.txt" % _peer_index, _authority_pk)
# Read peer 1's pk to verify mod won
var pk1 := ""
for _attempt in range(20):
var pk1f := FileAccess.open("/tmp/layer9_peer_1_pk.txt", FileAccess.READ)
if pk1f != null:
pk1 = pk1f.get_as_text().strip_edges()
pk1f.close()
break
await create_timer(0.1).timeout
if _authority_pk != pk1:
_fail("Phase 4: new authority=%s, expected mod pk=%s" % [_authority_pk.left(12), pk1.left(12)])
return
_log("Phase 4 PASS — new authority is mod (peer 1): %s" % _authority_pk.left(12))
_write_status("PASS")
_log("ALL PHASES PASSED")
_peer.close()
quit()
# ---- Role Handling ----
func _handle_role_announcement(text: String) -> void:
# Format: "role:<pk_hex>:<role>"
var parts := text.split(":")
if parts.size() < 3:
return
var pk_hex := parts[1]
var role := parts[2]
if pk_hex == _my_pk_hex:
return # skip self
if _roles_received.has(pk_hex):
return # already known
_roles_received[pk_hex] = role
_log("role received: %s=%s (total: %d/%d)" % [pk_hex.left(12), role, _roles_received.size(), PEER_COUNT])
# Find peer_id for this pk (we don't have pk-to-peer_id mapping from swarm,
# so we use the order of connected peers)
var peer_id := 0
for pid in _connected_peers:
peer_id = pid
if not _peer_id_in_registry(pid):
break
_peer_registry[pk_hex] = {
"peer_id": peer_id,
"role": role,
"role_priority": _role_priority(role)
}
func _peer_id_in_registry(pid: int) -> bool:
for pk_hex in _peer_registry:
if _peer_registry[pk_hex]["peer_id"] == pid:
return true
return false
# ---- Authority Election (matches hyper-tps menu_main.gd) ----
func _elect_authority() -> void:
var candidates: Array = []
# Include self
if _my_role != "read-only" and not _my_pk_hex.is_empty():
candidates.append({
"pk": _my_pk_hex,
"role_priority": _role_priority(_my_role)
})
# Include all registered peers
for pk_hex in _peer_registry:
if pk_hex == _my_pk_hex:
continue
var entry: Dictionary = _peer_registry[pk_hex]
if entry["role"] != "read-only":
candidates.append({
"pk": pk_hex,
"role_priority": entry["role_priority"]
})
# Sort: lowest role_priority first (admin=1), then lowest pk (deterministic)
candidates.sort_custom(func(a, b):
if a["role_priority"] != b["role_priority"]:
return a["role_priority"] < b["role_priority"]
return a["pk"] < b["pk"])
if candidates.is_empty():
_authority_pk = ""
return
_authority_pk = candidates[0]["pk"]
_log("authority elected: %s (priority=%d)" % [_authority_pk.left(12), candidates[0]["role_priority"]])
func _drain_and_handle_packets() -> void:
while _peer.get_available_packet_count() > 0:
var pkt: PackedByteArray = _peer.get_packet()
var text: String = pkt.get_string_from_utf8()
if text.begins_with("role:"):
_handle_role_announcement(text)
static func _role_priority(role: String) -> int:
match role:
"admin": return 1
"mod": return 2
"peer": return 3
_: return 99
# ---- Signal Handlers ----
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)
# ---- File I/O ----
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/layer9_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()