extends SceneTree
## Layer 13 — Real assets over the mesh (textures, models, audio) (HyperOverlay)
##
## Layer 12 proves bytes move: 400 KB of synthetic noise, two peers, compared for
## equality. That is the transport floor and it is not the same claim as "the game
## can send its textures to a peer". This layer makes that second claim, and the
## difference is deliberate in three places:
##
## 1. REAL ASSETS. The files are the game's own PBR maps and a JPEG, at their
## real sizes (MBs, not KB), so the chunk scheduler is exercised across
## hundreds of chunks per asset rather than a handful.
## 2. THE ARRIVAL MUST DECODE. Every fetched file is run through
## Image.load_*_from_buffer and compared to the source image's dimensions and
## decoded pixels. A byte-comparison alone would pass on a file that is
## corrupt in a way only the decoder notices, and "it transferred" is not the
## property anyone wants — "it is still a texture" is.
## 3. A SET, CONCURRENTLY. A level needs its whole material set, not one file.
## All of them are requested before any of them completes.
##
## Test C is the one with no precedent anywhere in this suite: a THIRD peer
## fetches an asset AFTER the original holder has been closed. It can only be
## served by the peer that previously fetched it. That re-serve is what makes
## content distribution cost the origin less than linear, and until now it was an
## asserted-nowhere property of ContentOverlay._on_fetched.
##
## Usage: godot4 --headless --path project --script res://layer13_textures.gd
const POLL_HZ := 0.05
const MESH_WAIT := 25.0
const XFER_WAIT := 90.0
var _pass := 0
var _fail := 0
var _bootstrap: PackedStringArray
var _dir := "/tmp/layer13"
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 13: Real assets over the mesh (textures, models, audio) ===")
var f := FileAccess.open("/tmp/layer_bootstrap.txt", FileAccess.READ)
if f == null:
print("FAIL: Cannot read /tmp/layer_bootstrap.txt")
_finish("no bootstrap")
return
_bootstrap = PackedStringArray(f.get_as_text().strip_edges().split(","))
f.close()
await _run()
_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:
if p != null: p.poll()
for o in overlays:
if o != null: o.poll()
## Decode a file as an image, by extension. Returns null if it does not decode —
## which is the failure this layer exists to catch.
func _decode(path: String) -> Image:
var f := FileAccess.open(path, FileAccess.READ)
if f == null:
return null
var buf := f.get_buffer(f.get_length())
f.close()
# Sniff the MAGIC, do not trust the extension. Half the game's bridge_*.png
# files are actually JPEGs with a .png name, and dispatching on the extension
# reported them as corrupt arrivals when they had transferred perfectly — a
# test bug that reads exactly like a transport bug.
var img := Image.new()
var err := ERR_INVALID_DATA
if buf.size() >= 8 and buf[0] == 0x89 and buf[1] == 0x50 and buf[2] == 0x4E and buf[3] == 0x47:
err = img.load_png_from_buffer(buf)
elif buf.size() >= 3 and buf[0] == 0xFF and buf[1] == 0xD8 and buf[2] == 0xFF:
err = img.load_jpg_from_buffer(buf)
return img if err == OK else null
func _read_bytes(path: String) -> PackedByteArray:
var f := FileAccess.open(path, FileAccess.READ)
if f == null:
return PackedByteArray()
var b := f.get_buffer(f.get_length())
f.close()
return b
## A 4-byte big-endian read, for container headers that declare their own length.
func _be32(b: PackedByteArray, at: int) -> int:
return (b[at] << 24) | (b[at + 1] << 16) | (b[at + 2] << 8) | b[at + 3]
func _le32(b: PackedByteArray, at: int) -> int:
return b[at] | (b[at + 1] << 8) | (b[at + 2] << 16) | (b[at + 3] << 24)
## Assert a fetched file is byte-identical, and then validate it AS ITS FORMAT.
##
## Byte-identity is the transport's actual guarantee — content-addressed, verified
## per chunk against the holder's advertised blake2b — so it is the assertion that
## must hold for EVERY type. The per-format checks below exist because a file that
## compares equal but no longer parses would mean the comparison was wrong, and
## because "the overlay is type-agnostic" deserves evidence per type rather than
## one image and an argument.
func _assert_asset(label: String, src: String, dest: String) -> void:
if not FileAccess.file_exists(dest):
_test("%s: destination exists" % label, false)
return
var want := _read_bytes(src)
var got := _read_bytes(dest)
_test("%s: %d bytes arrived (of %d)" % [label, got.size(), want.size()], got.size() == want.size())
_test("%s: bytes identical" % label, got == want)
if got != want:
return
# ---- images: decode and compare pixels ----
var src_img := _decode(src)
if src_img != null:
var dst_img := _decode(dest)
if dst_img == null:
_test("%s: arrival decodes as an image" % label, false)
return
_test("%s: decodes %dx%d, pixels match source" % [label, dst_img.get_width(), dst_img.get_height()],
dst_img.get_width() == src_img.get_width()
and dst_img.get_height() == src_img.get_height()
and dst_img.get_data() == src_img.get_data())
return
# ---- glTF binary: magic + the length the container declares about itself ----
if got.size() >= 12 and got[0] == 0x67 and got[1] == 0x6C and got[2] == 0x54 and got[3] == 0x46:
var declared := _le32(got, 8)
_test("%s: glB magic intact, declared length %d == file size %d" % [label, declared, got.size()],
declared == got.size())
return
# ---- RIFF/WAVE: magic + the size field, which is total-8 ----
if got.size() >= 12 and got[0] == 0x52 and got[1] == 0x49 and got[2] == 0x46 and got[3] == 0x46:
var riff_size := _le32(got, 4)
var is_wave := got[8] == 0x57 and got[9] == 0x41 and got[10] == 0x56 and got[11] == 0x45
_test("%s: RIFF/WAVE intact, size field %d == %d" % [label, riff_size, got.size() - 8],
is_wave and riff_size == got.size() - 8)
return
# ---- MPEG audio: ID3 tag or a frame sync in the first bytes ----
if got.size() >= 3:
var id3 := got[0] == 0x49 and got[1] == 0x44 and got[2] == 0x33
var sync := got[0] == 0xFF and (got[1] & 0xE0) == 0xE0
if id3 or sync:
_test("%s: MPEG audio header intact (%s)" % [label, "ID3" if id3 else "frame sync"], true)
return
# Anything else: the byte-identity assertions above are the whole claim, and
# they are the same guarantee the typed cases rest on.
_test("%s: opaque binary, byte-identical" % label, true)
func _run() -> void:
var src_dir := _dir + "/src"
DirAccess.make_dir_recursive_absolute(_dir + "/a")
DirAccess.make_dir_recursive_absolute(_dir + "/b")
DirAccess.make_dir_recursive_absolute(_dir + "/c")
# The assets run_layer13.sh staged. Missing ones are skipped loudly rather
# than silently narrowing what this layer claims to have tested.
var assets: Array[String] = []
var d := DirAccess.open(src_dir)
if d != null:
for name in d.get_files():
assets.append(name)
assets.sort()
if assets.is_empty():
_test("staged texture assets present in %s" % src_dir, false)
return
var total_bytes := 0
for name in assets:
total_bytes += _read_bytes(src_dir.path_join(name)).size()
print("--- %d assets staged, %.1f MB total ---" % [assets.size(), total_bytes / 1048576.0])
_test("staged texture assets present (%d)" % assets.size(), true)
# ---- floor: DHT, then mesh. Same order as layer 12, for the same reason. ----
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)
var elapsed := 0.0
var dht_up := false
while 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
elapsed += POLL_HZ
_test("both DHTs active (%.1fs)" % elapsed, dht_up)
if not dht_up:
a.close(); b.close()
return
var topic := "layer13-textures".sha256_buffer()
a.swarm_join(topic, true, true)
b.swarm_join(topic, true, true)
elapsed = 0.0
var meshed := false
while elapsed < MESH_WAIT:
_pump([a, b], [])
if a.get_swarm_status().get("peers_connected", 0) >= 1 \
and b.get_swarm_status().get("peers_connected", 0) >= 1:
meshed = true
break
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
_test("peers meshed over the topic (%.1fs)" % elapsed, meshed)
if not meshed:
print(" (mesh never formed — everything below would be meaningless)")
a.close(); b.close()
return
# ---- overlays ----
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", ov_a.attach(a))
_test("fetcher attaches", ov_b.attach(b))
# 5s, for the reason layer 12 states: far below what a reconnect can supply,
# far above what a working connection needs. Do not widen it.
elapsed = 0.0
while elapsed < 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
elapsed += 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)
# ---- publish the whole set ----
var hashes := {} # name -> hash
for name in assets:
var path := src_dir.path_join(name)
var h: String = ov_a.content_hash_of(path)
if h.length() != 64:
_test("holder hashes %s" % name, false)
continue
ov_a.serve(h, path)
hashes[name] = h
_test("holder hashed and served all %d assets" % assets.size(), hashes.size() == assets.size())
# ---- Test A+B: fetch the ENTIRE set concurrently ----
print("--- fetching %d assets concurrently ---" % hashes.size())
var done := {} # hash -> path
var failed := {} # hash -> reason
ov_b.content_fetched.connect(func(h: String, p: String) -> void: done[h] = p)
ov_b.content_failed.connect(func(h: String, r: String) -> void: failed[h] = r)
var dests := {} # name -> dest path
for name: String in hashes:
var dest: String = _dir + "/b/" + name
DirAccess.remove_absolute(dest)
dests[name] = dest
ov_b.fetch(hashes[name], dest)
var t0 := Time.get_ticks_msec()
elapsed = 0.0
while elapsed < XFER_WAIT and (done.size() + failed.size()) < hashes.size():
_pump([a, b], [ov_a, ov_b])
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
var secs := (Time.get_ticks_msec() - t0) / 1000.0
_test("no fetch failed (%s)" % str(failed.values()), failed.is_empty())
_test("all %d assets completed in %.1fs" % [hashes.size(), secs], done.size() == hashes.size())
if secs > 0.0:
print(" [%.1f MB in %.1fs = %.1f MB/s]" % [total_bytes / 1048576.0, secs, (total_bytes / 1048576.0) / secs])
for name: String in dests:
_assert_asset(name, src_dir.path_join(name), dests[name])
# ---- Test C: a third peer fetches with the ORIGIN GONE ----
# The only remaining holder is B, which acquired the bytes moments ago. If the
# re-serve does not work this fetch has nobody to answer it and times out.
print("--- re-serve: third peer fetches with the origin closed ---")
var probe_name: String = assets[0]
var probe_hash: String = hashes.get(probe_name, "")
if probe_hash == "":
_test("re-serve: have a hash to probe", false)
else:
var c := _make_peer()
if c == null:
_test("third peer joins DHT", false)
else:
_test("third peer joins DHT", true)
elapsed = 0.0
var c_up := false
while elapsed < 10.0:
c.poll()
if c.get_dht_status().get("active", false):
c_up = true
break
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
_test("third peer DHT active", c_up)
c.swarm_join(topic, true, true)
# B must re-serve what it fetched. ContentOverlay does this in
# _on_fetched; here the test does it explicitly, because this layer is
# testing the overlay, not the autoload.
ov_b.serve(probe_hash, dests[probe_name])
elapsed = 0.0
var c_meshed := false
while elapsed < MESH_WAIT:
_pump([a, b, c], [ov_a, ov_b])
if c.get_swarm_status().get("peers_connected", 0) >= 1:
c_meshed = true
break
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
_test("third peer meshed (%.1fs)" % elapsed, c_meshed)
var ov_c: Object = ClassDB.instantiate(&"HyperOverlay")
_test("third overlay opens", ov_c.open(_dir + "/c"))
_test("third overlay attaches", ov_c.attach(c))
elapsed = 0.0
while elapsed < 10.0 and ov_c.peer_count() == 0:
_pump([a, b, c], [ov_a, ov_b, ov_c])
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
_test("third overlay has a channel (%d)" % ov_c.peer_count(), ov_c.peer_count() > 0)
# CLOSE THE ORIGIN. From here the bytes can only come from B.
ov_a.close()
a.close()
print(" [origin closed — only the peer that fetched it can answer]")
var c_done := {"path": "", "err": ""}
ov_c.content_fetched.connect(func(_h: String, p: String) -> void: c_done.path = p)
ov_c.content_failed.connect(func(_h: String, r: String) -> void: c_done.err = r)
var c_dest := _dir + "/c/" + probe_name
DirAccess.remove_absolute(c_dest)
ov_c.fetch(probe_hash, c_dest)
elapsed = 0.0
while elapsed < XFER_WAIT and c_done.path == "" and c_done.err == "":
_pump([b, c], [ov_b, ov_c])
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
_test("re-serve: fetch did not fail (%s)" % c_done.err, c_done.err == "")
_test("re-serve: completed in %.1fs" % elapsed, c_done.path == c_dest)
_assert_asset("re-serve/" + probe_name, src_dir.path_join(probe_name), c_dest)
ov_c.close(); ov_c.free()
c.close()
ov_b.close()
if is_instance_valid(ov_a): ov_a.free()
ov_b.free()
b.close()
func _finish(err: String) -> void:
if err != "":
print("=== Layer 13: FAILED (%s) ===" % err)
_write_status("FAIL")
quit(1)
return
print("\n %d pass, %d fail" % [_pass, _fail])
var ok := _fail == 0
print("=== Layer 13: %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/layer13_status.txt", FileAccess.WRITE)
if f: f.store_string(s); f.close()