extends SceneTree
## Layer 12 — Content-Addressed Transfer (HyperOverlay)
##
## Built on Layer 11's pattern, deliberately: peers connect by joining a DHT
## TOPIC against a testnet bootstrap. An earlier attempt at this test invented its
## own harness with create_server/create_client on loopback and no bootstrap,
## which cannot connect — and then spent several rounds blaming the overlay for
## it. The connection method is not a detail; it is the thing under test's floor.
##
## Sequence: mesh first, assert it, THEN attach the overlay. If the mesh check
## fails, nothing below it is meaningful and the run says so rather than
## reporting an overlay failure.
##
## Usage: godot4 --headless --path project --script res://layer12_overlay.gd
const POLL_HZ := 0.05
const MESH_WAIT := 25.0
const XFER_WAIT := 40.0
var _pass := 0
var _fail := 0
var _bootstrap: PackedStringArray
func _init() -> void:
_start.call_deferred()
func _test(name: String, ok: bool) -> void:
if ok: _pass += 1; print(" PASS ", name)
else: _fail += 1; print(" FAIL ", name)
func _start() -> void:
print("=== Layer 12: Content-Addressed Transfer ===")
var f := FileAccess.open("/tmp/layer_bootstrap.txt", FileAccess.READ)
if f == null:
print("FAIL: Cannot read /tmp/layer_bootstrap.txt")
_finish("FAIL: no bootstrap")
return
_bootstrap = PackedStringArray(f.get_as_text().strip_edges().split(","))
f.close()
await _test_fetch_between_two_peers()
_finish("")
func _make_peer() -> Object:
var peer: Object = ClassDB.instantiate(&"HyperswarmMultiplayerPeer")
var kp: Dictionary = peer.generate_dht_keypair()
if peer.join_dht(kp["public_key"], kp["secret_key"], _bootstrap) != OK:
return null
return peer
func _pump(peers: Array, overlays: Array) -> void:
for p in peers: p.poll()
for o in overlays: o.poll()
func _test_fetch_between_two_peers() -> void:
print("--- Test: fetch a file from a peer ---")
var dir := "/tmp/layer12"
DirAccess.make_dir_recursive_absolute(dir + "/a")
DirAccess.make_dir_recursive_absolute(dir + "/b")
# ~400 KB of deterministic noise: several chunks, so this exercises the chunk
# scheduler rather than a single-chunk special case.
var src := dir + "/asset.bin"
var data := PackedByteArray()
data.resize(400 * 1024)
var x := 987654321
for i in data.size():
x = (x * 1103515245 + 12345) & 0x7fffffff
data[i] = (x >> 16) & 0xff
var wf := FileAccess.open(src, FileAccess.WRITE)
wf.store_buffer(data)
wf.close()
var a := _make_peer()
var b := _make_peer()
if a == null or b == null:
_test("join_dht for both peers", false)
return
_test("join_dht for both peers", true)
# WAIT FOR THE DHT TO COME UP BEFORE JOINING A TOPIC. Layer 11 does this and
# it is not optional: swarm_join on a peer whose DHT is not yet active
# announces and looks up against nothing, and the mesh silently never forms.
# Skipping it is what made the first run of this layer fail.
var dht_elapsed := 0.0
var dht_up := false
while dht_elapsed < 10.0:
a.poll(); b.poll()
if a.get_dht_status().get("active", false) and b.get_dht_status().get("active", false):
dht_up = true
break
await create_timer(POLL_HZ).timeout
dht_elapsed += POLL_HZ
_test("both DHTs active (%.1fs)" % dht_elapsed, dht_up)
if not dht_up:
a.close(); b.close()
return
var topic := "layer12-overlay".sha256_buffer()
a.swarm_join(topic, true, true)
b.swarm_join(topic, true, true)
# ---- floor: the mesh must form before anything else is meaningful ----
var waited := 0.0
var meshed := false
while waited < MESH_WAIT:
_pump([a, b], [])
var sa: Dictionary = a.get_swarm_status()
var sb: Dictionary = b.get_swarm_status()
if sa.get("peers_connected", 0) >= 1 and sb.get("peers_connected", 0) >= 1:
meshed = true
break
await create_timer(POLL_HZ).timeout
waited += POLL_HZ
_test("peers meshed over the topic (%.1fs)" % waited, meshed)
if not meshed:
print(" (mesh never formed — overlay results below would be meaningless)")
a.close(); b.close()
return
# ---- now the overlay ----
var ov_a: Object = ClassDB.instantiate(&"HyperOverlay") # holder
var ov_b: Object = ClassDB.instantiate(&"HyperOverlay") # fetcher
_test("holder overlay opens", ov_a.open(dir + "/a"))
_test("fetcher overlay opens", ov_b.open(dir + "/b"))
_test("holder attaches to its peer", ov_a.attach(a))
_test("fetcher attaches to its peer", ov_b.attach(b))
# ATTACHING TO AN ALREADY-MESHED SWARM MUST WORK IMMEDIATELY, and this window is
# what asserts it. The sequence above — mesh, assert the mesh, then attach — is
# the one this file recommends and the one a game will use, so a peer that is
# already connected when the overlay attaches has to be picked up there and then.
#
# This was 30.0s, and that window was hiding a bug rather than measuring a
# property. The measured gap (19.4s one run, never within 30s the next) is not a
# latency at all: the FIRST UDX connection is dead on arrival — it addresses a
# stream id that was never allocated, so both peers send a Noise header, receive
# none, and close with -110 after ~15s. Only the retry carries anything. See
# hyper-games #15 for the trace.
#
# 5s is deliberately far below what that retry can supply and far above what a
# working connection needs, so this assertion distinguishes the two mechanisms
# rather than timing one of them. IT IS EXPECTED TO FAIL until #15 is fixed —
# that is the point of it. Do not widen this window to make the suite green.
waited = 0.0
while waited < 5.0 and (ov_a.peer_count() == 0 or ov_b.peer_count() == 0):
_pump([a, b], [ov_a, ov_b])
await create_timer(POLL_HZ).timeout
waited += POLL_HZ
_test("overlay channel on the holder (%d)" % ov_a.peer_count(), ov_a.peer_count() > 0)
_test("overlay channel on the fetcher (%d)" % ov_b.peer_count(), ov_b.peer_count() > 0)
print(" [overlay channels opened %.1fs after the mesh formed]" % waited)
var oid: String = ov_a.content_hash_of(src)
_test("holder hashes the asset (%s...)" % oid.substr(0, 12), oid.length() == 64)
ov_a.serve(oid, src)
var result := {"path": "", "err": ""}
ov_b.content_fetched.connect(func(_h: String, p: String) -> void: result.path = p)
ov_b.content_failed.connect(func(_h: String, r: String) -> void: result.err = r)
var dest := dir + "/b/fetched.bin"
DirAccess.remove_absolute(dest)
ov_b.fetch(oid, dest)
waited = 0.0
while waited < XFER_WAIT and result.path == "" and result.err == "":
_pump([a, b], [ov_a, ov_b])
await create_timer(POLL_HZ).timeout
waited += POLL_HZ
_test("fetch did not fail (%s)" % result.err, result.err == "")
_test("fetch completed in %.1fs" % waited, result.path == dest)
if FileAccess.file_exists(dest):
var rf := FileAccess.open(dest, FileAccess.READ)
var got := rf.get_buffer(rf.get_length())
rf.close()
_test("%d bytes arrived" % got.size(), got.size() == data.size())
_test("bytes are identical to the original", got == data)
else:
_test("destination file exists", false)
ov_b.close(); ov_a.close()
ov_a.free(); ov_b.free()
a.close(); b.close()
func _finish(err: String) -> void:
if err != "":
print("=== Layer 12: FAILED (%s) ===" % err)
_write_status("FAIL")
quit(1)
return
print("\n %d pass, %d fail" % [_pass, _fail])
var ok := _fail == 0
print("=== Layer 12: %s ===" % ("PASS" if ok else "FAILED"))
_write_status("PASS" if ok else "FAIL")
quit(0 if ok else 1)
func _write_status(s: String) -> void:
var f := FileAccess.open("/tmp/layer12_status.txt", FileAccess.WRITE)
if f: f.store_string(s); f.close()