extends SceneTree
## Gate: PerfMetrics' network sampling — spec/bot-testbed.md (b).
##
## Pure where it can be: percentile/jitter are arithmetic and must be checked on
## KNOWN inputs, because an instrument that returns a plausible number for every
## input is not evidence. The recorder is checked by writing a real file.
var ok := true
func _p(c: bool, m: String) -> void:
print(("ok " if c else "FAIL: ") + m)
if not c: ok = false
func _initialize() -> void:
var pm: Node = load("res://autoloads/perf_metrics.gd").new()
root.add_child(pm)
# --- percentile, on inputs whose answer is arithmetic, not opinion ---
_p(pm._percentile([1, 2, 3, 4, 5], 0.50) == 3, "p50 of 1..5 == 3")
_p(pm._percentile([1, 2, 3, 4, 5], 0.95) == 5, "p95 of 1..5 == 5")
_p(pm._percentile([5, 1, 4, 2, 3], 0.50) == 3, "p50 sorts first (unsorted input)")
_p(pm._percentile([], 0.5) == 0, "empty ring -> 0, not a crash")
_p(pm._percentile([7], 0.95) == 7, "single sample -> itself")
# --- jitter = mean absolute deviation ---
_p(pm._jitter([10, 10, 10]) == 0, "flat link -> 0 jitter")
_p(pm._jitter([0, 20]) == 10, "0/20 -> 10 (mean abs deviation)")
_p(pm._jitter([5]) == 0, "single sample -> 0, no divide-by-zero")
# --- snapshot shape, with no mesh at all ---
var snap: Dictionary = pm.snapshot()
for k in ["uptime_ms", "peers", "joined_total", "left_total", "first_peer_ms", "rtt"]:
if not snap.has(k):
_p(false, "snapshot missing key: " + k)
_p(snap.has("rtt") and (snap["rtt"] is Dictionary), "snapshot has an rtt map")
_p(int(snap["peers"]) == 0, "no mesh -> peers 0 (not a crash)")
_p(int(snap["first_peer_ms"]) == -1, "no peer yet -> first_peer_ms -1")
# --- the recorder actually writes NDJSON ---
var path := "user://gate_netstats.ndjson"
OS.set_environment("HYPER_TPS_NETLOG", ProjectSettings.globalize_path(path))
pm._net_record("peer_up", {"peer": 7, "peers": 1})
pm._net_record("rtt", {"peer": 7, "p50": 42})
# DO NOT FLUSH HERE. This gate used to call pm._net_log.flush() before reading,
# and that is exactly what let the flush-every-16 bug ship: two records buffered
# and the gate flushed them itself, so the file always had content in the test
# and 0 bytes in a real 5-minute soak. Reading without flushing is the whole
# assertion -- the recorder must make each record durable on its own.
var f := FileAccess.open(ProjectSettings.globalize_path(path), FileAccess.READ)
if f == null:
_p(false, "net log was not created")
else:
var lines: Array[String] = []
while not f.eof_reached():
var l := f.get_line()
if l.strip_edges() != "":
lines.append(l)
f.close()
_p(lines.size() == 2, "two records written (got %d)" % lines.size())
var first: Variant = JSON.parse_string(lines[0]) if lines.size() > 0 else null
_p(first is Dictionary, "record is valid JSON (NDJSON, one object per line)")
if first is Dictionary:
_p(String((first as Dictionary).get("kind", "")) == "peer_up", "kind tagged")
_p((first as Dictionary).has("t"), "record carries a relative timestamp")
# --- churn counters move on the signals, not on a timer ---
pm._on_net_peer_connected(3)
_p(int(pm.snapshot()["joined_total"]) == 1, "peer_connected increments joins")
_p(int(pm.snapshot()["first_peer_ms"]) >= 0, "first_peer_ms recorded on first join")
pm._on_net_peer_disconnected(3)
_p(int(pm.snapshot()["left_total"]) == 1, "peer_disconnected increments leaves")
print("GATE net_stats: %s" % ("PASS" if ok else "FAIL"))
quit(0 if ok else 1)