ARQ end-to-end delivery bench: three-ledger harness; count spsc-fat pool-exhaustion drops

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).

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 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" |

examples/duplex/main.cpp

@@ -123,6 +123,50 @@ 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");
+  const long v = e ? std::strtol(e, nullptr, 0) : 100L;
+  return v > 0 ? v : 100L; /* 0/garbage would divide-by-zero the modulo */
+}();
+/* 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<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 +192,38 @@ static void packet_processor(const Packet &packet) {
     }
     return;
   }
+  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);
+    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 && (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;

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 +

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<unsigned long long> pool_dropped{0};
   std::mutex queue_mu;
   std::condition_variable queue_cv;
   std::deque<std::pair<uint8_t *, int>> queue;
@@ -210,7 +221,14 @@ 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 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)
+      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 +450,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));
       }
     }
   }