From dcd8e4d6924a5c72a219b80713ca4863c59c8f6f Mon Sep 17 00:00:00 2001 From: Joseph <162703152+josephnef@users.noreply.github.com> Date: Mon, 3 Aug 2026 16:35:58 +0300 Subject: [PATCH 01/12] ARQ end-to-end delivery bench: three-ledger harness; count spsc-fat pool-exhaustion drops MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Field failure shape (PixelPilot, 8812AU drone -> 8812EU ground): hardware ARQ logs a packet delivered-after-retry while the RX app never receives it. The existing ACK harnesses (ack_responder_check.sh, ack_txreport_matrix.sh) judge the loop from the TX side's CCX reports alone, so this class was structurally invisible to them. tests/arq_e2e_delivery.sh + arq_e2e_analyze.py close the loop with three per-frame ledgers on one channel: DUT = 8812EU examples/duplex (the one-handle RX+TX topology, ACK responder armed, host ledger via rx.seq), drone-sim = 8812CU txdemo (retrying unicast QoS-Data, per-frame tx.report — the halmac SW_DEFINE tag echoes the payload counter mod 256, so reports join to payloads frame-exactly), witness = 8814AU rxdemo (independent on-air ledger). The ledger identity (ok - acked_undelivered - tail_suspect + dropped_but_delivered = dut_pctrs) held delta-0 on every joinable run, 129k-410k frames each. Measured, both directions: - The hardware ACK is admission-gated on the 8812EU. 150 ms consumer freezes against a ~2.3k fps flood dropped 14,214 frames inside the receiver, in 143 contiguous stall-window gaps - and every one of them was reported ok=0 and retried by the peer. Zero ACKed-but-undelivered in ~460k ok'd frames across all default-ring arms: backpressure that reaches the chip keeps ARQ truthful. - The same stalls under RxMode::SpscFat reproduce the field failure: at pool exhaustion the pump preserves its never-block invariant by re-arming the URB with the received buffer and dropping the payload - after the chip admitted and ACKed the frame. 5,667 ACKed-but-undelivered (~99% witness-confirmed on air), replicated within 2%. The TX peer logs them delivered and never retries; with FEC off nothing else can repair them. So the drop is now counted (AsyncRxShared::pool_dropped, emitted in rx.ring) instead of silent, the ARQ caveat is documented at the mode comment and in CLAUDE.md, and examples/duplex grows the rx.seq/pctr ledger plus the consumer-cost knobs (DEVOURER_RX_SINK_SPIN_US / _STALL_MS / _STALL_EVERY, mirroring rxdemo) so the duplex ground-station topology stays benchable. Two traps the analyzer encodes: a killed event stream truncates its stdout tail and masquerades as end-of-run loss (TAIL_GUARD=512 excludes it; the harness runs per-line flush), and rx.ring telemetry is structurally blind at the loss instant because the frozen pump is what would have sampled it - the per-frame ledgers, not the ring counters, carry the verdict. Co-Authored-By: Claude Opus 4.8 --- CLAUDE.md | 10 +- examples/duplex/main.cpp | 73 ++++++++++ src/UsbTransport.cpp | 22 ++- tests/arq_e2e_analyze.py | 299 ++++++++++++++++++++++++++++++++++++++ tests/arq_e2e_delivery.sh | 197 +++++++++++++++++++++++++ 5 files changed, 597 insertions(+), 4 deletions(-) create mode 100644 tests/arq_e2e_analyze.py create mode 100644 tests/arq_e2e_delivery.sh diff --git a/CLAUDE.md b/CLAUDE.md index 51542b1..7575a2e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -421,7 +421,15 @@ guard time, dynamic beacon grants, ACK/TxReport, per-UE RX attribution) are the PHY ceiling by amortizing per-frame overhead — an occupancy metric can't show it, count delivered payload. `SetAckResponder(mac)` arms the hardware ACK/BlockAck responder; with a unicast TA on the soliciting frame this closes -a hardware-ARQ loop (autonomous MAC retransmission until ACK). Per-frame TX +a hardware-ARQ loop (autonomous MAC retransmission until ACK). The ACK's +horizon is chip-FIFO **admission** (bench: 8812EU responder, +`tests/arq_e2e_delivery.sh` per-frame ledgers): receiver-side congestion +upstream of admission declines the ACK, so the loop sees and retries it — but +a host stage that drops on a full queue while keeping URBs armed (the +spsc-fat ring at pool exhaustion — counted as `pool_dropped` in `rx.ring`) +turns the same loss into ACKed-but-undelivered that the TX peer logs as +delivered and never retries. Under ARQ or delivery accounting, prefer +backpressure to the chip over drop-on-full host buffering. Per-frame TX outcomes surface as `tx.report` events (CCX via C2H) — the TX-side link sensor; C2H rides the RX path, so J1/J2 TX-only sessions see none (run `DEVOURER_TX_WITH_RX=thread`; J3's coex thread drains C2H regardless). diff --git a/examples/duplex/main.cpp b/examples/duplex/main.cpp index 2b5da8b..2cef34e 100644 --- a/examples/duplex/main.cpp +++ b/examples/duplex/main.cpp @@ -123,6 +123,47 @@ static devourer::EventSink *g_ev = nullptr; static const bool g_tx_status_enabled = std::getenv("DEVOURER_TX_STATUS") != nullptr; +/* DEVOURER_RX_PCTR + DEVOURER_RX_AGG_SA: per-frame rx.seq delivery ledger, + * mirroring examples/rx/main.cpp — pctr is the u32 txdemo stamps at the + * QoS-Data body start (MPDU offset 26). In this demo the SA gate is required: + * the ledger's transmitter is a different station than the canonical-SA + * rx.frame stream above (the ARQ end-to-end bench keys on unicast frames + * whose TA can't be the group-address canonical SA). */ +static const bool g_rx_pctr = []() { + const char *e = std::getenv("DEVOURER_RX_PCTR"); + return e != nullptr && std::strcmp(e, "0") != 0; +}(); +static uint8_t g_seq_sa[6] = {}; +static const bool g_seq_sa_set = []() { + const char *e = std::getenv("DEVOURER_RX_AGG_SA"); + if (e == nullptr || *e == '\0') + return false; + const auto m = devourer::parse_mac(e); + if (!m) + return false; + std::memcpy(g_seq_sa, m->data(), 6); + return true; +}(); + +/* DEVOURER_RX_SINK_SPIN_US / DEVOURER_RX_SINK_STALL_MS+_EVERY: the same + * consumer-cost models as examples/rx/main.cpp — a per-frame busy-spin (the + * inline wfb-ng FEC+AES+UDP cost PixelPilot pays on this thread) and a + * periodic multi-ms stall (GC pause / consumer preemption). Both run on the + * libusb pump thread, which is exactly the point. */ +static const long g_rx_sink_spin_us = []() { + const char *e = std::getenv("DEVOURER_RX_SINK_SPIN_US"); + return e ? std::strtol(e, nullptr, 0) : 0L; +}(); +static const long g_rx_stall_ms = []() { + const char *e = std::getenv("DEVOURER_RX_SINK_STALL_MS"); + return e ? std::strtol(e, nullptr, 0) : 0L; +}(); +static const long g_rx_stall_every = []() { + const char *e = std::getenv("DEVOURER_RX_SINK_STALL_EVERY"); + return e ? std::strtol(e, nullptr, 0) : 100L; +}(); +static long g_rx_seen = 0; + static void packet_processor(const Packet &packet) { if (packet.RxAtrib.pkt_rpt_type == RX_PACKET_TYPE::C2H_PACKET) { if (!g_tx_status_enabled) return; @@ -148,6 +189,38 @@ static void packet_processor(const Packet &packet) { } return; } + ++g_rx_seen; + if (g_rx_sink_spin_us > 0) { + const auto deadline = std::chrono::steady_clock::now() + + std::chrono::microseconds(g_rx_sink_spin_us); + while (std::chrono::steady_clock::now() < deadline) { + /* busy-wait: a sleep would yield the pump thread and defeat the model */ + } + } + if (g_rx_stall_ms > 0 && (g_rx_seen % g_rx_stall_every) == 0) { + const auto deadline = std::chrono::steady_clock::now() + + std::chrono::milliseconds(g_rx_stall_ms); + while (std::chrono::steady_clock::now() < deadline) { + /* periodic consumer hiccup */ + } + } + + /* rx.seq — the ARQ bench's host-delivery ground truth: one lean event per + * SA-matched frame, same fields as rxdemo's so the analyzers are shared. */ + if (g_rx_pctr && g_seq_sa_set && packet.Data.size() >= 30 && + std::memcmp(packet.Data.data() + 10, g_seq_sa, 6) == 0) { + uint32_t pctr; + std::memcpy(&pctr, packet.Data.data() + 26, 4); + devourer::Ev(*g_ev, "rx.seq") + .t() /* host monotonic ms — correlates a pctr gap with an rx.ring dip */ + .f("pctr", (unsigned long long)pctr) + .f("tsfl", packet.RxAtrib.tsfl) + .f("seq", packet.RxAtrib.seq_num) + .f("crc", packet.RxAtrib.crc_err ? 1 : 0) + .f("paggr", packet.RxAtrib.paggr ? 1 : 0) + .f("ppdu", packet.RxAtrib.ppdu_cnt); + } + if (packet.Data.size() < 16) return; if (std::memcmp(packet.Data.data() + 10, kCanonicalSa, 6) != 0) return; long hits = ++g_rx_hits; diff --git a/src/UsbTransport.cpp b/src/UsbTransport.cpp index dfc1ae2..83db10f 100644 --- a/src/UsbTransport.cpp +++ b/src/UsbTransport.cpp @@ -85,8 +85,19 @@ struct AsyncRxShared { * the consumer stalls or is preempted — converting a chip-FIFO overflow * (dropped frames) into bounded host-queue backlog (delayed frames). The fix * for a stalling/preempted CONSUMER, where reorder-pool can't help because it - * still consumes on the pump thread. */ + * still consumes on the pump thread. + * + * ARQ caveat, bench-measured (tests/arq_e2e_delivery.sh): keeping the ring + * armed means the chip ADMITS AND ACKS every frame — so a frame dropped at + * pool exhaustion below is an ACKed-but-undelivered loss the hardware-ARQ + * peer will log as delivered and never retry. The default async ring loses + * the same frames chip-side instead, where the 8812EU declines the ACK and + * the ARQ loop recovers them (14,214/14,214 stall-window drops reported + * ok=0 there, vs 5,667 ok=1-but-lost here). Under an ARQ or + * delivery-accounting scheme, prefer backpressure over this mode, or watch + * pool_dropped. */ bool spsc = false; + std::atomic pool_dropped{0}; std::mutex queue_mu; std::condition_variable queue_cv; std::deque> queue; @@ -210,7 +221,10 @@ extern "C" void LIBUSB_CALL devourer_rx_cb(libusb_transfer *t) { /* Pool exhausted (consumer hopelessly behind under sustained overload) or * submit failed: preserve the pump's never-block invariant by re-arming * with the received buffer and DROPPING this frame — a bounded loss, vs the - * cascade an inline consume would trigger. */ + * cascade an inline consume would trigger. The chip already ACKed this + * frame (see the mode comment above): count it, never drop silently. */ + if (rlen > 0) + s->pool_dropped.fetch_add(1, std::memory_order_relaxed); if (resubmit && libusb_submit_transfer(t) == 0) { if (s->telemetry) s->armed.fetch_add(1, std::memory_order_relaxed); @@ -432,7 +446,9 @@ void UsbTransport::rx_loop( .f("empties", (unsigned long long)sh.empties.load( std::memory_order_relaxed)) .f("pool_free", pool_free) - .f("qdepth", qdepth); + .f("qdepth", qdepth) + .f("pool_dropped", (unsigned long long)sh.pool_dropped.load( + std::memory_order_relaxed)); } } } diff --git a/tests/arq_e2e_analyze.py b/tests/arq_e2e_analyze.py new file mode 100644 index 0000000..7852ff6 --- /dev/null +++ b/tests/arq_e2e_analyze.py @@ -0,0 +1,299 @@ +#!/usr/bin/env python3 +"""arq_e2e_analyze.py — join the three ARQ end-to-end ledgers. + +Inputs (from tests/arq_e2e_delivery.sh): + --dut duplex JSONL: rx.seq (host delivery ledger), stream.ctl (phase + markers in file order), rx.ring (ring telemetry) + --drone txdemo JSONL: tx.report (per-frame CCX: ok/retries/tag), tx.stats + --wit rxdemo JSONL: rx.seq (independent witness ledger) + +The drone's halmac tx.report echoes the descriptor SW_DEFINE low byte ("tag"), +stamped from the same per-frame counter txdemo writes into the payload +("pctr"), so tag == pctr (mod 256). Reports arrive in send order; unwrap the +tag into a relative index, then search the 256-aligned base offset that best +matches the observed pctr sets (the true base reconstructs the delivered set +almost perfectly; a wrong base matches ~1/256 at random). + +Verdict per phase: of the frames the drone believes DELIVERED (ok=1), how many +never reached the DUT host ledger — split by retries and by whether the +independent witness decoded them (witness-yes = loss inside the DUT chain). +""" +import argparse +import json +import sys +from collections import defaultdict + + +def iter_ev(path, names): + want = tuple('{"ev":"%s"' % n for n in names) + with open(path, errors="replace") as f: + for idx, line in enumerate(f): + if not line.startswith(want): + continue + try: + yield idx, json.loads(line) + except json.JSONDecodeError: + continue + + +def load_rx_ledger(path): + """pctr -> list of dicts (file order); plus event stream for phases.""" + ledger = defaultdict(list) + ctls = [] # file line idx of each stream.ctl (phase starts) + rings = [] # (line_idx, ev) for rx.ring + for idx, ev in iter_ev(path, ("rx.seq", "stream.ctl", "rx.ring")): + e = ev.get("ev") + if e == "rx.seq": + ledger[int(ev["pctr"])].append( + {"idx": idx, "t": ev.get("t"), "crc": ev.get("crc", 0)}) + elif e == "stream.ctl": + ctls.append(idx) + else: + rings.append((idx, ev)) + return ledger, ctls, rings + + +def load_reports(path): + reports = [] # file order: dict(tag, ok, retries, missed) + submitted = 0 + for _, ev in iter_ev(path, ("tx.report", "tx.stats")): + if ev.get("ev") == "tx.stats": + submitted = max(submitted, int(ev.get("submitted", 0))) + continue + if "tag" not in ev: # non-halmac format: tag join impossible + continue + reports.append({"tag": int(ev["tag"]), "ok": bool(ev.get("ok")), + "retries": int(ev.get("retries", 0)), + "missed": int(ev.get("missed", 0))}) + return reports, submitted + + +def unwrap_tags(reports): + """Relative frame index per report from mod-256 tag deltas.""" + rel, r = [], 0 + for j, rp in enumerate(reports): + if j: + r += (rp["tag"] - reports[j - 1]["tag"]) % 256 + rel.append(r) + return rel + + +def find_base(reports, rel, pctr_sets, max_pctr): + """base = tag0 + 256*m maximizing delivered-set overlap.""" + if not reports: + return None, 0, 0 + tag0 = reports[0]["tag"] + ok_rel = [rel[j] for j, rp in enumerate(reports) if rp["ok"]] + union = set() + for s in pctr_sets: + union |= set(s) + best, second, best_m = -1, -1, 0 + m_hi = (max_pctr + 4096) // 256 + 2 + # Reports are send-ordered and dense, so the first report's absolute index + # sits near the earliest delivered pctr — search a window around it first + # (the full scan is O(m_hi * n_ok) and unaffordable at high frame rates). + lo_pctr = min(union) if union else 0 + near = range(max(0, (lo_pctr - 2048)) // 256, + min(m_hi, (lo_pctr + 2048) // 256 + 1)) + for candidates in (near, range(m_hi)): + for m in candidates: + base = tag0 + 256 * m + score = sum(1 for r in ok_rel if (base + r - rel[0]) in union) + if score > best: + second, best, best_m = best, score, m + elif score > second: + second = score + if best >= max(10, len(ok_rel) // 2): + break # near-window hit — skip the exhaustive scan + return tag0 + 256 * best_m - rel[0], best, second + + +def phase_names(phases, cycles): + per = [p.strip() for p in phases.split(",") if p.strip()] + return [f"{p}" for _ in range(cycles) for p in per] + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--dut", required=True) + ap.add_argument("--drone", required=True) + ap.add_argument("--wit", required=True) + ap.add_argument("--phases", default="6M:10,idle,6M:30,idle") + ap.add_argument("--cycles", type=int, default=5) + ap.add_argument("--detail", type=int, default=30, + help="max acked-undelivered rows to print") + a = ap.parse_args() + + dut, ctls, rings = load_rx_ledger(a.dut) + wit, _, _ = load_rx_ledger(a.wit) + reports, submitted = load_reports(a.drone) + if not reports: + print("VERDICT: NO tx.report events with tags — nothing to join") + return 3 + + rel = unwrap_tags(reports) + max_pctr = max(list(dut.keys()) + list(wit.keys()) + [0]) + base, best, second = find_base(reports, rel, + (dut.keys(), wit.keys()), max_pctr) + n_ok = sum(1 for r in reports if r["ok"]) + print(f"# reports={len(reports)} ok={n_ok} submitted={submitted} " + f"dut_pctrs={len(dut)} wit_pctrs={len(wit)}") + # At ~100% delivery every 256-shift matches nearly everything, so the + # runner-up margin alone can't certify the base; the per-frame ledger + # identity printed after the join (dut_pctrs vs ok+drop_dlvd-au) is the + # real alignment proof. + margin_ok = second < 0 or best > 2 * max(second, 1) or \ + (best - second) > max(20, n_ok // 100) + print(f"# tag-align: base={base} matched={best}/{n_ok} " + f"(runner-up {second}) — " + f"{'OK' if margin_ok else 'WEAK-MARGIN (see ledger identity below)'}") + if best < max(10, n_ok // 4): + print("VERDICT: tag alignment failed — ledgers don't join; " + "check SA gates / report coverage") + return 3 + + # Phase attribution: each DUT rx.seq line idx -> phase via ctl markers. + names = phase_names(a.phases, a.cycles) + if len(ctls) != len(names): + print(f"# WARNING: {len(ctls)} stream.ctl markers vs " + f"{len(names)} expected phases — trailing phases truncated") + names = names[:len(ctls)] + + def phase_of_idx(idx): + p = -1 + for i, c in enumerate(ctls): + if idx >= c: + p = i + else: + break + return p # -1 = warmup + + # Delivered frame -> phase (first delivery's line idx). + k_phase = {} + for k, hits in dut.items(): + k_phase[k] = phase_of_idx(hits[0]["idx"]) + delivered_sorted = sorted(k_phase) + + def attribute(k): + """Phase of an UNdelivered frame k: nearest delivered neighbours.""" + import bisect + i = bisect.bisect_left(delivered_sorted, k) + lo = delivered_sorted[i - 1] if i else None + hi = delivered_sorted[i] if i < len(delivered_sorted) else None + pl = k_phase.get(lo, -2) + ph = k_phase.get(hi, -2) + if pl == ph: + return pl + return ph if hi is not None and lo is None else pl + + # Ring telemetry per phase: min armed depth + empties/completions deltas. + ring_by_phase = defaultdict(lambda: {"min_armed": None, "empties": 0, + "completions": 0}) + prev = {} + for idx, ev in rings: + p = phase_of_idx(idx) + st = ring_by_phase[p] + ma = ev.get("min_armed") + if ma is not None: + st["min_armed"] = ma if st["min_armed"] is None else min( + st["min_armed"], ma) + for f in ("empties", "completions"): + v = int(ev.get(f, 0)) + st[f] += max(0, v - prev.get(f, v)) + prev[f] = v + + # The join. Frames in the last TAIL_GUARD indices of the run are excluded + # from the headline: the receivers' event streams end by process kill, so + # a missing pctr there is indistinguishable from stream-tail truncation + # (measured: 19 consecutive "losses" at the exact end of a run that were + # an unflushed-stdout artifact, not RF or USB). + TAIL_GUARD = 512 + max_index = max([base + rel[-1]] + list(dut.keys()) + list(wit.keys())) + tail_cutoff = max_index - TAIL_GUARD + per = defaultdict(lambda: defaultdict(int)) + detail = [] + for j, rp in enumerate(reports): + k = base + rel[j] + in_dut = k in dut + in_wit = k in wit + p = k_phase.get(k) if in_dut else attribute(k) + st = per[p] + st["reports"] += 1 + st["retries_sum"] += rp["retries"] + if rp["ok"]: + st["ok"] += 1 + if rp["retries"] > 0: + st["ok_retried"] += 1 + if not in_dut: + if k >= tail_cutoff: + st["tail_suspect"] += 1 + continue + st["acked_undelivered"] += 1 + if rp["retries"] > 0: + st["au_retried"] += 1 + if in_wit: + st["au_witnessed"] += 1 + if len(detail) < a.detail: + detail.append((k, rp["retries"], p, in_wit)) + else: + st["dropped"] += 1 + if in_dut: + st["dropped_but_delivered"] += 1 + for k, hits in dut.items(): + if len(hits) > 1: + per[k_phase[k]]["dup_delivered"] += 1 + + print() + hdr = (f"{'phase':>10} {'reports':>8} {'ok%':>6} {'ok_rtry':>8} " + f"{'ACKED_UNDELIV':>14} {'au_rtry':>8} {'au_wit':>7} " + f"{'drop':>6} {'drop_dlvd':>9} {'dupD':>5} " + f"{'min_armed':>9} {'empties':>8}") + print(hdr) + total_au = 0 + for p in sorted(per): + st = per[p] + name = "warmup" if p == -1 else ( + names[p] if 0 <= p < len(names) else f"?{p}") + okp = 100.0 * st["ok"] / st["reports"] if st["reports"] else 0.0 + rg = ring_by_phase.get(p, {}) + total_au += st["acked_undelivered"] + print(f"{name:>10} {st['reports']:>8} {okp:>6.1f} " + f"{st['ok_retried']:>8} {st['acked_undelivered']:>14} " + f"{st['au_retried']:>8} {st['au_witnessed']:>7} " + f"{st['dropped']:>6} {st['dropped_but_delivered']:>9} " + f"{st['dup_delivered']:>5} " + f"{str(rg.get('min_armed', '-')):>9} " + f"{rg.get('empties', 0):>8}") + + print() + if detail: + print("first acked-undelivered frames (k, retries, phase, witnessed):") + for k, r, p, w in detail: + name = "warmup" if p == -1 else ( + names[p] if p is not None and 0 <= p < len(names) else f"?{p}") + print(f" k={k} retries={r} phase={name} wit={'Y' if w else 'n'}") + # Ledger identity: every ok'd frame must be delivered (minus the au set), + # plus the ok=0-but-delivered strays. Holding to a few frames certifies + # both the base alignment and the ledgers themselves. + tot_dd = sum(st["dropped_but_delivered"] for st in per.values()) + tot_tail = sum(st["tail_suspect"] for st in per.values()) + ident = n_ok - total_au - tot_tail + tot_dd + print(f"ledger identity: ok - acked_undelivered - tail_suspect " + f"+ dropped_but_delivered = {ident} vs dut_pctrs = {len(dut)} " + f"(delta {len(dut) - ident})") + if tot_tail: + print(f"tail_suspect: {tot_tail} ok'd frames missing within the last " + f"{TAIL_GUARD} indices — excluded (stream-end truncation)") + verdict = "REPRODUCED" if total_au > 0 else "NOT-REPRODUCED" + print(f"\nVERDICT: {verdict} — acked_undelivered={total_au} across all " + f"phases (drone said ok, DUT host never delivered)") + print(json.dumps({"ev": "arqe2e.verdict", "acked_undelivered": total_au, + "tail_suspect": tot_tail, + "reports": len(reports), "ok": n_ok, + "dut_pctrs": len(dut), "wit_pctrs": len(wit), + "base": base})) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/arq_e2e_delivery.sh b/tests/arq_e2e_delivery.sh new file mode 100644 index 0000000..53736b7 --- /dev/null +++ b/tests/arq_e2e_delivery.sh @@ -0,0 +1,197 @@ +#!/usr/bin/env bash +# +# arq_e2e_delivery.sh — does a hardware-ACKed frame actually reach the RX app? +# +# Field report (PixelPilot, 8812AU drone -> 8812EU ground): with hardware ARQ +# (DEVOURER_TX_RETRY_LIMIT + DEVOURER_ACK_RESPONDER) the drone's tx.report says +# a packet was delivered after a retry, yet the ground app never receives it — +# "ACKed-but-undelivered". The existing ACK harnesses judge the loop from the +# TX side's CCX reports only, so this failure was structurally invisible. +# +# This bench closes the loop with three per-frame ledgers on one channel: +# DUT 8812EU (J3), examples/duplex — the PixelPilot one-handle topology: +# RX loop + stdin-fed uplink bursts, ACK responder armed, host +# delivery ledger = rx.seq (pctr), ring telemetry = rx.ring. +# DRONE 8812CU (J3), txdemo — unicast QoS-Data at the responder MAC, +# normal ack-policy, retrying descriptor, per-frame tx.report. +# The halmac report tag echoes the frame counter (mod 256), which +# IS the payload pctr, so reports join to payloads exactly. +# WITNESS 8814AU (J1), rxdemo — passive monitor with the same rx.seq ledger: +# what a quiet, healthy receiver decodes on this channel. A pctr the +# witness has and the DUT lacks is a loss INSIDE the DUT chain. +# +# The DUT's uplink alternates phases inside ONE session (idle control / +# PixelPilot burst shape / 3x burst), marked by stream.ctl events in file +# order. The drone-sim defaults to DEVOURER_DIS_CCA=1: in the field the drone +# is effectively a hidden node to the ground's low-power feedback, so it +# transmits INTO the bursts; near-field CSMA would defer instead and suppress +# the collision population under test. +# +# sudo bash tests/arq_e2e_delivery.sh +# RETRY_LIMIT=8 CYCLES=3 CH=36 sudo bash tests/arq_e2e_delivery.sh +# DUT_RX_MODE=spsc-fat sudo bash tests/arq_e2e_delivery.sh # mitigation arm +set -u + +ROOT="$(cd "$(dirname "$0")/.." && pwd)" +BUILD=${BUILD:-$ROOT/build} + +DUT_VID=${DUT_VID:-0x0bda}; DUT_PID=${DUT_PID:-0xa81a} # RTL8812EU (J3) +DRONE_VID=${DRONE_VID:-0x0bda}; DRONE_PID=${DRONE_PID:-0xc812} # RTL8812CU (J3) +WIT_VID=${WIT_VID:-0x0bda}; WIT_PID=${WIT_PID:-0x8813} # RTL8814AU (J1) + +CH=${CH:-36} +MAC1=${MAC1:-02:12:34:56:78:9a} # DUT responder identity = drone RA +TX_SA=${TX_SA:-02:aa:bb:cc:dd:01} # drone TA (unicast — the I/G footgun) +RETRY_LIMIT=${RETRY_LIMIT:-3} # field report used 3 +DRONE_RATE=${DRONE_RATE:-MCS3} +DRONE_PAYLOAD=${DRONE_PAYLOAD:-512} # >= 30 so the pctr stamp fits +DRONE_GAP_US=${DRONE_GAP_US:-1000} # ~1k fps video-sim +DRONE_DIS_CCA=${DRONE_DIS_CCA:-1} +PWR_QDB=${PWR_QDB:--40} # de-saturate the near field (-10 dB) + +CYCLES=${CYCLES:-5} +PHASE_S=${PHASE_S:-8} +PHASES=${PHASES:-"6M:10,idle,6M:30,idle"} # burst first: preflight sees TX +PERIOD_MS=${PERIOD_MS:-100} # PixelPilot adaptive-link period +PSDU=${PSDU:-100} # PixelPilot uplink PSDU +WARMUP_S=${WARMUP_S:-12} # feeder idle lead-in (duplex bring-up) +PREFLIGHT_S=${PREFLIGHT_S:-14} # liveness check delay after drone start +DUT_RX_MODE=${DUT_RX_MODE:-} # empty = default async ring +DUT_POOL_SPARE=${DUT_POOL_SPARE:-16} # only read by the pool modes +DUT_SPIN_US=${DUT_SPIN_US:-0} # per-frame inline consumer cost model +DUT_STALL_MS=${DUT_STALL_MS:-0} # periodic consumer hiccup (GC pause) +DUT_STALL_EVERY=${DUT_STALL_EVERY:-1500} + +MODS=${MODS:-"rtw88_8812au rtw88_8821au rtw88_8822bu rtw88_8814au rtw88_8822cu rtw88_8822eu"} +BLACKLIST=/etc/modprobe.d/zz-temp-blacklist-arqe2e.conf +OUT=${OUT:-/tmp/arq-e2e/$(date +%Y%m%d-%H%M%S)} +mkdir -p "$OUT" + +[ "$(id -u)" = 0 ] || { echo "must run as root"; exit 3; } +for b in duplex txdemo rxdemo; do + [ -x "$BUILD/$b" ] || { echo "build $b first"; exit 3; } +done + +# Total span the drone must cover: feeder warmup + all phases. +NPHASES=$(awk -F, '{print NF}' <<<"$PHASES") +SPAN=$(( ${WARMUP_S%.*} + CYCLES * NPHASES * ${PHASE_S%.*} + 10 )) + +WIT_PIDF=""; DUT_PIDF=""; FEED_PIDF=""; DRONE_PIDF="" +cleanup() { + trap - EXIT INT TERM + for p in "$FEED_PIDF" "$DRONE_PIDF" "$DUT_PIDF" "$WIT_PIDF"; do + [ -n "$p" ] && kill "$p" 2>/dev/null + done + pkill -x duplex 2>/dev/null; pkill -x txdemo 2>/dev/null + pkill -x rxdemo 2>/dev/null + rm -f "$BLACKLIST" "$OUT/fifo" + wait 2>/dev/null +} +trap cleanup EXIT INT TERM + +# Temp-blacklist (modprobe -r does NOT survive the re-enumeration that +# claim_interface_then_reset triggers) + unbind by exact VID:PID. Never touch +# device class 09: the dongles sit behind Realtek-branded hubs and a +# vendor-only match would unbind the hub driver and drop the whole tree. +: > "$BLACKLIST" +for m in $MODS; do echo "blacklist $m" >> "$BLACKLIST"; modprobe -r "$m" 2>/dev/null; done +ADAPTERS="${DUT_VID#0x}${DUT_PID#0x} ${DRONE_VID#0x}${DRONE_PID#0x} ${WIT_VID#0x}${WIT_PID#0x}" +for d in /sys/bus/usb/devices/*/; do + [ -f "$d/idVendor" ] || continue + [ "$(cat "$d/bDeviceClass" 2>/dev/null)" = "09" ] && continue + id="$(cat "$d/idVendor")$(cat "$d/idProduct")" + case " $ADAPTERS " in *" $id "*) ;; *) continue ;; esac + for i in "$d"*:*; do + [ -d "$i" ] && [ -e "$i/driver" ] || continue + basename "$i" > "$(readlink -f "$i/driver")/unbind" 2>/dev/null && + echo "[arq-e2e] unbound $(basename "$i")" + done +done +sleep 1 + +echo "[arq-e2e] ch=$CH retry_limit=$RETRY_LIMIT drone=$DRONE_RATE/${DRONE_GAP_US}us/${DRONE_PAYLOAD}B" \ + "dis_cca=$DRONE_DIS_CCA phases=$PHASES cycles=$CYCLES span=${SPAN}s" \ + "dut_rx_mode=${DUT_RX_MODE:-async} out=$OUT" + +# --- witness: quiet monitor, third ledger -------------------------------- +env DEVOURER_VID="$WIT_VID" DEVOURER_PID="$WIT_PID" DEVOURER_CHANNEL="$CH" \ + DEVOURER_RX_PCTR=1 DEVOURER_RX_AGG_SA="$TX_SA" \ + DEVOURER_LOG_LEVEL=warn DEVOURER_EVENTS=stdout \ + "$BUILD/rxdemo" >"$OUT/wit.jsonl" 2>"$OUT/wit.err" & +WIT_PIDF=$! +sleep 8 + +# --- DUT: PixelPilot topology — duplex RX + burst uplink, responder armed --- +mkfifo "$OUT/fifo" +env DEVOURER_VID="$DUT_VID" DEVOURER_PID="$DUT_PID" DEVOURER_CHANNEL="$CH" \ + DEVOURER_ACK_RESPONDER="$MAC1" \ + DEVOURER_RX_PCTR=1 DEVOURER_RX_AGG_SA="$TX_SA" DEVOURER_RX_RING_MS=200 \ + ${DUT_RX_MODE:+DEVOURER_RX_MODE="$DUT_RX_MODE"} \ + ${DUT_RX_MODE:+DEVOURER_RX_POOL_SPARE="$DUT_POOL_SPARE"} \ + DEVOURER_RX_SINK_SPIN_US="$DUT_SPIN_US" \ + DEVOURER_RX_SINK_STALL_MS="$DUT_STALL_MS" \ + DEVOURER_RX_SINK_STALL_EVERY="$DUT_STALL_EVERY" \ + DEVOURER_TX_PWR_OFFSET_QDB="$PWR_QDB" \ + DEVOURER_LOG_LEVEL=info DEVOURER_EVENTS=stdout \ + "$BUILD/duplex" --interval-ms 0 --max-psdu 4096 \ + <"$OUT/fifo" >"$OUT/dut.jsonl" 2>"$OUT/dut.err" & +DUT_PIDF=$! + +python3 "$ROOT/tests/pp109_uplink_feeder.py" \ + --phases "$PHASES" --cycles "$CYCLES" --phase-s "$PHASE_S" \ + --period-ms "$PERIOD_MS" --psdu "$PSDU" --warmup-s "$WARMUP_S" \ + >"$OUT/fifo" 2>"$OUT/feeder.log" & +FEED_PIDF=$! + +# --- drone-sim: retrying unicast flood, per-frame reports ------------------- +# Started immediately: its ~6-8 s bring-up overlaps the feeder's idle warmup, +# so the flood is airing before phase 1's first burst. +env DEVOURER_VID="$DRONE_VID" DEVOURER_PID="$DRONE_PID" DEVOURER_CHANNEL="$CH" \ + DEVOURER_TX_QOS_DATA=1 DEVOURER_TX_RA="$MAC1" DEVOURER_TX_SA="$TX_SA" \ + DEVOURER_TX_RATE="$DRONE_RATE" DEVOURER_TX_PAYLOAD_BYTES="$DRONE_PAYLOAD" \ + DEVOURER_TX_GAP_US="$DRONE_GAP_US" \ + DEVOURER_TX_REPORT=1 DEVOURER_TX_RETRY_LIMIT="$RETRY_LIMIT" \ + DEVOURER_DIS_CCA="$DRONE_DIS_CCA" \ + DEVOURER_TX_PWR_OFFSET_QDB="$PWR_QDB" \ + DEVOURER_LOG_LEVEL=warn DEVOURER_EVENTS=stdout \ + timeout -s INT "$SPAN" "$BUILD/txdemo" \ + >"$OUT/drone.jsonl" 2>"$OUT/drone.err" & +DRONE_PIDF=$! + +# --- preflight: fail fast with a diagnosis, not a zero-filled matrix -------- +# Judged after warmup + one burst phase: every ledger must be alive by then. +sleep "$(( ${WARMUP_S%.*} + PREFLIGHT_S ))" +fail="" +n_dut=$(grep -c '"ev":"rx.seq"' "$OUT/dut.jsonl" 2>/dev/null || true) +n_wit=$(grep -c '"ev":"rx.seq"' "$OUT/wit.jsonl" 2>/dev/null || true) +n_rpt=$(grep -c '"ev":"tx.report"' "$OUT/drone.jsonl" 2>/dev/null || true) +n_ok=$(grep '"ev":"tx.report"' "$OUT/drone.jsonl" 2>/dev/null | grep -c '"ok":true' || true) +hits=$(grep '"ev":"rx.txhit"' "$OUT/wit.jsonl" 2>/dev/null | tail -1 | + sed -n 's/.*"hits":\([0-9]*\).*/\1/p'); hits=${hits:-0} +[ "${n_dut:-0}" -ge 50 ] || fail="$fail DUT-rx.seq=$n_dut(<50:not-receiving-flood)" +[ "${n_wit:-0}" -ge 50 ] || fail="$fail WIT-rx.seq=$n_wit(<50:witness-deaf)" +[ "${n_rpt:-0}" -ge 20 ] || fail="$fail tx.report=$n_rpt(<20:reports-missing)" +if [ "${n_rpt:-0}" -ge 20 ] && [ $((n_ok * 2)) -lt "$n_rpt" ]; then + fail="$fail ok=$n_ok/$n_rpt(responder-not-ACKing)" +fi +[ "$hits" -ge 5 ] || fail="$fail wit-txhit=$hits(<5:DUT-uplink-not-airing)" +if [ -n "$fail" ]; then + echo "[arq-e2e] PREFLIGHT FAILED:$fail" + echo "[arq-e2e] logs: $OUT (dut.err / drone.err / wit.err)" + exit 3 +fi +echo "[arq-e2e] preflight OK: dut_seq=$n_dut wit_seq=$n_wit reports=$n_rpt" \ + "ok=$n_ok wit_txhit=$hits — running matrix (${SPAN}s total)" + +wait "$FEED_PIDF"; FEED_PIDF="" +sleep 2 +kill "$DRONE_PIDF" 2>/dev/null; wait "$DRONE_PIDF" 2>/dev/null; DRONE_PIDF="" +sleep 1 +kill "$DUT_PIDF" 2>/dev/null; DUT_PIDF="" +kill "$WIT_PIDF" 2>/dev/null; WIT_PIDF="" +sleep 1 + +python3 "$ROOT/tests/arq_e2e_analyze.py" \ + --dut "$OUT/dut.jsonl" --drone "$OUT/drone.jsonl" --wit "$OUT/wit.jsonl" \ + --phases "$PHASES" --cycles "$CYCLES" | tee "$OUT/report.txt" +echo "[arq-e2e] logs: $OUT" From 111a0675335efc04e351b5d3f06a7ce1cf25c199 Mon Sep 17 00:00:00 2001 From: Joseph <162703152+josephnef@users.noreply.github.com> Date: Mon, 3 Aug 2026 16:56:19 +0300 Subject: [PATCH 02/12] arq-e2e review fixes: compact verdict JSONL, stall-knob clamp, atomic rx counter, teardown-gated pool_dropped, /run blacklist MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Qodo review round on #359, all five findings addressed and re-validated on air (third spsc-fat replication: acked_undelivered 5,468 vs 5,667/5,570 on the prior runs, pool_dropped 3,477 — the counter still tracks exhaustion drops with the teardown gate in place): - arqe2e.verdict now emits compact separators — machine events must stay in the grep-able {"ev":"name",...} form (docs/logging.md). - DEVOURER_RX_SINK_STALL_EVERY=0/garbage no longer divides by zero: parse clamps to the 100 default, in duplex and in rxdemo's identical knob. - duplex g_rx_seen is atomic (the RX callback can run on the TX thread's event pump too) and the stall decision uses the increment's return value. - pool_dropped counts only resubmit-eligible drops: a teardown-window frame is intentional loss, not an overload signal. - The temp module blacklist moved to /run/modprobe.d (read like /etc/modprobe.d, tmpfs) so it self-cleans on reboot even after SIGKILL — it still must exist during the run because modprobe -r does not survive the re-enumeration claim_interface_then_reset triggers. Co-Authored-By: Claude Opus 4.8 --- examples/duplex/main.cpp | 11 +++++++---- examples/rx/main.cpp | 3 ++- src/UsbTransport.cpp | 6 ++++-- tests/arq_e2e_analyze.py | 4 +++- tests/arq_e2e_delivery.sh | 6 +++++- 5 files changed, 21 insertions(+), 9 deletions(-) diff --git a/examples/duplex/main.cpp b/examples/duplex/main.cpp index 2cef34e..6f056cc 100644 --- a/examples/duplex/main.cpp +++ b/examples/duplex/main.cpp @@ -160,9 +160,12 @@ static const long g_rx_stall_ms = []() { }(); static const long g_rx_stall_every = []() { const char *e = std::getenv("DEVOURER_RX_SINK_STALL_EVERY"); - return e ? std::strtol(e, nullptr, 0) : 100L; + const long v = e ? std::strtol(e, nullptr, 0) : 100L; + return v > 0 ? v : 100L; /* 0/garbage would divide-by-zero the modulo */ }(); -static long g_rx_seen = 0; +/* Atomic: the RX callback can run on the TX thread's event pump too (libusb's + * sync API pumps events; see AsyncRxShared in src/UsbTransport.cpp). */ +static std::atomic g_rx_seen{0}; static void packet_processor(const Packet &packet) { if (packet.RxAtrib.pkt_rpt_type == RX_PACKET_TYPE::C2H_PACKET) { @@ -189,7 +192,7 @@ static void packet_processor(const Packet &packet) { } return; } - ++g_rx_seen; + const long rx_seen = ++g_rx_seen; if (g_rx_sink_spin_us > 0) { const auto deadline = std::chrono::steady_clock::now() + std::chrono::microseconds(g_rx_sink_spin_us); @@ -197,7 +200,7 @@ static void packet_processor(const Packet &packet) { /* busy-wait: a sleep would yield the pump thread and defeat the model */ } } - if (g_rx_stall_ms > 0 && (g_rx_seen % g_rx_stall_every) == 0) { + if (g_rx_stall_ms > 0 && (rx_seen % g_rx_stall_every) == 0) { const auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(g_rx_stall_ms); while (std::chrono::steady_clock::now() < deadline) { diff --git a/examples/rx/main.cpp b/examples/rx/main.cpp index a57d51d..baf5827 100644 --- a/examples/rx/main.cpp +++ b/examples/rx/main.cpp @@ -644,7 +644,8 @@ static const long g_rx_stall_ms = []() { }(); static const long g_rx_stall_every = []() { const char *e = std::getenv("DEVOURER_RX_SINK_STALL_EVERY"); - return e ? std::strtol(e, nullptr, 0) : 100L; + const long v = e ? std::strtol(e, nullptr, 0) : 100L; + return v > 0 ? v : 100L; /* 0/garbage would divide-by-zero the modulo */ }(); /* DEVOURER_RX_PCTR: emit a lean rx.seq event (payload counter + tsfl + crc + diff --git a/src/UsbTransport.cpp b/src/UsbTransport.cpp index 83db10f..7580c3a 100644 --- a/src/UsbTransport.cpp +++ b/src/UsbTransport.cpp @@ -222,8 +222,10 @@ extern "C" void LIBUSB_CALL devourer_rx_cb(libusb_transfer *t) { * submit failed: preserve the pump's never-block invariant by re-arming * with the received buffer and DROPPING this frame — a bounded loss, vs the * cascade an inline consume would trigger. The chip already ACKed this - * frame (see the mode comment above): count it, never drop silently. */ - if (rlen > 0) + * frame (see the mode comment above): count it, never drop silently. + * `resubmit` gates the count: a teardown-window frame (stop requested) is + * intentional loss, not an overload signal. */ + if (resubmit && rlen > 0) s->pool_dropped.fetch_add(1, std::memory_order_relaxed); if (resubmit && libusb_submit_transfer(t) == 0) { if (s->telemetry) diff --git a/tests/arq_e2e_analyze.py b/tests/arq_e2e_analyze.py index 7852ff6..16e6e54 100644 --- a/tests/arq_e2e_analyze.py +++ b/tests/arq_e2e_analyze.py @@ -287,11 +287,13 @@ def attribute(k): verdict = "REPRODUCED" if total_au > 0 else "NOT-REPRODUCED" print(f"\nVERDICT: {verdict} — acked_undelivered={total_au} across all " f"phases (drone said ok, DUT host never delivered)") + # separators: machine events must be the grep-able {"ev":"name",...} form + # (docs/logging.md) — default json.dumps inserts spaces. print(json.dumps({"ev": "arqe2e.verdict", "acked_undelivered": total_au, "tail_suspect": tot_tail, "reports": len(reports), "ok": n_ok, "dut_pctrs": len(dut), "wit_pctrs": len(wit), - "base": base})) + "base": base}, separators=(",", ":"))) return 0 diff --git a/tests/arq_e2e_delivery.sh b/tests/arq_e2e_delivery.sh index 53736b7..77aa4f4 100644 --- a/tests/arq_e2e_delivery.sh +++ b/tests/arq_e2e_delivery.sh @@ -63,7 +63,10 @@ DUT_STALL_MS=${DUT_STALL_MS:-0} # periodic consumer hiccup (GC pause) DUT_STALL_EVERY=${DUT_STALL_EVERY:-1500} MODS=${MODS:-"rtw88_8812au rtw88_8821au rtw88_8822bu rtw88_8814au rtw88_8822cu rtw88_8822eu"} -BLACKLIST=/etc/modprobe.d/zz-temp-blacklist-arqe2e.conf +# /run/modprobe.d (tmpfs): modprobe reads it like /etc/modprobe.d, but the +# blacklist self-cleans on reboot even if the trap never runs (SIGKILL, power +# loss) — a temp blacklist must never outlive the bench across boots. +BLACKLIST=/run/modprobe.d/zz-temp-blacklist-arqe2e.conf OUT=${OUT:-/tmp/arq-e2e/$(date +%Y%m%d-%H%M%S)} mkdir -p "$OUT" @@ -93,6 +96,7 @@ trap cleanup EXIT INT TERM # claim_interface_then_reset triggers) + unbind by exact VID:PID. Never touch # device class 09: the dongles sit behind Realtek-branded hubs and a # vendor-only match would unbind the hub driver and drop the whole tree. +mkdir -p "$(dirname "$BLACKLIST")" # /run/modprobe.d may not exist yet : > "$BLACKLIST" for m in $MODS; do echo "blacklist $m" >> "$BLACKLIST"; modprobe -r "$m" 2>/dev/null; done ADAPTERS="${DUT_VID#0x}${DUT_PID#0x} ${DRONE_VID#0x}${DRONE_PID#0x} ${WIT_VID#0x}${WIT_PID#0x}" From 8aa60315c8f381e3cb026be271c36b968d4c5418 Mon Sep 17 00:00:00 2001 From: Joseph <162703152+josephnef@users.noreply.github.com> Date: Mon, 3 Aug 2026 17:11:40 +0300 Subject: [PATCH 03/12] arq-e2e review round 2: short-run guard fails loud, sampled base-scan fallback, checkout-scoped cleanup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - A run spanning fewer than 2x TAIL_GUARD indices now refuses to conclude (INCONCLUSIVE-SHORT-RUN, exit 3, verdict field in the machine event) instead of letting the tail window silently swallow every missing frame into tail_suspect and print a vacuous NOT-REPRODUCED. - The exhaustive base-scan fallback scores candidates on a 2k-report sample (a wrong base matches ~1/256 at random, so the sample separates decisively) and re-scores only the winner in full — non-joining logs now fail fast instead of costing O(m_hi * n_ok). - The cleanup backstop pkills full paths under this checkout's build dir rather than bare comm names, so a root-run bench can no longer kill unrelated duplex/txdemo/rxdemo instances elsewhere on the host. Validated: full-run parity on the round-1 logs (same au, ~5 s), a 341-index truncation triggers the INCONCLUSIVE path, and a fourth on-air spsc-fat replication (acked_undelivered 5,739; series 5,667/5,570/5,468/5,739) exits with zero leftover processes. Co-Authored-By: Claude Opus 4.8 --- tests/arq_e2e_analyze.py | 40 ++++++++++++++++++++++++++++++--------- tests/arq_e2e_delivery.sh | 5 +++-- 2 files changed, 34 insertions(+), 11 deletions(-) diff --git a/tests/arq_e2e_analyze.py b/tests/arq_e2e_analyze.py index 16e6e54..166c211 100644 --- a/tests/arq_e2e_analyze.py +++ b/tests/arq_e2e_analyze.py @@ -91,21 +91,29 @@ def find_base(reports, rel, pctr_sets, max_pctr): m_hi = (max_pctr + 4096) // 256 + 2 # Reports are send-ordered and dense, so the first report's absolute index # sits near the earliest delivered pctr — search a window around it first - # (the full scan is O(m_hi * n_ok) and unaffordable at high frame rates). + # with full precision. The exhaustive fallback scores on a SAMPLE of the + # reports (a wrong base matches ~1/256 at random, so 2k samples separate + # right from wrong decisively) and only the winner is re-scored in full — + # otherwise non-joining logs cost O(m_hi * n_ok) and the analyzer hangs + # instead of failing fast. lo_pctr = min(union) if union else 0 near = range(max(0, (lo_pctr - 2048)) // 256, min(m_hi, (lo_pctr + 2048) // 256 + 1)) - for candidates in (near, range(m_hi)): + sample = ok_rel[:2000] + for candidates, pool in ((near, ok_rel), (range(m_hi), sample)): for m in candidates: base = tag0 + 256 * m - score = sum(1 for r in ok_rel if (base + r - rel[0]) in union) + score = sum(1 for r in pool if (base + r) in union) if score > best: second, best, best_m = best, score, m elif score > second: second = score - if best >= max(10, len(ok_rel) // 2): - break # near-window hit — skip the exhaustive scan - return tag0 + 256 * best_m - rel[0], best, second + if best >= max(10, len(pool) // 2): + break # strong hit — skip / end the wider scan + best, second = -1, -1 # sampled scores are not comparable to full ones + full = sum(1 for r in ok_rel + if (tag0 + 256 * best_m + r) in union) + return tag0 + 256 * best_m - rel[0], full, second def phase_names(phases, cycles): @@ -210,6 +218,14 @@ def attribute(k): TAIL_GUARD = 512 max_index = max([base + rel[-1]] + list(dut.keys()) + list(wit.keys())) tail_cutoff = max_index - TAIL_GUARD + # On a run shorter than 2x the guard the tail window swallows most of the + # frames and a NOT-REPRODUCED verdict would be vacuous — refuse to conclude + # rather than silently reclassify real losses as truncation. + short_run = max_index < 2 * TAIL_GUARD + if short_run: + print(f"# WARNING: run spans only {max_index + 1} frame indices " + f"(< 2x TAIL_GUARD={TAIL_GUARD}) — verdict will be " + f"INCONCLUSIVE-SHORT-RUN") per = defaultdict(lambda: defaultdict(int)) detail = [] for j, rp in enumerate(reports): @@ -284,17 +300,23 @@ def attribute(k): if tot_tail: print(f"tail_suspect: {tot_tail} ok'd frames missing within the last " f"{TAIL_GUARD} indices — excluded (stream-end truncation)") - verdict = "REPRODUCED" if total_au > 0 else "NOT-REPRODUCED" + if total_au > 0: + verdict = "REPRODUCED" + elif short_run: + verdict = "INCONCLUSIVE-SHORT-RUN" + else: + verdict = "NOT-REPRODUCED" print(f"\nVERDICT: {verdict} — acked_undelivered={total_au} across all " f"phases (drone said ok, DUT host never delivered)") # separators: machine events must be the grep-able {"ev":"name",...} form # (docs/logging.md) — default json.dumps inserts spaces. - print(json.dumps({"ev": "arqe2e.verdict", "acked_undelivered": total_au, + print(json.dumps({"ev": "arqe2e.verdict", "verdict": verdict, + "acked_undelivered": total_au, "tail_suspect": tot_tail, "reports": len(reports), "ok": n_ok, "dut_pctrs": len(dut), "wit_pctrs": len(wit), "base": base}, separators=(",", ":"))) - return 0 + return 3 if verdict == "INCONCLUSIVE-SHORT-RUN" else 0 if __name__ == "__main__": diff --git a/tests/arq_e2e_delivery.sh b/tests/arq_e2e_delivery.sh index 77aa4f4..991285f 100644 --- a/tests/arq_e2e_delivery.sh +++ b/tests/arq_e2e_delivery.sh @@ -85,8 +85,9 @@ cleanup() { for p in "$FEED_PIDF" "$DRONE_PIDF" "$DUT_PIDF" "$WIT_PIDF"; do [ -n "$p" ] && kill "$p" 2>/dev/null done - pkill -x duplex 2>/dev/null; pkill -x txdemo 2>/dev/null - pkill -x rxdemo 2>/dev/null + # Backstop scoped to THIS checkout's binaries (full-path match) — a bare + # `pkill -x duplex` from root would kill unrelated instances host-wide. + for b in duplex txdemo rxdemo; do pkill -f "^$BUILD/$b" 2>/dev/null; done rm -f "$BLACKLIST" "$OUT/fifo" wait 2>/dev/null } From 95ab617f8f555ba4a118a7b8ef8bc3223c140d0a Mon Sep 17 00:00:00 2001 From: Joseph <162703152+josephnef@users.noreply.github.com> Date: Mon, 3 Aug 2026 17:21:00 +0300 Subject: [PATCH 04/12] arq-e2e analyzer: flush the verdict machine-event line MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The one JSONL machine event the analyzer emits now flushes explicitly, so it lands atomically even under a live consumer. The human report stays on stdout by design: this is an offline report generator the harness tees into report.txt — the same contract as rxq_analyze.py / pp109_starve_analyze.py / ack_txreport_analyze.py — not a demo's runtime event plane, where the two-plane stdout/stderr split applies. Co-Authored-By: Claude Opus 4.8 --- tests/arq_e2e_analyze.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/tests/arq_e2e_analyze.py b/tests/arq_e2e_analyze.py index 166c211..95bbfa4 100644 --- a/tests/arq_e2e_analyze.py +++ b/tests/arq_e2e_analyze.py @@ -310,12 +310,17 @@ def attribute(k): f"phases (drone said ok, DUT host never delivered)") # separators: machine events must be the grep-able {"ev":"name",...} form # (docs/logging.md) — default json.dumps inserts spaces. + # flush: the one machine-event line lands atomically even if a consumer + # reads the stream live. The human report around it is deliberately on + # stdout — this is an offline report generator (the harness tees it into + # report.txt), same contract as rxq_analyze.py / pp109_starve_analyze.py, + # not a demo's runtime event plane. print(json.dumps({"ev": "arqe2e.verdict", "verdict": verdict, "acked_undelivered": total_au, "tail_suspect": tot_tail, "reports": len(reports), "ok": n_ok, "dut_pctrs": len(dut), "wit_pctrs": len(wit), - "base": base}, separators=(",", ":"))) + "base": base}, separators=(",", ":")), flush=True) return 3 if verdict == "INCONCLUSIVE-SHORT-RUN" else 0 From 2559ec7070311e98bb31f38f0ab6f491edc33414 Mon Sep 17 00:00:00 2001 From: Joseph <162703152+josephnef@users.noreply.github.com> Date: Mon, 3 Aug 2026 17:35:50 +0300 Subject: [PATCH 05/12] docs/logging.md: rx.ring schema catches up with the pool modes; rx.seq emitter gains duplex MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit rx.ring documents the reorder-pool/spsc-fat mode values, qdepth, and the new pool_dropped counter — including what a pool_dropped frame means under hardware ARQ (chip-ACKed, so the TX peer counts it delivered) and the counters' structural blind spot while the pump is frozen, which the per-frame rx.seq ledger covers. rx.seq is emitted by duplex too, where the SA gate is required. Co-Authored-By: Claude Opus 4.8 --- docs/logging.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/logging.md b/docs/logging.md index ffbf898..9822059 100644 --- a/docs/logging.md +++ b/docs/logging.md @@ -90,8 +90,8 @@ Emitters: L = library, RX/TX/... = demo. Optional fields in [brackets]; | `rx.body` | RX (`DEVOURER_DUMP_BODY`) | rate, rssi[2], evm[2], snr[2], crc, len, body hex | | `rx.corrupt` | RX (`DEVOURER_RX_DUMP_ALL`) | len, crc, icv, rate, bw, stbc, ldpc, sgi, rssi[2], evm[2], snr[2] | | `rx.txhit` | RX, TX | hits, total_rx, len, seq, paggr, ppdu, rate, bw, stbc, ldpc, ppdu_type — canonical-SA (57:42:75:05:d6:00) matcher; rate/ldpc prove what encoding was decoded (8814A reports ldpc=0 always — no HW indicator); ppdu_type is the AX RXD format nibble (7=HE_SU, 8=HE_ERSU; 255 on pre-AX chips) | -| `rx.seq` | RX (`DEVOURER_RX_PCTR`) | pctr, tsfl, seq, crc, paggr, ppdu — the ground-truth per-frame delivery sequence for the RX-ring loss study: pctr is the u32 the txdemo QoS-Data path stamps at MPDU offset 26, so gaps in it are per-frame loss; paggr/ppdu carry the aggregate structure the host-vs-RF discriminator keys on. Lean by design (no body hex) so the emit can't perturb the pump thread. SA gate follows `DEVOURER_RX_AGG_SA`, else canonical SA | -| `rx.ring` | L (`DEVOURER_RX_RING_MS`) | t, mode ("async"/"sync"), n_urbs, armed (URBs posted to the HCD and awaiting a frame — the depth that starves under a slow inline consumer), min_armed (low-water mark since the last emit), cb_max_us (worst inline-consume latency in the window), resubmit_fail, completions (cumulative URB callbacks), empties (cumulative callbacks that left the ring with zero posted URBs), pool_free (−1 = no host pool). The mechanism-proof telemetry: empties/completions is the host-starvation rate — near-0 under RF loss (the ring stays armed because frames don't arrive), high under host starvation (frames out-race resubmit and drain the ring). Counted in the callback, so robust to the pump-thread starvation that makes the periodic emit sparse | +| `rx.seq` | RX, duplex (`DEVOURER_RX_PCTR`) | pctr, tsfl, seq, crc, paggr, ppdu — the ground-truth per-frame delivery sequence for the RX-ring loss study: pctr is the u32 the txdemo QoS-Data path stamps at MPDU offset 26, so gaps in it are per-frame loss; paggr/ppdu carry the aggregate structure the host-vs-RF discriminator keys on. Lean by design (no body hex) so the emit can't perturb the pump thread. SA gate follows `DEVOURER_RX_AGG_SA` (required in duplex, whose canonical-SA `rx.frame` stream is a different transmitter), else canonical SA | +| `rx.ring` | L (`DEVOURER_RX_RING_MS`) | t, mode ("async"/"sync"/"reorder-pool"/"spsc-fat"), n_urbs, armed (URBs posted to the HCD and awaiting a frame — the depth that starves under a slow inline consumer), min_armed (low-water mark since the last emit), cb_max_us (worst inline-consume latency in the window), resubmit_fail, completions (cumulative URB callbacks), empties (cumulative callbacks that left the ring with zero posted URBs), pool_free (−1 = no host pool), qdepth (spsc-fat consumer-queue depth; 0 elsewhere), pool_dropped (cumulative frames dropped at spsc-fat pool exhaustion — each was already chip-ACKed, so a hardware-ARQ peer counts it delivered; `tests/arq_e2e_delivery.sh` measures it). The mechanism-proof telemetry: empties/completions is the host-starvation rate — near-0 under RF loss (the ring stays armed because frames don't arrive), high under host starvation (frames out-race resubmit and drain the ring). Counted in the callback, so robust to the pump-thread starvation that makes the periodic emit sparse — but blind while the pump itself is frozen: a stalled consumer drops frames these counters never see, which the per-frame `rx.seq` ledger exists to catch | | `rx.count` | TX (its RX thread) | total, len | | `rx.path` | RX (`DEVOURER_RX_ALLPATHS`) | seq, rssi[4], snr[4], evm[4] | | `rx.path_mask` | L (toggle spec) | t, mask "0xNN" | From 7f26e99cf2e5cbfd20b57f2cc8eef1285d88303f Mon Sep 17 00:00:00 2001 From: Joseph <162703152+josephnef@users.noreply.github.com> Date: Mon, 3 Aug 2026 17:43:20 +0300 Subject: [PATCH 06/12] arq-e2e: ship the uplink feeder the harness depends on; propagate the analyzer verdict MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit tests/pp109_uplink_feeder.py was still untracked from the pp109 bench set, so the harness could not run from a clean checkout — it is the PixelPilot-shaped burst generator (SET_RATE phase markers, per-phase burst counts) both benches share. And the analyzer's exit code (3 = INCONCLUSIVE-SHORT-RUN) now survives the tee: the harness exits with the verdict. Co-Authored-By: Claude Opus 4.8 --- tests/arq_e2e_delivery.sh | 2 + tests/pp109_uplink_feeder.py | 102 +++++++++++++++++++++++++++++++++++ 2 files changed, 104 insertions(+) create mode 100644 tests/pp109_uplink_feeder.py diff --git a/tests/arq_e2e_delivery.sh b/tests/arq_e2e_delivery.sh index 991285f..f3a8e03 100644 --- a/tests/arq_e2e_delivery.sh +++ b/tests/arq_e2e_delivery.sh @@ -199,4 +199,6 @@ sleep 1 python3 "$ROOT/tests/arq_e2e_analyze.py" \ --dut "$OUT/dut.jsonl" --drone "$OUT/drone.jsonl" --wit "$OUT/wit.jsonl" \ --phases "$PHASES" --cycles "$CYCLES" | tee "$OUT/report.txt" +ANALYZE_RC=${PIPESTATUS[0]} # tee would otherwise eat the analyzer verdict echo "[arq-e2e] logs: $OUT" +exit "$ANALYZE_RC" diff --git a/tests/pp109_uplink_feeder.py b/tests/pp109_uplink_feeder.py new file mode 100644 index 0000000..54b924e --- /dev/null +++ b/tests/pp109_uplink_feeder.py @@ -0,0 +1,102 @@ +#!/usr/bin/env python3 +"""Feed examples/duplex's stdin with PixelPilot's uplink traffic shape. + +PixelPilot's ground station sends its adaptive-link + mavlink uplink as +wfb-ng FEC blocks: k=1 n=5 means every message is put on the air five times, and +Transmitter::sendPacket emits a block's fragments back-to-back with no spacing. +Measured on the phone: ~102 frames/s arriving as bursts of 10-12 frames inside +1-2 ms, once per ~100 ms adaptive-link period. That burst shape is the input +here -- BURST frames written to the pipe at once (the duplex TX thread drains +them back-to-back), then idle until the next period. + +The run is a sequence of PHASES, each PHASE_S seconds. A phase is either idle +(no uplink at all -- the "muted" control) or an uplink at one rate. Every phase +opens with a SET_RATE control op, which makes duplex emit exactly one +`stream.ctl` JSONL line: the Nth `stream.ctl` in duplex's event stream is the +start of the Nth phase, so the analyzer can bin `rx.frame` counts per phase from +file order alone -- no clock correlation between two processes. + +Alternating idle/uplink phases inside ONE duplex session is deliberate: it holds +the RF path, the chip's thermal state and the flooder constant across arms, so +the only thing changing is whether (and how) this adapter transmits. + + python3 tests/pp109_uplink_feeder.py --phases idle,6M,MCS0,MCS7 --cycles 4 +""" +import argparse +import struct +import sys +import time + +CTL_SET_RATE = 2 + + +def write_ctl(out, op, payload=b""): + """ -- duplex's control escape.""" + body = bytes([op]) + payload + out.write(struct.pack("" or ":" -- the per-burst frame count is + # the second, rate-independent way to scale airtime, so a rate sweep and a + # burst sweep can disagree and that disagreement is informative. + phases = [p.strip() for p in a.phases.split(",") if p.strip()] + out = sys.stdout.buffer + body = bytes(range(256)) * ((a.psdu // 256) + 1) + body = body[:a.psdu] + + time.sleep(a.warmup_s) + plan = [] + for c in range(a.cycles): + for ph in phases: + # Phase marker + live rate switch. 'idle' still switches (to 6M) so + # every phase costs exactly one stream.ctl -- the marker must not + # itself differ between arms. + rate, burst = (ph.split(":", 1) + [None])[:2] if ":" in ph else (ph, None) + burst = int(burst) if burst else a.burst + spec = "6M" if rate == "idle" else rate + write_ctl(out, CTL_SET_RATE, spec.encode()) + t_end = time.monotonic() + a.phase_s + sent = 0 + if rate == "idle": + time.sleep(a.phase_s) + else: + period = a.period_ms / 1000.0 + nxt = time.monotonic() + while time.monotonic() < t_end: + for _ in range(burst): + write_psdu(out, body) + out.flush() + sent += burst + nxt += period + d = nxt - time.monotonic() + if d > 0: + time.sleep(d) + plan.append((c, ph, sent)) + print(f"[feeder] cycle={c} phase={ph} frames={sent}", file=sys.stderr, + flush=True) + print("[feeder] done: " + ";".join(f"{c}:{p}:{n}" for c, p, n in plan), + file=sys.stderr, flush=True) + + +if __name__ == "__main__": + main() From 37b67ea91192d46db24f74bfa76e3a3fe3d230bd Mon Sep 17 00:00:00 2001 From: Joseph <162703152+josephnef@users.noreply.github.com> Date: Mon, 3 Aug 2026 17:54:22 +0300 Subject: [PATCH 07/12] arq-e2e round 3 review: float-safe span math, boundary attribution bucket, feeder BrokenPipe exit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - SPAN and the preflight sleep compute in float (awk) — fractional WARMUP_S/PHASE_S no longer truncate the drone's timeout short of the feeder's real runtime. - An undelivered frame whose nearest delivered neighbours straddle a phase edge lands in its own "boundary" row instead of silently biasing the earlier phase's count. Total acked_undelivered is invariant (re-checked on the round-1 logs: 5,739). - The feeder exits quietly when its pipe reader goes away (bench teardown or duplex crash) instead of tracebacking; os._exit skips the interpreter's flush-at-exit re-raise. Co-Authored-By: Claude Opus 4.8 --- tests/arq_e2e_analyze.py | 31 +++++++++++++++++++++---------- tests/arq_e2e_delivery.sh | 9 ++++++--- tests/pp109_uplink_feeder.py | 11 ++++++++++- 3 files changed, 37 insertions(+), 14 deletions(-) diff --git a/tests/arq_e2e_analyze.py b/tests/arq_e2e_analyze.py index 95bbfa4..48bb1ba 100644 --- a/tests/arq_e2e_analyze.py +++ b/tests/arq_e2e_analyze.py @@ -121,6 +121,14 @@ def phase_names(phases, cycles): return [f"{p}" for _ in range(cycles) for p in per] +def phase_label(p, names): + if p == -1: + return "warmup" + if p == -3: + return "boundary" # undelivered, nearest neighbours straddle phases + return names[p] if p is not None and 0 <= p < len(names) else f"?{p}" + + def main(): ap = argparse.ArgumentParser() ap.add_argument("--dut", required=True) @@ -183,16 +191,21 @@ def phase_of_idx(idx): delivered_sorted = sorted(k_phase) def attribute(k): - """Phase of an UNdelivered frame k: nearest delivered neighbours.""" + """Phase of an UNdelivered frame k: nearest delivered neighbours. + Neighbours straddling a phase edge get the distinct 'boundary' bucket + (-3) — silently picking a side would bias per-phase counts.""" import bisect i = bisect.bisect_left(delivered_sorted, k) lo = delivered_sorted[i - 1] if i else None hi = delivered_sorted[i] if i < len(delivered_sorted) else None - pl = k_phase.get(lo, -2) - ph = k_phase.get(hi, -2) - if pl == ph: - return pl - return ph if hi is not None and lo is None else pl + if lo is None and hi is None: + return -1 + if lo is None: + return k_phase[hi] + if hi is None: + return k_phase[lo] + pl, ph = k_phase[lo], k_phase[hi] + return pl if pl == ph else -3 # Ring telemetry per phase: min armed depth + empties/completions deltas. ring_by_phase = defaultdict(lambda: {"min_armed": None, "empties": 0, @@ -268,8 +281,7 @@ def attribute(k): total_au = 0 for p in sorted(per): st = per[p] - name = "warmup" if p == -1 else ( - names[p] if 0 <= p < len(names) else f"?{p}") + name = phase_label(p, names) okp = 100.0 * st["ok"] / st["reports"] if st["reports"] else 0.0 rg = ring_by_phase.get(p, {}) total_au += st["acked_undelivered"] @@ -285,8 +297,7 @@ def attribute(k): if detail: print("first acked-undelivered frames (k, retries, phase, witnessed):") for k, r, p, w in detail: - name = "warmup" if p == -1 else ( - names[p] if p is not None and 0 <= p < len(names) else f"?{p}") + name = phase_label(p, names) print(f" k={k} retries={r} phase={name} wit={'Y' if w else 'n'}") # Ledger identity: every ok'd frame must be delivered (minus the au set), # plus the ok=0-but-delivered strays. Holding to a few frames certifies diff --git a/tests/arq_e2e_delivery.sh b/tests/arq_e2e_delivery.sh index f3a8e03..c958fe0 100644 --- a/tests/arq_e2e_delivery.sh +++ b/tests/arq_e2e_delivery.sh @@ -75,9 +75,12 @@ for b in duplex txdemo rxdemo; do [ -x "$BUILD/$b" ] || { echo "build $b first"; exit 3; } done -# Total span the drone must cover: feeder warmup + all phases. +# Total span the drone must cover: feeder warmup + all phases. Float-safe: +# WARMUP_S/PHASE_S may be fractional and truncating them would shorten the +# drone's timeout below the feeder's real runtime. NPHASES=$(awk -F, '{print NF}' <<<"$PHASES") -SPAN=$(( ${WARMUP_S%.*} + CYCLES * NPHASES * ${PHASE_S%.*} + 10 )) +SPAN=$(awk -v w="$WARMUP_S" -v c="$CYCLES" -v n="$NPHASES" -v p="$PHASE_S" \ + 'BEGIN{printf "%d", w + c*n*p + 11}') WIT_PIDF=""; DUT_PIDF=""; FEED_PIDF=""; DRONE_PIDF="" cleanup() { @@ -165,7 +168,7 @@ DRONE_PIDF=$! # --- preflight: fail fast with a diagnosis, not a zero-filled matrix -------- # Judged after warmup + one burst phase: every ledger must be alive by then. -sleep "$(( ${WARMUP_S%.*} + PREFLIGHT_S ))" +sleep "$(awk -v w="$WARMUP_S" -v p="$PREFLIGHT_S" 'BEGIN{print w + p}')" fail="" n_dut=$(grep -c '"ev":"rx.seq"' "$OUT/dut.jsonl" 2>/dev/null || true) n_wit=$(grep -c '"ev":"rx.seq"' "$OUT/wit.jsonl" 2>/dev/null || true) diff --git a/tests/pp109_uplink_feeder.py b/tests/pp109_uplink_feeder.py index 54b924e..e88224b 100644 --- a/tests/pp109_uplink_feeder.py +++ b/tests/pp109_uplink_feeder.py @@ -99,4 +99,13 @@ def main(): if __name__ == "__main__": - main() + try: + main() + except BrokenPipeError: + # The pipe reader (duplex) went away — a bench teardown or crash, not + # a feeder bug. Exit quietly; os._exit skips the interpreter's stdout + # flush-at-exit, which would raise the same error again. + print("[feeder] reader closed the pipe — stopping", file=sys.stderr, + flush=True) + import os + os._exit(1) From e4893272f5ae57fb2e238a6960da70d50641f499 Mon Sep 17 00:00:00 2001 From: Joseph <162703152+josephnef@users.noreply.github.com> Date: Mon, 3 Aug 2026 18:01:58 +0300 Subject: [PATCH 08/12] arq-e2e round 4 review: pool_dropped covers both drop causes explicitly; rx.ring sync-mode caveat MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pool_dropped deliberately counts every received frame discarded on the spsc-fat drop path — pool exhaustion and failed re-arm alike, both post-ACK host drops; the re-arm failure stays separable via resubmit_fail. The comment and docs/logging.md now say exactly that, and the rx.ring row notes sync mode emits a reduced line without the pool fields. Co-Authored-By: Claude Opus 4.8 --- docs/logging.md | 2 +- src/UsbTransport.cpp | 4 +++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/docs/logging.md b/docs/logging.md index 9822059..fed2dab 100644 --- a/docs/logging.md +++ b/docs/logging.md @@ -91,7 +91,7 @@ Emitters: L = library, RX/TX/... = demo. Optional fields in [brackets]; | `rx.corrupt` | RX (`DEVOURER_RX_DUMP_ALL`) | len, crc, icv, rate, bw, stbc, ldpc, sgi, rssi[2], evm[2], snr[2] | | `rx.txhit` | RX, TX | hits, total_rx, len, seq, paggr, ppdu, rate, bw, stbc, ldpc, ppdu_type — canonical-SA (57:42:75:05:d6:00) matcher; rate/ldpc prove what encoding was decoded (8814A reports ldpc=0 always — no HW indicator); ppdu_type is the AX RXD format nibble (7=HE_SU, 8=HE_ERSU; 255 on pre-AX chips) | | `rx.seq` | RX, duplex (`DEVOURER_RX_PCTR`) | pctr, tsfl, seq, crc, paggr, ppdu — the ground-truth per-frame delivery sequence for the RX-ring loss study: pctr is the u32 the txdemo QoS-Data path stamps at MPDU offset 26, so gaps in it are per-frame loss; paggr/ppdu carry the aggregate structure the host-vs-RF discriminator keys on. Lean by design (no body hex) so the emit can't perturb the pump thread. SA gate follows `DEVOURER_RX_AGG_SA` (required in duplex, whose canonical-SA `rx.frame` stream is a different transmitter), else canonical SA | -| `rx.ring` | L (`DEVOURER_RX_RING_MS`) | t, mode ("async"/"sync"/"reorder-pool"/"spsc-fat"), n_urbs, armed (URBs posted to the HCD and awaiting a frame — the depth that starves under a slow inline consumer), min_armed (low-water mark since the last emit), cb_max_us (worst inline-consume latency in the window), resubmit_fail, completions (cumulative URB callbacks), empties (cumulative callbacks that left the ring with zero posted URBs), pool_free (−1 = no host pool), qdepth (spsc-fat consumer-queue depth; 0 elsewhere), pool_dropped (cumulative frames dropped at spsc-fat pool exhaustion — each was already chip-ACKed, so a hardware-ARQ peer counts it delivered; `tests/arq_e2e_delivery.sh` measures it). The mechanism-proof telemetry: empties/completions is the host-starvation rate — near-0 under RF loss (the ring stays armed because frames don't arrive), high under host starvation (frames out-race resubmit and drain the ring). Counted in the callback, so robust to the pump-thread starvation that makes the periodic emit sparse — but blind while the pump itself is frozen: a stalled consumer drops frames these counters never see, which the per-frame `rx.seq` ledger exists to catch | +| `rx.ring` | L (`DEVOURER_RX_RING_MS`) | t, mode ("async"/"sync"/"reorder-pool"/"spsc-fat"), n_urbs, armed (URBs posted to the HCD and awaiting a frame — the depth that starves under a slow inline consumer), min_armed (low-water mark since the last emit), cb_max_us (worst inline-consume latency in the window), resubmit_fail, completions (cumulative URB callbacks), empties (cumulative callbacks that left the ring with zero posted URBs), pool_free (−1 = no host pool), qdepth (spsc-fat consumer-queue depth; 0 in the other ring modes), pool_dropped (cumulative received frames discarded on the spsc-fat drop path — pool exhausted, or a failed re-arm, the latter also ticking resubmit_fail; each was already chip-ACKed, so a hardware-ARQ peer counts it delivered; `tests/arq_e2e_delivery.sh` measures it). Sync mode emits a reduced line without the pool fields (no pool_free/qdepth/pool_dropped). The mechanism-proof telemetry: empties/completions is the host-starvation rate — near-0 under RF loss (the ring stays armed because frames don't arrive), high under host starvation (frames out-race resubmit and drain the ring). Counted in the callback, so robust to the pump-thread starvation that makes the periodic emit sparse — but blind while the pump itself is frozen: a stalled consumer drops frames these counters never see, which the per-frame `rx.seq` ledger exists to catch | | `rx.count` | TX (its RX thread) | total, len | | `rx.path` | RX (`DEVOURER_RX_ALLPATHS`) | seq, rssi[4], snr[4], evm[4] | | `rx.path_mask` | L (toggle spec) | t, mask "0xNN" | diff --git a/src/UsbTransport.cpp b/src/UsbTransport.cpp index 7580c3a..4111664 100644 --- a/src/UsbTransport.cpp +++ b/src/UsbTransport.cpp @@ -222,7 +222,9 @@ extern "C" void LIBUSB_CALL devourer_rx_cb(libusb_transfer *t) { * submit failed: preserve the pump's never-block invariant by re-arming * with the received buffer and DROPPING this frame — a bounded loss, vs the * cascade an inline consume would trigger. The chip already ACKed this - * frame (see the mode comment above): count it, never drop silently. + * frame (see the mode comment above): count every received frame dropped + * here, exhaustion and failed-re-arm alike — both are post-ACK host drops, + * and the re-arm failure is separable because it also ticks resubmit_fail. * `resubmit` gates the count: a teardown-window frame (stop requested) is * intentional loss, not an overload signal. */ if (resubmit && rlen > 0) From 8e15c9764013388378190674b919a2a3d0ff0fe7 Mon Sep 17 00:00:00 2001 From: Joseph <162703152+josephnef@users.noreply.github.com> Date: Mon, 3 Aug 2026 18:25:16 +0300 Subject: [PATCH 09/12] arq-e2e round 5 review: membership over key views, exact sync-mode rx.ring field list MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit find_base tests membership against the two ledgers' key views instead of materializing a ~1M-int union per analysis, and the rx.ring row now lists the sync-mode reduced line exactly as emitted (pool_free pinned at -1; no qdepth/pool_dropped/completions/empties) — verified against the emit site. Co-Authored-By: Claude Opus 4.8 --- docs/logging.md | 2 +- tests/arq_e2e_analyze.py | 14 ++++++++------ 2 files changed, 9 insertions(+), 7 deletions(-) diff --git a/docs/logging.md b/docs/logging.md index fed2dab..a6d9bf8 100644 --- a/docs/logging.md +++ b/docs/logging.md @@ -91,7 +91,7 @@ Emitters: L = library, RX/TX/... = demo. Optional fields in [brackets]; | `rx.corrupt` | RX (`DEVOURER_RX_DUMP_ALL`) | len, crc, icv, rate, bw, stbc, ldpc, sgi, rssi[2], evm[2], snr[2] | | `rx.txhit` | RX, TX | hits, total_rx, len, seq, paggr, ppdu, rate, bw, stbc, ldpc, ppdu_type — canonical-SA (57:42:75:05:d6:00) matcher; rate/ldpc prove what encoding was decoded (8814A reports ldpc=0 always — no HW indicator); ppdu_type is the AX RXD format nibble (7=HE_SU, 8=HE_ERSU; 255 on pre-AX chips) | | `rx.seq` | RX, duplex (`DEVOURER_RX_PCTR`) | pctr, tsfl, seq, crc, paggr, ppdu — the ground-truth per-frame delivery sequence for the RX-ring loss study: pctr is the u32 the txdemo QoS-Data path stamps at MPDU offset 26, so gaps in it are per-frame loss; paggr/ppdu carry the aggregate structure the host-vs-RF discriminator keys on. Lean by design (no body hex) so the emit can't perturb the pump thread. SA gate follows `DEVOURER_RX_AGG_SA` (required in duplex, whose canonical-SA `rx.frame` stream is a different transmitter), else canonical SA | -| `rx.ring` | L (`DEVOURER_RX_RING_MS`) | t, mode ("async"/"sync"/"reorder-pool"/"spsc-fat"), n_urbs, armed (URBs posted to the HCD and awaiting a frame — the depth that starves under a slow inline consumer), min_armed (low-water mark since the last emit), cb_max_us (worst inline-consume latency in the window), resubmit_fail, completions (cumulative URB callbacks), empties (cumulative callbacks that left the ring with zero posted URBs), pool_free (−1 = no host pool), qdepth (spsc-fat consumer-queue depth; 0 in the other ring modes), pool_dropped (cumulative received frames discarded on the spsc-fat drop path — pool exhausted, or a failed re-arm, the latter also ticking resubmit_fail; each was already chip-ACKed, so a hardware-ARQ peer counts it delivered; `tests/arq_e2e_delivery.sh` measures it). Sync mode emits a reduced line without the pool fields (no pool_free/qdepth/pool_dropped). The mechanism-proof telemetry: empties/completions is the host-starvation rate — near-0 under RF loss (the ring stays armed because frames don't arrive), high under host starvation (frames out-race resubmit and drain the ring). Counted in the callback, so robust to the pump-thread starvation that makes the periodic emit sparse — but blind while the pump itself is frozen: a stalled consumer drops frames these counters never see, which the per-frame `rx.seq` ledger exists to catch | +| `rx.ring` | L (`DEVOURER_RX_RING_MS`) | t, mode ("async"/"sync"/"reorder-pool"/"spsc-fat"), n_urbs, armed (URBs posted to the HCD and awaiting a frame — the depth that starves under a slow inline consumer), min_armed (low-water mark since the last emit), cb_max_us (worst inline-consume latency in the window), resubmit_fail, completions (cumulative URB callbacks), empties (cumulative callbacks that left the ring with zero posted URBs), pool_free (−1 = no host pool), qdepth (spsc-fat consumer-queue depth; 0 in the other ring modes), pool_dropped (cumulative received frames discarded on the spsc-fat drop path — pool exhausted, or a failed re-arm, the latter also ticking resubmit_fail; each was already chip-ACKed, so a hardware-ARQ peer counts it delivered; `tests/arq_e2e_delivery.sh` measures it). Sync mode emits a reduced line (pool_free pinned at −1; no qdepth/pool_dropped/completions/empties). The mechanism-proof telemetry: empties/completions is the host-starvation rate — near-0 under RF loss (the ring stays armed because frames don't arrive), high under host starvation (frames out-race resubmit and drain the ring). Counted in the callback, so robust to the pump-thread starvation that makes the periodic emit sparse — but blind while the pump itself is frozen: a stalled consumer drops frames these counters never see, which the per-frame `rx.seq` ledger exists to catch | | `rx.count` | TX (its RX thread) | total, len | | `rx.path` | RX (`DEVOURER_RX_ALLPATHS`) | seq, rssi[4], snr[4], evm[4] | | `rx.path_mask` | L (toggle spec) | t, mask "0xNN" | diff --git a/tests/arq_e2e_analyze.py b/tests/arq_e2e_analyze.py index 48bb1ba..55f5603 100644 --- a/tests/arq_e2e_analyze.py +++ b/tests/arq_e2e_analyze.py @@ -84,9 +84,9 @@ def find_base(reports, rel, pctr_sets, max_pctr): return None, 0, 0 tag0 = reports[0]["tag"] ok_rel = [rel[j] for j, rp in enumerate(reports) if rp["ok"]] - union = set() - for s in pctr_sets: - union |= set(s) + # Membership tests run against the ledgers' own key views — materializing + # a union would copy ~1M ints per analysis for nothing. + d0, d1 = pctr_sets best, second, best_m = -1, -1, 0 m_hi = (max_pctr + 4096) // 256 + 2 # Reports are send-ordered and dense, so the first report's absolute index @@ -96,14 +96,15 @@ def find_base(reports, rel, pctr_sets, max_pctr): # right from wrong decisively) and only the winner is re-scored in full — # otherwise non-joining logs cost O(m_hi * n_ok) and the analyzer hangs # instead of failing fast. - lo_pctr = min(union) if union else 0 + lo_pctr = min((min(d0) if d0 else 0), (min(d1) if d1 else 0)) near = range(max(0, (lo_pctr - 2048)) // 256, min(m_hi, (lo_pctr + 2048) // 256 + 1)) sample = ok_rel[:2000] for candidates, pool in ((near, ok_rel), (range(m_hi), sample)): for m in candidates: base = tag0 + 256 * m - score = sum(1 for r in pool if (base + r) in union) + score = sum(1 for r in pool + if (base + r) in d0 or (base + r) in d1) if score > best: second, best, best_m = best, score, m elif score > second: @@ -112,7 +113,8 @@ def find_base(reports, rel, pctr_sets, max_pctr): break # strong hit — skip / end the wider scan best, second = -1, -1 # sampled scores are not comparable to full ones full = sum(1 for r in ok_rel - if (tag0 + 256 * best_m + r) in union) + if (tag0 + 256 * best_m + r) in d0 + or (tag0 + 256 * best_m + r) in d1) return tag0 + 256 * best_m - rel[0], full, second From ec8d765e05769dc4fdc0f89e420b157992abadb9 Mon Sep 17 00:00:00 2001 From: Joseph <162703152+josephnef@users.noreply.github.com> Date: Mon, 3 Aug 2026 18:32:57 +0300 Subject: [PATCH 10/12] arq-e2e: SPAN ceils instead of truncating MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The drone timeout is an upper bound on the feeder's runtime, so fractional WARMUP_S/PHASE_S round up rather than down — truncation leaned on the slack constant to stay safe. Co-Authored-By: Claude Opus 4.8 --- tests/arq_e2e_delivery.sh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/arq_e2e_delivery.sh b/tests/arq_e2e_delivery.sh index c958fe0..05d6982 100644 --- a/tests/arq_e2e_delivery.sh +++ b/tests/arq_e2e_delivery.sh @@ -80,7 +80,7 @@ done # drone's timeout below the feeder's real runtime. NPHASES=$(awk -F, '{print NF}' <<<"$PHASES") SPAN=$(awk -v w="$WARMUP_S" -v c="$CYCLES" -v n="$NPHASES" -v p="$PHASE_S" \ - 'BEGIN{printf "%d", w + c*n*p + 11}') + 'BEGIN{v = w + c*n*p + 11; printf "%d", (v == int(v)) ? v : int(v) + 1}') WIT_PIDF=""; DUT_PIDF=""; FEED_PIDF=""; DRONE_PIDF="" cleanup() { From 24ae535f2d39ae20530fe6099af8d83499c1e03a Mon Sep 17 00:00:00 2001 From: Joseph <162703152+josephnef@users.noreply.github.com> Date: Mon, 3 Aug 2026 18:38:50 +0300 Subject: [PATCH 11/12] arq-e2e analyzer: max over key views, no list materialization Co-Authored-By: Claude Opus 4.8 --- tests/arq_e2e_analyze.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/arq_e2e_analyze.py b/tests/arq_e2e_analyze.py index 55f5603..0099f90 100644 --- a/tests/arq_e2e_analyze.py +++ b/tests/arq_e2e_analyze.py @@ -150,7 +150,7 @@ def main(): return 3 rel = unwrap_tags(reports) - max_pctr = max(list(dut.keys()) + list(wit.keys()) + [0]) + max_pctr = max(max(dut, default=0), max(wit, default=0)) base, best, second = find_base(reports, rel, (dut.keys(), wit.keys()), max_pctr) n_ok = sum(1 for r in reports if r["ok"]) @@ -231,7 +231,7 @@ def attribute(k): # (measured: 19 consecutive "losses" at the exact end of a run that were # an unflushed-stdout artifact, not RF or USB). TAIL_GUARD = 512 - max_index = max([base + rel[-1]] + list(dut.keys()) + list(wit.keys())) + max_index = max(base + rel[-1], max(dut, default=0), max(wit, default=0)) tail_cutoff = max_index - TAIL_GUARD # On a run shorter than 2x the guard the tail window swallows most of the # frames and a NOT-REPRODUCED verdict would be vacuous — refuse to conclude From 49feee108307b9340e36cec8d7c71bbf026379bd Mon Sep 17 00:00:00 2001 From: Joseph <162703152+josephnef@users.noreply.github.com> Date: Mon, 3 Aug 2026 18:58:51 +0300 Subject: [PATCH 12/12] arq-e2e: escape the ERE metacharacters in the cleanup pkill path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit $BUILD is caller-controlled and pkill -f matches an ERE — a path like ".../build+asan" would broaden the match past this checkout's binaries. Co-Authored-By: Claude Opus 4.8 --- tests/arq_e2e_delivery.sh | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/tests/arq_e2e_delivery.sh b/tests/arq_e2e_delivery.sh index 05d6982..00e2917 100644 --- a/tests/arq_e2e_delivery.sh +++ b/tests/arq_e2e_delivery.sh @@ -90,7 +90,10 @@ cleanup() { done # Backstop scoped to THIS checkout's binaries (full-path match) — a bare # `pkill -x duplex` from root would kill unrelated instances host-wide. - for b in duplex txdemo rxdemo; do pkill -f "^$BUILD/$b" 2>/dev/null; done + # $BUILD is caller-controlled and pkill -f takes an ERE: escape the + # metacharacters so a path like ".../build+asan" cannot broaden the match. + esc_build=$(printf '%s' "$BUILD" | sed 's/[][\\.^$*+?(){}|]/\\&/g') + for b in duplex txdemo rxdemo; do pkill -f "^$esc_build/$b" 2>/dev/null; done rm -f "$BLACKLIST" "$OUT/fifo" wait 2>/dev/null }