extends SceneTree
## Layer 5 — Peer B (Verifier): GameState Router over P2P
##
## Connects to Peer A via swarm_join (P2P mesh), receives state updates,
## verifies routing behavior.
## Tests: REALTIME vs DURABLE channel routing, promote/demote, role permissions.
const TIMEOUT := 20.0
const STATUS_FILE := "/tmp/layer5_peer_b_status.txt"
const POLL_HZ := 0.05
var _peer: Object
var _connected := false
var _pass := 0
var _fail_count := 0
var _total := 0
var _done := false
var _received: Array[Dictionary] = []
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("[B] dht pk: %s" % dht_pk.hex_encode().left(16))
_peer.peer_connected.connect(func(id: int) -> void:
print("[B] 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("[B] DHT active")
# Sync: signal ready and wait for peer A
_write_file("/tmp/layer5_peer_b_ready.txt", "ready")
elapsed = 0.0
while elapsed < 15.0:
if FileAccess.file_exists("/tmp/layer5_peer_a_ready.txt"):
break
_peer.poll()
await create_timer(0.1).timeout
elapsed += 0.1
print("[B] both peers ready, joining swarm")
# Swarm join — same topic as Peer A
var topic := "test-gamestate-l5".sha256_buffer()
err = _peer.swarm_join(topic, true, true)
if err != OK:
_fail("swarm_join: %d" % err)
return
print("[B] swarm_join called, waiting for peer A...")
# Wait for connection
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("Could not connect to peer A")
return
# Signal connected and wait for peer A
_write_file("/tmp/layer5_peer_b_connected.txt", "connected")
elapsed = 0.0
while elapsed < 10.0:
if FileAccess.file_exists("/tmp/layer5_peer_a_connected.txt"):
break
_peer.poll()
while _peer.get_available_packet_count() > 0:
_peer.get_packet() # drain
await create_timer(0.1).timeout
elapsed += 0.1
_write_status("CONNECTED")
print("[B] connected, receiving state tests...")
# Set up local GameState router for verification
var routes: Dictionary = {
"player:pos:*": 0, # REALTIME
"player:inv": 1, # DURABLE
"world:*": 2, # BOTH
}
# Receive and verify all test messages
var skip_durable := false
elapsed = 0.0
while elapsed < TIMEOUT and not _done:
_peer.poll()
while _peer.get_available_packet_count() > 0:
var pkt: PackedByteArray = _peer.get_packet()
var text := pkt.get_string_from_utf8()
print("[B] received: %s" % text.left(100))
var parsed: Variant = JSON.parse_string(text)
if not parsed is Dictionary:
continue
var msg: Dictionary = parsed
var msg_type: String = msg.get("type", "")
_received.append(msg)
match msg_type:
"realtime":
# Test 1: REALTIME state received via P2P packet
_test("REALTIME packet received", true)
_test("REALTIME key matches", msg.get("key", "") == "player:pos:1")
# Verify route resolves to REALTIME
var channel := _resolve_route("player:pos:1", routes)
_test("Route player:pos:* → REALTIME", channel == 0)
"durable":
# Test 2: DURABLE notification received
_test("DURABLE notification received", true)
_test("DURABLE key matches", msg.get("key", "") == "player:inv")
var channel := _resolve_route("player:inv", routes)
_test("Route player:inv → DURABLE", channel == 1)
"skip_durable":
skip_durable = true
print("[B] skipping durable tests (no daemon)")
"both":
# Test 3: BOTH channel state
_test("BOTH channel received", true)
_test("BOTH key matches", msg.get("key", "") == "world:wall")
var channel := _resolve_route("world:wall", routes)
_test("Route world:* → BOTH", channel == 2)
"promote":
# Test 4: Promote moves REALTIME → DURABLE
_test("Promote received", true)
_test("Promote key is player:pos:1", msg.get("key", "") == "player:pos:1")
# After promote, the key should route to DURABLE
routes["player:pos:1"] = 1 # Simulate promote
var channel := _resolve_route("player:pos:1", routes)
_test("After promote → DURABLE", channel == 1)
"demote":
# Test 5: Demote moves DURABLE → REALTIME
_test("Demote received", true)
_test("Demote key is world:wall", msg.get("key", "") == "world:wall")
# After demote, should route to REALTIME
routes["world:wall"] = 0 # Simulate demote
var channel := _resolve_route("world:wall", routes)
_test("After demote → REALTIME", channel == 0)
"role_check":
# Test 6: Role check — admin required, peer can't write
var min_role: String = msg.get("min_write_role", "peer")
var writer_role: String = msg.get("writer_role", "peer")
var can_write := _check_role(min_role, writer_role)
_test("Role check: admin required, peer rejected", not can_write)
"end":
# Send ack back
var ack := '{"ack":"done","pass":%d,"fail":%d,"total":%d}' % [_pass, _fail_count, _total]
var ack_buf := ack.to_utf8_buffer()
_peer.set_target_peer(0)
_peer.set_transfer_mode(MultiplayerPeer.TRANSFER_MODE_RELIABLE)
_peer.put_packet(ack_buf)
_done = true
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
# Report
print("")
print("=== Layer 5 (Peer B): %d/%d passed ===" % [_pass, _total])
if _fail_count > 0:
_write_status("FAIL: %d/%d" % [_pass, _total])
else:
_write_status("PASS")
_peer.close()
quit()
func _resolve_route(key: String, routes: Dictionary) -> int:
# Check exact match first
if routes.has(key):
return routes[key]
# Prefix match
var best_len := 0
var best_ch := 0 # Default: REALTIME
for pattern: String in routes:
if pattern.ends_with("*"):
var prefix: String = pattern.left(pattern.length() - 1)
if key.begins_with(prefix) and prefix.length() > best_len:
best_len = prefix.length()
best_ch = routes[pattern]
return best_ch
func _check_role(min_role: String, writer_role: String) -> bool:
if min_role == "peer":
return true
match min_role:
"admin":
return writer_role == "admin"
"mod":
return writer_role == "admin" or writer_role == "mod"
return true
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 _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("[B] FAIL: %s" % msg)
_write_status("FAIL: %s" % msg)
if _peer:
_peer.close()
quit()