extends SceneTree
## Layer 5 — Peer A (Writer): GameState Router over P2P
##
## Tests the GameState router splitting state between REALTIME (@rpc over P2P UDX)
## and DURABLE (keet-daemon Autobase). Two peers via swarm_join mesh.
## Peer A writes state, Peer B verifies.
const TIMEOUT := 20.0
const STATUS_FILE := "/tmp/layer5_peer_a_status.txt"
const POLL_HZ := 0.05
var _peer: Object
var _connected := false
func _init() -> void:
_write_status("STARTING")
_start.call_deferred()
func _start() -> void:
# 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"]
print("[A] dht pk: %s" % dht_pk.hex_encode().left(16))
_peer.peer_connected.connect(func(id: int) -> void:
print("[A] peer_connected: %d" % id)
_connected = true
)
# 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
print("[A] DHT active")
# Sync: wait for peer B to be ready
_write_file("/tmp/layer5_peer_a_ready.txt", "ready")
elapsed = 0.0
while elapsed < 15.0:
if FileAccess.file_exists("/tmp/layer5_peer_b_ready.txt"):
break
_peer.poll()
await create_timer(0.1).timeout
elapsed += 0.1
print("[A] both peers ready, joining swarm")
# Swarm join
var topic := "test-gamestate-l5".sha256_buffer()
err = _peer.swarm_join(topic, true, true)
if err != OK:
_fail("swarm_join: %d" % err)
return
_write_status("JOINING")
print("[A] swarm_join called, waiting for peer B...")
# Wait for peer B to connect
elapsed = 0.0
while elapsed < TIMEOUT and not _connected:
_peer.poll()
while _peer.get_available_packet_count() > 0:
_peer.get_packet() # drain
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
if not _connected:
_fail("Peer B never connected")
return
print("[A] peer B connected, starting state tests...")
# Sync: wait for peer B to also be connected before sending
_write_file("/tmp/layer5_peer_a_connected.txt", "connected")
elapsed = 0.0
while elapsed < 10.0:
if FileAccess.file_exists("/tmp/layer5_peer_b_connected.txt"):
break
_peer.poll()
while _peer.get_available_packet_count() > 0:
_peer.get_packet()
await create_timer(0.1).timeout
elapsed += 0.1
await create_timer(0.3).timeout
# ---- Test 1: REALTIME state (via raw packet) ----
var rt_msg := '{"type":"realtime","key":"player:pos:1","value":"Vector3(1,2,3)"}'.to_utf8_buffer()
_peer.set_target_peer(0)
_peer.set_transfer_mode(MultiplayerPeer.TRANSFER_MODE_RELIABLE)
_peer.put_packet(rt_msg)
print("[A] sent realtime state: player:pos:1")
await create_timer(0.3).timeout
_peer.poll()
# ---- Test 2: DURABLE state (via keet-daemon if available) ----
var has_daemon := FileAccess.file_exists("/tmp/keet-daemon.sock")
if has_daemon:
print("[A] keet-daemon available — testing durable state")
var dur_msg := '{"type":"durable","key":"player:inv","value":"sword:1"}'.to_utf8_buffer()
_peer.set_target_peer(0)
_peer.set_transfer_mode(MultiplayerPeer.TRANSFER_MODE_RELIABLE)
_peer.put_packet(dur_msg)
print("[A] sent durable notification: player:inv")
else:
print("[A] no keet-daemon — skipping durable tests")
var skip_msg := '{"type":"skip_durable"}'.to_utf8_buffer()
_peer.set_target_peer(0)
_peer.set_transfer_mode(MultiplayerPeer.TRANSFER_MODE_RELIABLE)
_peer.put_packet(skip_msg)
await create_timer(0.3).timeout
_peer.poll()
# ---- Test 3: BOTH channel (realtime + durable) ----
var both_msg := '{"type":"both","key":"world:wall","value":"pos:5,5,5"}'.to_utf8_buffer()
_peer.set_target_peer(0)
_peer.set_transfer_mode(MultiplayerPeer.TRANSFER_MODE_RELIABLE)
_peer.put_packet(both_msg)
print("[A] sent both-channel state: world:wall")
await create_timer(0.3).timeout
_peer.poll()
# ---- Test 4: Promote (REALTIME → DURABLE) ----
var promote_msg := '{"type":"promote","key":"player:pos:1"}'.to_utf8_buffer()
_peer.set_target_peer(0)
_peer.set_transfer_mode(MultiplayerPeer.TRANSFER_MODE_RELIABLE)
_peer.put_packet(promote_msg)
print("[A] sent promote: player:pos:1")
await create_timer(0.3).timeout
_peer.poll()
# ---- Test 5: Demote (DURABLE → REALTIME) ----
var demote_msg := '{"type":"demote","key":"world:wall"}'.to_utf8_buffer()
_peer.set_target_peer(0)
_peer.set_transfer_mode(MultiplayerPeer.TRANSFER_MODE_RELIABLE)
_peer.put_packet(demote_msg)
print("[A] sent demote: world:wall")
await create_timer(0.3).timeout
_peer.poll()
# ---- Test 6: Role check ----
var role_msg := '{"type":"role_check","min_write_role":"admin","writer_role":"peer"}'.to_utf8_buffer()
_peer.set_target_peer(0)
_peer.set_transfer_mode(MultiplayerPeer.TRANSFER_MODE_RELIABLE)
_peer.put_packet(role_msg)
print("[A] sent role check")
await create_timer(0.3).timeout
_peer.poll()
# ---- Test 7: End signal ----
var end_msg := '{"type":"end"}'.to_utf8_buffer()
_peer.set_target_peer(0)
_peer.set_transfer_mode(MultiplayerPeer.TRANSFER_MODE_RELIABLE)
_peer.put_packet(end_msg)
print("[A] sent end signal")
# Wait for peer B to confirm
elapsed = 0.0
while elapsed < 10.0:
_peer.poll()
while _peer.get_available_packet_count() > 0:
var pkt: PackedByteArray = _peer.get_packet()
var text := pkt.get_string_from_utf8()
print("[A] received: %s" % text.left(80))
if text.find('"ack":"done"') >= 0:
_write_status("PASS")
print("[A] SUCCESS — peer B confirmed all tests")
_peer.close()
quit()
return
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
_write_status("PASS") # A's job is to send — B reports results
print("[A] done sending (timeout waiting for ack)")
_peer.close()
quit()
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(STATUS_FILE, status)
func _fail(msg: String) -> void:
print("[A] FAIL: %s" % msg)
_write_status("FAIL: %s" % msg)
if _peer:
_peer.close()
quit()