rx.parse_abort: count the abandoned-aggregate RX loss on every generation

docs/logging.md

@@ -91,6 +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, rate (hw rate index of this copy — retransmissions of one pctr can air at different rates, so per-copy rate + tsfl order reconstructs the fw fallback ladder on air, `tests/retry_ladder_probe.sh`) — 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.parse_abort` | RX (every generation, always on) | t, off, buf_len, remaining, frame_len, drvinfo, shift, total (cumulative) — the RX descriptor walk abandoned the rest of a bulk-IN aggregate; semantics doc-commented at `src/RxParseAbort.h`. No event in a session = the walk never aborted |
 | `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 at spsc-fat pool exhaustion under the `drop` policy — 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), pool_stalls (the `backpressure`-policy counterpart, cumulative URB park events: the payload still reaches the consumer, the ring shrinks and the chip declines further ACKs, so overload loss stays ARQ-visible — `DEVOURER_RX_POOL_EXHAUST`, default backpressure; `tests/arq_e2e_delivery.sh` measures both). Sync mode emits a reduced line (pool_free pinned at −1; no qdepth/pool_dropped/pool_stalls/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] |

src/RxParseAbort.h

@@ -0,0 +1,52 @@
+#ifndef DEVOURER_RX_PARSE_ABORT_H
+#define DEVOURER_RX_PARSE_ABORT_H
+
+/* rx.parse_abort — the RX descriptor walk hit a malformed/truncated
+ * descriptor mid-aggregate and abandoned the rest of the bulk-IN buffer.
+ * Every abandoned frame was already admitted by the chip (and, with an ACK
+ * responder armed, already ACKed to the peer), so this is post-admission
+ * loss a hardware-ARQ peer counts as delivered — it must never be silent.
+ * Normal end-of-aggregate zero padding (all-zero remainder) is excluded.
+ * Shared by every generation's RX walk; schema: docs/logging.md. */
+
+#include <cstddef>
+#include <cstdint>
+
+#include "Event.h"
+
+namespace devourer {
+
+/* Returns true when the remainder was a real abort (event emitted),
+ * false for benign all-zero padding. `total` is the caller's cumulative
+ * abort counter, incremented on emit. */
+inline bool emit_rx_parse_abort(EventSink &sink, const uint8_t *rem,
+                                size_t rem_len, long long off,
+                                long long buf_len, long long frame_len,
+                                long long drvinfo, long long shift,
+                                long long &total) {
+  if (!sink.enabled())
+    return false;
+  bool all_zero = true;
+  for (size_t i = 0; i < rem_len; ++i)
+    if (rem[i] != 0) {
+      all_zero = false;
+      break;
+    }
+  if (all_zero)
+    return false;
+  ++total;
+  Ev(sink, "rx.parse_abort")
+      .t()
+      .f("off", off)
+      .f("buf_len", buf_len)
+      .f("remaining", static_cast<long long>(rem_len))
+      .f("frame_len", frame_len)
+      .f("drvinfo", drvinfo)
+      .f("shift", shift)
+      .f("total", total);
+  return true;
+}
+
+} // namespace devourer
+
+#endif /* DEVOURER_RX_PARSE_ABORT_H */

src/jaguar1/FrameParser.cpp

@@ -1,5 +1,7 @@
 #include "FrameParser.h"
 
+#include "RxParseAbort.h" /* rx.parse_abort — abandoned-aggregate event */
+
 #define CONFIG_USB_RX_AGGREGATION 1
 
 #define RXDESC_SIZE 24
@@ -182,6 +184,11 @@ std::vector<Packet> FrameParser::recvbuf2recvframe(std::span<uint8_t> ptr) {
           "RX Warning!,pkt_len <= 0 or pkt_offset > transfer_len; pkt_len: "
           "{}, pkt_offset: {}, transfer_len: {}",
           pattrib.pkt_len, pkt_offset, pbuf.size());
+      devourer::emit_rx_parse_abort(
+          _logger->events(), pbuf.data(), pbuf.size(),
+          static_cast<long long>(pbuf.data() - ptr.data()),
+          static_cast<long long>(ptr.size()), pattrib.pkt_len,
+          pattrib.drvinfo_sz, pattrib.shift_sz, _parse_aborts);
       break;
     }
 

src/jaguar1/FrameParser.h

@@ -235,6 +235,7 @@ enum _PUBLIC_ACTION
 class FrameParser
 {
     Logger_t _logger;
+    long long _parse_aborts = 0; /* cumulative rx.parse_abort count */
 
   public:
     FrameParser(Logger_t logger);

src/jaguar2/RtlJaguar2Device.cpp

@@ -17,6 +17,7 @@
 #include "RadiotapPeek.h"
 #include "RadiotapTxFlags.h" /* HT MCS field decoder (LDPC/STBC) */
 #include "TxAggPlan.h"
+#include "RxParseAbort.h" /* rx.parse_abort — abandoned-aggregate event */
 #include "TxReport.h"
 
 #include "BeamformingSounder.h"
@@ -548,15 +549,22 @@ void RtlJaguar2Device::StartRxLoop(Action_ParsedRadioPacket packetProcessor) {
   /* RX loop: async bulk-IN URB queue; walk the aggregated 8822B RX descriptors
    * per completion and hand each PSDU to the packet processor. */
   uint64_t frames = 0, reads = 0;
+  long long parse_aborts = 0;
   auto on_data = [&](const uint8_t *data, int n) {
     cfo_tick();
     if (++reads <= 8)
       _logger->info("Jaguar2 RX: completion #{} -> {} bytes", reads, n);
     uint32_t off = 0;
     while (off + jaguar2::RXDESC_SIZE_8822B <= static_cast<uint32_t>(n)) {
-      jaguar2::Rx8822bFrame f;
-      if (!jaguar2::parse_rx_8822b(data + off, static_cast<size_t>(n) - off, f))
+      jaguar2::Rx8822bFrame f{};
+      if (!jaguar2::parse_rx_8822b(data + off, static_cast<size_t>(n) - off,
+                                   f)) {
+        devourer::emit_rx_parse_abort(_logger->events(), data + off,
+                                      static_cast<size_t>(n) - off, off, n,
+                                      f.frame_len, f.drvinfo_size, f.shift,
+                                      parse_aborts);
         break;
+      }
       if (_packetProcessor) {
         Packet p{};
         p.RxAtrib.pkt_len = static_cast<uint16_t>(f.frame_len);

src/jaguar3/RtlJaguar3Device.cpp

@@ -12,6 +12,7 @@
 #include "RadiotapPeek.h"   /* send_packets batch pre-parse */
 #include "RadiotapTxFlags.h" /* HT MCS field decoder (LDPC/STBC) */
 #include "TxAggPlan.h"      /* USB TX aggregation URB packing */
+#include "RxParseAbort.h" /* rx.parse_abort — abandoned-aggregate event */
 #include "TxReport.h"     /* CCX TX-status report decode + tx.report event */
 
 #include "BeamformingSounder.h" /* generation-neutral BF self-sounding recipe */
@@ -274,15 +275,22 @@ void RtlJaguar3Device::StartRxLoop(Action_ParsedRadioPacket packetProcessor) {
                     static_cast<int>(avg_khz), _xtal_cap);
   };
   /* Process one bulk-IN completion: walk the aggregated 8822C RX descriptors. */
+  long long parse_aborts = 0;
   auto on_data = [&](const uint8_t *data, int n) {
     cfo_tick();
     if (++reads <= 8)
       _logger->info("Jaguar3 RX: async completion #{} -> {} bytes", reads, n);
     uint32_t off = 0;
     while (off + jaguar3::RXDESC_SIZE_8822C <= static_cast<uint32_t>(n)) {
-      jaguar3::Rx8822cFrame f;
-      if (!jaguar3::parse_rx_8822c(data + off, static_cast<size_t>(n) - off, f))
+      jaguar3::Rx8822cFrame f{};
+      if (!jaguar3::parse_rx_8822c(data + off, static_cast<size_t>(n) - off,
+                                   f)) {
+        devourer::emit_rx_parse_abort(_logger->events(), data + off,
+                                      static_cast<size_t>(n) - off, off, n,
+                                      f.frame_len, f.drvinfo_size, f.shift,
+                                      parse_aborts);
         break;
+      }
       if (_packetProcessor) {
         Packet p{};
         p.RxAtrib.pkt_len = static_cast<uint16_t>(f.frame_len);

src/kestrel/RtlKestrelDevice.cpp

@@ -11,6 +11,7 @@
 #include "KestrelLe.h"
 #include "RadiotapBuilder.h" /* build_stream_radiotap — host-injected trigger */
 #include "RadiotapPeek.h"
+#include "RxParseAbort.h" /* rx.parse_abort — abandoned-aggregate event */
 #include "RateDefinitions.h" /* MGN_* rate enum */
 #include "MacRegAx.h"
 #include "SignalStop.h" /* g_devourer_should_stop — set by demo signal handlers */
@@ -320,15 +321,21 @@ void RtlKestrelDevice::StartRxLoop(Action_ParsedRadioPacket packetProcessor) {
    * divergence between the dies. */
   const uint16_t drv_info_unit =
       _variant == kestrel::ChipVariant::C8852C ? 16 : 8;
+  long long parse_aborts = 0;
   _device.bulk_read_async_loop(
       32768, 8,
       [&, drv_info_unit](const uint8_t *data, int n) {
         uint32_t off = 0;
         while (off + 16 <= static_cast<uint32_t>(n)) {
           kestrel::KestrelRxFrame f;
           if (!kestrel::parse_rx_8852b(data + off, static_cast<size_t>(n) - off,
-                                       f, drv_info_unit))
+                                       f, drv_info_unit)) {
+            devourer::emit_rx_parse_abort(_logger->events(), data + off,
+                                          static_cast<size_t>(n) - off, off, n,
+                                          f.payload_len, f.drvinfo_size,
+                                          f.shift, parse_aborts);
             break;
+          }
           if (f.rpkt_type == kestrel::RPKT_TYPE_PPDU && f.payload_len >= 8) {
             /* Full physts parse (header per-path rssi_td + IE01 avg SNR +
              * IE04..07 per-path SNR/EVM pages) — kestrel::parse_physts_8852.

tests/parse_abort_smoke.sh

@@ -0,0 +1,64 @@
+#!/usr/bin/env bash
+#
+# parse_abort_smoke.sh — per-generation ambient-RX smoke for rx.parse_abort.
+#
+# The rx.parse_abort event (src/RxParseAbort.h) fires when a generation's RX
+# descriptor walk abandons a bulk-IN buffer on a malformed descriptor. Two
+# properties need hardware on every family: frames still flow (the walk is
+# untouched on the success path), and the all-zero-padding exclusion holds for
+# that family's aggregate format (no spurious event flood on ambient traffic).
+#
+#   sudo bash tests/parse_abort_smoke.sh          # every known plugged DUT
+#   DUTS="0x8813 0xb812" sudo bash tests/parse_abort_smoke.sh
+set -u
+
+ROOT="$(cd "$(dirname "$0")/.." && pwd)"
+BUILD=${BUILD:-$ROOT/build}
+CH=${CH:-6}          # 2.4 GHz: ambient beacons guarantee RX traffic
+DWELL_S=${DWELL_S:-20}
+# J1 8814AU, J2 8822BU, J3 8812CU, Kestrel. The J3 default is the 8812CU, not
+# the 8822EU: the 8822E's DPDT front end decodes no ambient 2.4 GHz on this
+# bench (green init, DIG sees energy, zero frames) while its 5 GHz RX is
+# proven — the walk under test is identical on both dies.
+DUTS=${DUTS:-"0x8813 0xb812 0xc812 0x0101"}
+OUT=${OUT:-/tmp/parse-abort-smoke}
+
+[ "$(id -u)" = 0 ] || { echo "must run as root"; exit 3; }
+[ -x "$BUILD/rxdemo" ] || { echo "build rxdemo first"; exit 3; }
+mkdir -p "$OUT"
+
+MODS="rtw88_8812au rtw88_8821au rtw88_8822bu rtw88_8814au rtw88_8822cu rtw88_8822eu rtw89_8852bu rtw89_8852cu"
+BLACKLIST=/run/modprobe.d/zz-temp-blacklist-pabort.conf
+cleanup() {
+  trap - EXIT INT TERM
+  esc_build=$(printf '%s' "$BUILD" | sed 's/[][\\.^$*+?(){}|]/\\&/g')
+  pkill -f "^$esc_build/rxdemo" 2>/dev/null
+  rm -f "$BLACKLIST"
+  wait 2>/dev/null
+}
+trap cleanup EXIT INT TERM
+mkdir -p "$(dirname "$BLACKLIST")"
+: > "$BLACKLIST"
+for m in $MODS; do echo "blacklist $m" >> "$BLACKLIST"; modprobe -r "$m" 2>/dev/null; done
+
+rc=0
+for pid in $DUTS; do
+  vid=0x0bda
+  [ "$pid" = "0x0101" ] && vid=0x35bc
+  log="$OUT/rx-${vid#0x}${pid#0x}.jsonl"
+  echo "[pabort] DUT $vid:$pid — ${DWELL_S}s ambient RX on ch$CH"
+  env DEVOURER_VID="$vid" DEVOURER_PID="$pid" DEVOURER_CHANNEL="$CH" \
+      DEVOURER_RX_AGG_SA=any \
+      DEVOURER_LOG_LEVEL=warn DEVOURER_EVENTS=stdout \
+      timeout -s INT "$DWELL_S" "$BUILD/rxdemo" >"$log" 2>"$OUT/rx-${vid#0x}${pid#0x}.err"
+  pkts=$(grep -cF '"ev":"rx.pkt"' "$log" || true)
+  frames=$(grep -cF '"ev":"rx.frame"' "$log" || true)
+  aborts=$(grep -cF '"ev":"rx.parse_abort"' "$log" || true)
+  verdict=OK
+  # rx.pkt samples (first 10 + every 100th) — >=2 proves the walk delivers.
+  [ "${pkts:-0}" -ge 2 ] || { verdict="FAIL(no-rx)"; rc=1; }
+  [ "${aborts:-0}" -eq 0 ] || { verdict="FAIL(aborts=$aborts)"; rc=1; }
+  echo "[pabort] $vid:$pid rx.pkt=$pkts rx.frame=$frames parse_aborts=$aborts -> $verdict"
+done
+echo "[pabort] logs: $OUT"
+exit "$rc"