Windowed RX receipts: app-layer delivery truth, frame-exact against the rx.seq ledger (#363 step 3)

CMakeLists.txt

@@ -755,6 +755,17 @@ target_link_libraries(TxQuiesceSelftest PRIVATE devourer PkgConfig::libusb)
 
 add_test(NAME tx_quiesce_seam COMMAND TxQuiesceSelftest)
 
+# Headless guard for the windowed RX-receipt primitives (src/cell/RxReceipt.h):
+# TLV round-trip, ring eviction, late accounting, ledger merge idempotence,
+# strict-parse rejections. The on-air halves are tests/receipt_verify.py over
+# an arq_e2e run.
+add_executable(ReceiptSelftest
+    tests/receipt_selftest.cpp
+)
+target_link_libraries(ReceiptSelftest PRIVATE devourer)
+
+add_test(NAME receipt_roundtrip COMMAND ReceiptSelftest)
+
 # Headless guard for the active idle-noise-floor math (src/NoiseFloorMath.h): the
 # phydm sign/pwdb helpers the Jaguar1 0x0FA0 debug-port measurement uses. Pure
 # math, header-only.

docs/logging.md

@@ -111,6 +111,8 @@ Emitters: L = library, RX/TX/... = demo. Optional fields in [brackets];
 | `tx.agg` | L (`DEVOURER_TX_USB_AGG`, send_packets) | frames, bytes, shim, ok — one per multi-frame bulk-OUT URB |
 | `tx.report` | L (`DEVOURER_TX_REPORT`, CCX decode) | t, state (0=delivered, 1=retry-drop), ok, retries, final_rate, queue_time_raw, bmc, macid, fmt ("8812"\|"halmac"); halmac adds tag (SW_DEFINE echo), rts_retries, missed (fw-stuffed constant on Jaguar3 — tag gaps are the drop signal; `tests/txrpt_coverage_attrib.py`) — t is the achieved-report-rate timebase (the CCX emission ceiling is reports/s) |
 | `tx.status` | RX, duplex (C2H TX_RPT decode) | hoff, queue, retry, airtime_us, rate |
+| `tx.receipt` | TX (its RX thread, `DEVOURER_TX_RECEIPTS`) | t, fresh, total, covered, receipts, tlv hex — one event per absorbed windowed RX receipt (src/cell/RxReceipt.h), WITH the raw TLV so `tests/receipt_verify.py` can replay the merge and compare frame-exactly against the receiver's rx.seq ledger; decimating would break that comparison |
+| `receipt.tx` | duplex (`DEVOURER_RX_RECEIPT_MS`) | t, n, ok, tlv_len, late — receipt-injection cadence; late counts frames that arrived below the receipt window (reordered past coverage — size the window per the RxReceipt.h sizing rule) |
 | `tx.queue` | RX (`DEVOURER_QUEUE_POLL_MS`, 8814) | q1…q5 "0x%08x" |
 | `tx.contx` | TX (continuous mode) | mcs, t_ms |
 

docs/scheduled-mac.md

@@ -225,6 +225,20 @@ separately proven (`tests/ack_responder_check.sh`).
    99.71% against a 99.99% ledger truth — pessimistic by the ACK-loss
    asymmetry, the safe direction). Pick N ≥ fps/1300, or account
    report-less frames as "unknown".
+
+   Above both sits the app-layer truth tier: **windowed RX receipts**
+   (`src/cell/RxReceipt.h`) — the receiver's application notes every consumed
+   frame index in a sliding bitmap and mails overlapping receipt TLVs back on
+   its feedback path (`DEVOURER_RX_RECEIPT_MS` in duplex; the TX side absorbs
+   with `DEVOURER_TX_RECEIPTS` and emits `tx.receipt`). Neither the ACK
+   horizon nor the CCX ceiling applies: delivery is counted where it is
+   consumed. Measured frame-exact against the receiver's own rx.seq ledger —
+   126,594 frames clean and 349,455 frames under 150 ms consumer stalls at
+   2.4 k fps (`tests/receipt_verify.py`) — after one sizing lesson the header
+   documents: the window must exceed the worst backlog drain in frames (a
+   2,048-bit window leaked 2,846 delivered frames out of coverage when a
+   stalled spsc-fat pool drained ~3 k frames in one receipt interval; the
+   8192 default clears that bench worst case ~2.7×).
 2. **Closed-loop hardware ACK + autonomous retry is GO on Jaguar1 and
    Jaguar3** (100% delivery, retries ≈ 0.2–0.3) including retargeting an
    arbitrary UE MAC mid-session (re-arm `SetAckResponder`, change the

examples/duplex/main.cpp

@@ -65,6 +65,7 @@
 #include "RxPacket.h"
 #include "RadiotapBuilder.h"
 #include "RtlAdapter.h"
+#include "cell/RxReceipt.h"
 #if defined(DEVOURER_HAVE_JAGUAR1)
 #include "jaguar1/RtlJaguarDevice.h"
 #endif
@@ -145,6 +146,41 @@ static const bool g_seq_sa_set = []() {
   return true;
 }();
 
+/* DEVOURER_RX_RECEIPT_MS: windowed RX receipts (src/cell/RxReceipt.h) — the
+ * app-layer delivery truth. Every SA-matched pctr frame is noted in a sliding
+ * bitmap window, and every RECEIPT_MS a receipt frame (802.11 data, TA =
+ * DEVOURER_RX_RECEIPT_SA, RA = the tracked DEVOURER_RX_AGG_SA transmitter,
+ * body = the versioned TLV) is injected on this same handle — the feedback
+ * path. Windows overlap, so losing individual receipt frames costs nothing.
+ * Requires DEVOURER_RX_PCTR + DEVOURER_RX_AGG_SA (the ledger's identity). */
+static const long g_receipt_ms = []() {
+  const char *e = std::getenv("DEVOURER_RX_RECEIPT_MS");
+  return e ? std::strtol(e, nullptr, 0) : 0L;
+}();
+static uint8_t g_receipt_sa[6] = {0x02, 0x44, 0x52, 0x00, 0x00, 0x01};
+static const bool g_receipt_sa_ok = []() {
+  const char *e = std::getenv("DEVOURER_RX_RECEIPT_SA");
+  if (e == nullptr || *e == '\0')
+    return true; /* keep the default */
+  const auto m = devourer::parse_mac(e);
+  if (!m)
+    return false;
+  std::memcpy(g_receipt_sa, m->data(), 6);
+  return true;
+}();
+/* DEVOURER_RX_RECEIPT_WINDOW: bits of receipt coverage (default 8192). Size
+ * it past the worst backlog drain — see the sizing note in RxReceipt.h. */
+static const uint16_t g_receipt_window_bits = []() {
+  const char *e = std::getenv("DEVOURER_RX_RECEIPT_WINDOW");
+  const long v = e ? std::strtol(e, nullptr, 0) : 8192L;
+  return static_cast<uint16_t>(v < 64 ? 64 : v > 65535 ? 65535 : v);
+}();
+static devourer::cell::ReceiptWindow g_receipt_window{g_receipt_window_bits};
+
+/* Concurrent send_packet callers (the stdin TX thread + the receipt timer)
+ * serialize here — the per-generation send paths are single-caller. */
+static std::mutex g_send_mu;
+
 /* 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
@@ -214,6 +250,8 @@ static void packet_processor(const Packet &packet) {
       std::memcmp(packet.Data.data() + 10, g_seq_sa, 6) == 0) {
     uint32_t pctr;
     std::memcpy(&pctr, packet.Data.data() + 26, 4);
+    if (g_receipt_ms > 0)
+      g_receipt_window.note(pctr);
     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)
@@ -329,7 +367,11 @@ static void tx_thread(TxArgs args) {
     }
     tx_buf.insert(tx_buf.end(), dot11.begin(), dot11.end());
     tx_buf.insert(tx_buf.end(), psdu.begin(), psdu.end());
-    bool ok = args.rtl->send_packet(tx_buf.data(), tx_buf.size());
+    bool ok;
+    {
+      std::lock_guard<std::mutex> lk(g_send_mu);
+      ok = args.rtl->send_packet(tx_buf.data(), tx_buf.size());
+    }
     ++tx_count;
     if (tx_count <= 5 || tx_count % 500 == 0) {
       devourer::Ev(*g_ev, "stream.tx")
@@ -467,6 +509,59 @@ int main(int argc, char **argv) {
   TxArgs txa{rtlDevice, interval_ms, max_psdu, &should_stop, logger};
   std::thread tx{tx_thread, std::move(txa)};
 
+  /* Receipt emitter (DEVOURER_RX_RECEIPT_MS): every tick, encode the current
+   * window and inject it as an 802.11 data frame at a fixed robust 6M —
+   * receipts are control-plane, not part of the adaptive-rate stream. The
+   * first ticks fire during bring-up and fail harmlessly (send rc=false);
+   * the cadence, not any one frame, is the contract. */
+  std::thread receipt;
+  if (g_receipt_ms > 0 && g_seq_sa_set && g_receipt_sa_ok) {
+    receipt = std::thread([rtlDevice, &should_stop]() {
+      const auto rt =
+          devourer::build_stream_radiotap(devourer::parse_tx_mode_str("6M"));
+      std::vector<uint8_t> frame;
+      std::vector<uint8_t> tlv(
+          devourer::cell::receipt_tlv_size(g_receipt_window_bits));
+      long emitted = 0;
+      while (!should_stop.load()) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(g_receipt_ms));
+        const size_t n =
+            g_receipt_window.encode(g_seq_sa, tlv.data(), tlv.size());
+        if (n == 0)
+          continue; /* nothing received yet */
+        frame.clear();
+        frame.insert(frame.end(), rt.begin(), rt.end());
+        /* Plain (non-QoS) data header: RA = the receipted transmitter,
+         * TA/BSSID = the receipt identity. Body at offset 24 = the TLV. */
+        const uint8_t hdr[24] = {
+            0x08, 0x00, 0x00, 0x00,
+            g_seq_sa[0], g_seq_sa[1], g_seq_sa[2],
+            g_seq_sa[3], g_seq_sa[4], g_seq_sa[5],
+            g_receipt_sa[0], g_receipt_sa[1], g_receipt_sa[2],
+            g_receipt_sa[3], g_receipt_sa[4], g_receipt_sa[5],
+            g_receipt_sa[0], g_receipt_sa[1], g_receipt_sa[2],
+            g_receipt_sa[3], g_receipt_sa[4], g_receipt_sa[5],
+            0x00, 0x00};
+        frame.insert(frame.end(), hdr, hdr + sizeof hdr);
+        frame.insert(frame.end(), tlv.data(), tlv.data() + n);
+        bool ok;
+        {
+          std::lock_guard<std::mutex> lk(g_send_mu);
+          ok = rtlDevice->send_packet(frame.data(), frame.size());
+        }
+        ++emitted;
+        if (emitted <= 3 || emitted % 50 == 0) {
+          devourer::Ev(*g_ev, "receipt.tx")
+              .t()
+              .f("n", emitted)
+              .f("ok", ok ? 1 : 0)
+              .f("tlv_len", n)
+              .f("late", (unsigned long long)g_receipt_window.late());
+        }
+      }
+    });
+  }
+
   logger->info("duplex entering RX loop on ch {} — TX thread ready",
                channel);
   // RX loop. Same Init() path as rxdemo; SelectedChannel sets up the
@@ -480,6 +575,7 @@ int main(int argc, char **argv) {
   // — none wired here, so Ctrl-C ends the process abruptly and the OS reaps
   // the TX thread).
   should_stop = true;
+  if (receipt.joinable()) receipt.join();
   if (tx.joinable()) tx.join();
   /* Device, then interface, handle and context (DeviceSession.h). Explicit
    * only because the process has nothing left to do here — the destructor

examples/tx/main.cpp

@@ -36,6 +36,7 @@
 #include "BfReportDetect.h"
 #include "ChannelFreq.h"
 #include "DeviceSession.h"
+#include "cell/RxReceipt.h"
 #include "HopSchedule.h"
 #include "hopset/HopsetAuthority.h"
 #include "hopset/HopsetEvents.h"
@@ -303,12 +304,44 @@ static bool hopset_sense_window(IRtlDevice *dev, uint32_t settle_us,
 }
 
 static int g_rx_count = 0;
+/* Windowed RX receipts (src/cell/RxReceipt.h): DEVOURER_TX_RECEIPTS=1 arms
+ * the transmitter-side ledger. Receipts name our TA — DEVOURER_TX_SA when
+ * set, else the canonical SA — and anything else refuses to absorb. */
+static const bool g_receipts_on =
+    std::getenv("DEVOURER_TX_RECEIPTS") != nullptr;
+static devourer::cell::ReceiptLedger g_receipt_ledger;
+static uint8_t g_receipt_ta[6] = {0x57, 0x42, 0x75, 0x05, 0xd6, 0x00};
+
 static void packetProcessor(const Packet &packet) {
   /* C2H packets are chip-side status (one per TX during concurrent TX+RX on
    * Jaguar3), not 802.11 frames — skip before counting/parsing. */
   if (packet.RxAtrib.pkt_rpt_type == RX_PACKET_TYPE::C2H_PACKET)
     return;
   ++g_rx_count;
+  /* Windowed RX receipts (DEVOURER_TX_RECEIPTS=1, needs TX_WITH_RX=thread):
+   * a plain data frame whose body parses as a receipt TLV naming OUR TA is
+   * the receiver's delivered-set update (src/cell/RxReceipt.h). Every
+   * absorbed receipt is emitted WITH its TLV hex — the verifier replays the
+   * merge against the receiver's own rx.seq ledger, so decimating here would
+   * break the frame-exact comparison. */
+  if (g_receipts_on &&
+      packet.Data.size() >= 24 + devourer::cell::kReceiptHeader &&
+      packet.Data[0] == 0x08) {
+    const uint8_t *tlv = packet.Data.data() + 24;
+    const size_t tlv_len = packet.Data.size() - 24;
+    const long fresh = g_receipt_ledger.absorb(tlv, tlv_len, g_receipt_ta);
+    if (fresh >= 0) {
+      devourer::Ev(*g_ev, "tx.receipt")
+          .t()
+          .f("fresh", fresh)
+          .f("total", (unsigned long long)g_receipt_ledger.delivered_total())
+          .f("covered",
+             (unsigned long long)g_receipt_ledger.highest_covered())
+          .f("receipts", (unsigned long long)g_receipt_ledger.receipts())
+          .hex("tlv", tlv, tlv_len);
+      return;
+    }
+  }
   /* RX liveness marker for the TX+RX=thread mode: first frame + every 500th.
    * Without it a deaf RX loop is indistinguishable from a quiet channel. */
   if (g_rx_count == 1 || g_rx_count % 500 == 0) {
@@ -1093,6 +1126,7 @@ int main(int argc, char **argv) {
       else
         logger->warn("DEVOURER_TX_SA unparseable — keeping the canonical SA");
     }
+    std::memcpy(g_receipt_ta, kQosSa, 6); /* receipts name this TA */
     uint8_t ra[6] = {0xff, 0xff, 0xff, 0xff, 0xff, 0xff};
     if (const char *e = std::getenv("DEVOURER_TX_RA")) {
       if (auto m = devourer::parse_mac(e))