extends SceneTree
## Layer 11 — In-Process Multi-Peer Test
##
## Models JS Hyperswarm test pattern: multiple swarm instances in a single process.
## Eliminates subprocess spawning, file-based sync, and inter-process timing issues.
## Each test creates N HyperswarmMultiplayerPeer instances, joins same topic, verifies mesh.
##
## Usage: godot4 --headless --path project --script res://layer11_inproc.gd
const POLL_HZ := 0.05
const MAX_WAIT := 20.0
var _pass := 0
var _fail := 0
var _total := 0
var _bootstrap: PackedStringArray
func _init() -> void:
_start.call_deferred()
func _start() -> void:
print("=== Layer 11: In-Process Multi-Peer Tests ===")
print("")
# Read bootstrap
var f := FileAccess.open("/tmp/layer_bootstrap.txt", FileAccess.READ)
if f == null:
print("FAIL: Cannot read /tmp/layer_bootstrap.txt")
_write_status("FAIL: no bootstrap")
quit()
return
_bootstrap = PackedStringArray(f.get_as_text().strip_edges().split(","))
f.close()
# Run test suites
await _test_2peer_mesh()
await _test_3peer_mesh()
await _test_flush_semantics()
await _test_message_roundtrip()
await _test_host_election()
await _test_disconnect_detect()
# Report
print("")
print("=== Layer 11: %d/%d passed ===" % [_pass, _total])
if _fail > 0:
_write_status("FAIL: %d/%d" % [_pass, _total])
else:
_write_status("PASS")
quit()
# ---- Test: 2-Peer Mesh ----
func _test_2peer_mesh() -> void:
print("--- Test: 2-peer mesh ---")
var peers := await _create_peers(2)
if peers.is_empty():
return
var topic := "inproc-2mesh".sha256_buffer()
for p: Dictionary in peers:
p["peer"].swarm_join(topic, true, true)
var ok := await _wait_for_mesh(peers, 1)
_test("2-peer mesh connects", ok)
_close_peers(peers)
# ---- Test: 3-Peer Mesh ----
func _test_3peer_mesh() -> void:
print("--- Test: 3-peer mesh ---")
var peers := await _create_peers(3)
if peers.is_empty():
return
var topic := "inproc-3mesh".sha256_buffer()
for p: Dictionary in peers:
p["peer"].swarm_join(topic, true, true)
var ok := await _wait_for_mesh(peers, 2)
_test("3-peer mesh connects", ok)
# Verify all peers see correct connection count
if ok:
for i in peers.size():
var ss: Dictionary = peers[i]["peer"].get_swarm_status()
_test(" peer %d: peers_connected=%d" % [i, ss.get("peers_connected", 0)],
ss.get("peers_connected", 0) == 2)
_close_peers(peers)
# ---- Test: Flush Semantics ----
func _test_flush_semantics() -> void:
print("--- Test: flush semantics ---")
var peers := await _create_peers(2)
if peers.is_empty():
return
var topic := "inproc-flush".sha256_buffer()
for p: Dictionary in peers:
p["peer"].swarm_join(topic, true, true)
# Wait for mesh
var ok := await _wait_for_mesh(peers, 1)
_test("flush: peers connect", ok)
if ok:
# JS Hyperswarm: await swarm.flush() waits for connecting===0 && queue empty.
# Wait for flush state (connecting==0).
var elapsed := 0.0
var flushed := false
while elapsed < 10.0 and not flushed:
flushed = true
for p: Dictionary in peers:
p["peer"].poll()
var ss: Dictionary = p["peer"].get_swarm_status()
if ss.get("connecting", -1) != 0:
flushed = false
if not flushed:
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
for i in peers.size():
var ss: Dictionary = peers[i]["peer"].get_swarm_status()
_test(" peer %d: flushed=%s connecting=%d queue=%d" % [
i, str(ss.get("flushed", false)),
ss.get("connecting", -1), ss.get("queue_size", -1)],
ss.get("flushed", false) == true)
_test(" peer %d: peers_connected=%d" % [i, ss.get("peers_connected", 0)],
ss.get("peers_connected", 0) == 1)
_close_peers(peers)
# ---- Test: Message Round-Trip ----
func _test_message_roundtrip() -> void:
print("--- Test: message round-trip ---")
var peers := await _create_peers(2)
if peers.is_empty():
return
var topic := "inproc-msg".sha256_buffer()
for p: Dictionary in peers:
p["peer"].swarm_join(topic, true, true)
var ok := await _wait_for_mesh(peers, 1)
if not ok:
_test("message: mesh failed", false)
_close_peers(peers)
return
# Peer 0 sends to peer 1
var msg := "hello-from-0".to_utf8_buffer()
peers[0]["peer"].set_target_peer(0) # broadcast
peers[0]["peer"].set_transfer_mode(MultiplayerPeer.TRANSFER_MODE_RELIABLE)
peers[0]["peer"].put_packet(msg)
# Poll and wait for peer 1 to receive
var received := false
var elapsed := 0.0
while elapsed < 5.0 and not received:
for p: Dictionary in peers:
p["peer"].poll()
# Check peer 1 for incoming
while peers[1]["peer"].get_available_packet_count() > 0:
var pkt: PackedByteArray = peers[1]["peer"].get_packet()
var text := pkt.get_string_from_utf8()
if text == "hello-from-0":
received = true
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
_test("message: peer 1 received from peer 0", received)
# Now peer 1 sends back
if received:
var reply := "reply-from-1".to_utf8_buffer()
peers[1]["peer"].set_target_peer(0)
peers[1]["peer"].set_transfer_mode(MultiplayerPeer.TRANSFER_MODE_RELIABLE)
peers[1]["peer"].put_packet(reply)
var got_reply := false
elapsed = 0.0
while elapsed < 5.0 and not got_reply:
for p: Dictionary in peers:
p["peer"].poll()
while peers[0]["peer"].get_available_packet_count() > 0:
var pkt: PackedByteArray = peers[0]["peer"].get_packet()
var text := pkt.get_string_from_utf8()
if text == "reply-from-1":
got_reply = true
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
_test("message: peer 0 received reply from peer 1", got_reply)
_close_peers(peers)
# ---- Test: Host Election ----
func _test_host_election() -> void:
print("--- Test: host election ---")
var peers := await _create_peers(3)
if peers.is_empty():
return
var topic := "inproc-election".sha256_buffer()
for p: Dictionary in peers:
p["peer"].swarm_join(topic, true, true)
var ok := await _wait_for_mesh(peers, 2)
if not ok:
_test("election: mesh failed", false)
_close_peers(peers)
return
# Deterministic election: lowest public key hex = host
var pk_hexes: Array = []
for p: Dictionary in peers:
pk_hexes.append(p["pk_hex"])
var sorted_pks: Array = pk_hexes.duplicate()
sorted_pks.sort()
var host_pk: String = sorted_pks[0]
_test("election: host is lowest pk (%s)" % host_pk.left(12), true)
# All peers should agree
for i in peers.size():
var is_host: bool = (peers[i]["pk_hex"] == host_pk)
_test(" peer %d (%s): %s" % [i, peers[i]["pk_hex"].left(8),
"HOST" if is_host else "CLIENT"], true)
_close_peers(peers)
# ---- Test: Disconnect Detection ----
func _test_disconnect_detect() -> void:
print("--- Test: disconnect detection ---")
var peers := await _create_peers(2)
if peers.is_empty():
return
var topic := "inproc-disconn".sha256_buffer()
for p: Dictionary in peers:
p["peer"].swarm_join(topic, true, true)
var ok := await _wait_for_mesh(peers, 1)
if not ok:
_test("disconnect: mesh failed", false)
_close_peers(peers)
return
_test("disconnect: initial mesh formed", true)
# Close peer 1
peers[1]["peer"].close()
# Poll peer 0 until it detects disconnection
var detected := false
var elapsed := 0.0
while elapsed < 10.0 and not detected:
peers[0]["peer"].poll()
if peers[0]["connected"].is_empty():
detected = true
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
_test("disconnect: peer 0 detects peer 1 gone", detected)
# Only close peer 0 (peer 1 already closed)
peers[0]["peer"].close()
# ---- Helpers ----
func _create_peers(count: int) -> Array:
var peers: Array = []
for i in count:
var peer: Object = ClassDB.instantiate(&"HyperswarmMultiplayerPeer")
var kp: Dictionary = peer.generate_dht_keypair()
var pk: PackedByteArray = kp["public_key"]
var pk_hex: String = pk.hex_encode()
# Track connections per peer
var connected: Dictionary = {}
peer.peer_connected.connect(func(id: int) -> void:
connected[id] = true)
peer.peer_disconnected.connect(func(id: int) -> void:
connected.erase(id))
var err = peer.join_dht(pk, kp["secret_key"], _bootstrap)
if err != OK:
_test("create peer %d: join_dht failed (%d)" % [i, err], false)
# Clean up already-created peers
for p: Dictionary in peers:
p["peer"].close()
return []
peers.append({
"peer": peer,
"kp": kp,
"pk": pk,
"pk_hex": pk_hex,
"connected": connected
})
# Wait for all DHT active
var elapsed := 0.0
while elapsed < 10.0:
var all_active := true
for p: Dictionary in peers:
p["peer"].poll()
if not p["peer"].get_dht_status().get("active", false):
all_active = false
if all_active:
break
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
var all_active := true
for p: Dictionary in peers:
if not p["peer"].get_dht_status().get("active", false):
all_active = false
if not all_active:
_test("create %d peers: DHT not active after 10s" % count, false)
for p: Dictionary in peers:
p["peer"].close()
return []
return peers
func _wait_for_mesh(peers: Array, expected_connections: int) -> bool:
var elapsed := 0.0
while elapsed < MAX_WAIT:
var all_connected := true
for p: Dictionary in peers:
p["peer"].poll()
# Drain packets during mesh formation
while p["peer"].get_available_packet_count() > 0:
p["peer"].get_packet()
if p["connected"].size() < expected_connections:
all_connected = false
if all_connected:
return true
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
# Print diagnostics
for i in peers.size():
var ss: Dictionary = peers[i]["peer"].get_swarm_status()
print(" [peer %d] connected=%d connecting=%d queue=%d known=%d all_conn=%d" % [
i, peers[i]["connected"].size(),
ss.get("connecting", -1), ss.get("queue_size", -1),
ss.get("peers_known", 0), ss.get("all_connections", 0)])
return false
func _close_peers(peers: Array) -> void:
for p: Dictionary in peers:
p["peer"].close()
# Brief pause to let cleanup complete
await create_timer(0.2).timeout
func _test(name: String, condition: bool) -> void:
_total += 1
if condition:
_pass += 1
print(" PASS %s" % name)
else:
_fail += 1
print(" FAIL %s" % name)
func _write_status(status: String) -> void:
var wf := FileAccess.open("/tmp/layer11_status.txt", FileAccess.WRITE)
wf.store_string(status)
wf.close()