diff --git a/CMakeLists.txt b/CMakeLists.txt index f8ba4abe..79ebe7e1 100644 --- a/CMakeLists.txt +++ b/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. diff --git a/docs/logging.md b/docs/logging.md index d505f8f1..7db607a1 100644 --- a/docs/logging.md +++ b/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 | diff --git a/docs/scheduled-mac.md b/docs/scheduled-mac.md index d07f5db2..96c07a0e 100644 --- a/docs/scheduled-mac.md +++ b/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 diff --git a/examples/duplex/main.cpp b/examples/duplex/main.cpp index 6f056cc0..7c005ff3 100644 --- a/examples/duplex/main.cpp +++ b/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(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 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 frame; + std::vector 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 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 diff --git a/examples/tx/main.cpp b/examples/tx/main.cpp index aee3f05e..f5f19d0b 100644 --- a/examples/tx/main.cpp +++ b/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)) diff --git a/src/cell/RxReceipt.h b/src/cell/RxReceipt.h new file mode 100644 index 00000000..944b96f0 --- /dev/null +++ b/src/cell/RxReceipt.h @@ -0,0 +1,270 @@ +/* Windowed RX receipts — app-layer delivery truth for a TX peer. + * + * The hardware ACK's horizon is chip-FIFO admission (SIFS timing: it can + * never confirm host delivery), and per-frame CCX reports are rate-bounded + * by the fw's ~1.3k reports/s emission ceiling. This is the tier above both: + * the RECEIVER's application counts what it actually consumed and mails a + * compact bitmap back on its existing feedback path; the transmitter merges + * those receipts into a delivered-set that is true at the only layer that + * matters. Same caller-side contract as UeRxAttribution: the demos feed + * frames in from the Packet callback, the device RX loops are untouched, and + * the library never assumes a payload convention — the caller supplies the + * u32 frame index (the bench wires txdemo's QoS-Data payload counter). + * + * Wire format (versioned TLV, little-endian, one transmitter per TLV): + * [0..2) magic "DR" + * [2] version (1) + * [3] flags (0) + * [4..10) ta — the transmitter this receipt covers; the TX side absorbs + * only receipts naming its own SA, which (with magic + version + + * exact-length) keeps arbitrary payloads from parsing as receipts + * [10..14) base — first frame index the bitmap covers + * [14..16) nbits — window length in frames + * [16..) bitmap, LSB-first: bit i set => frame base+i was received + * + * Receipts OVERLAP: every emission covers the last `window_bits` indices, so + * each frame appears in many consecutive receipts and individual receipt + * loss is harmless — the merge is idempotent. A frame index arriving BELOW + * the current window (reordered by more than window_bits) is counted as + * `late` and never receipted; size the window against the emission cadence + * (2048 bits at 2.4k fps ≈ 0.85 s of slack, orders beyond USB reordering). + */ +#ifndef DEVOURER_CELL_RX_RECEIPT_H +#define DEVOURER_CELL_RX_RECEIPT_H + +#include +#include +#include +#include +#include + +namespace devourer { +namespace cell { + +constexpr size_t kReceiptHeader = 16; +constexpr uint8_t kReceiptVersion = 1; + +constexpr size_t receipt_tlv_size(uint16_t nbits) { + return kReceiptHeader + (static_cast(nbits) + 7) / 8; +} + +/* Receiver half: a ring bitmap over the last `window_bits` frame indices. + * + * SIZING: the window must exceed the largest index jump the receiver's own + * buffering can produce between two encodes, or a post-stall backlog drain + * evicts frames before any receipt covers them (measured: a 150 ms consumer + * stall behind a 104-buffer spsc-fat pool drained ~3k frames in one receipt + * interval and a 2048-bit window leaked 2,846 delivered frames out of + * coverage). Rule: window_bits > pool_bytes / min_frame_bytes + one encode + * interval of arrivals. The 8192 default gives ~2.7x margin over that bench + * worst case at ~1 KB of bitmap per receipt (~1.4 ms at 6M). */ +class ReceiptWindow { +public: + explicit ReceiptWindow(uint16_t window_bits = 8192) + : _w(window_bits ? window_bits : 8192), + _bits((_w + 63) / 64, 0) {} + + /* Record a received frame index. Indices may repeat (duplicates are + * idempotent) and skip forward arbitrarily; an index below the current + * window is counted, not receipted. */ + void note(uint32_t idx) { + std::lock_guard lk(_mu); + if (!_any) { + _any = true; + _max = idx; + set_bit(idx); + return; + } + if (idx > _max) { + /* Advance: every ring slot between the old and new max is a NEW index + * whose bit must start clear. A jump >= _w wipes the whole ring. */ + const uint32_t advance = idx - _max; + const uint32_t to_clear = advance < _w ? advance : _w; + for (uint32_t k = 0; k < to_clear; ++k) + clear_bit(_max + 1 + k + (advance - to_clear)); + _max = idx; + set_bit(idx); + return; + } + if (_max - idx >= _w) { + ++_late; /* below the window — reordered past the receipt horizon */ + return; + } + set_bit(idx); + } + + /* Encode the current window into `out` (>= receipt_tlv_size(window_bits)). + * Returns the TLV length, or 0 if nothing has been noted yet. */ + size_t encode(const uint8_t ta[6], uint8_t *out, size_t out_len) const { + std::lock_guard lk(_mu); + if (!_any) + return 0; + const uint32_t base = _max >= _w - 1 ? _max - (_w - 1) : 0; + const uint16_t nbits = static_cast(_max - base + 1); + const size_t need = receipt_tlv_size(nbits); + if (out_len < need) + return 0; + out[0] = 'D'; + out[1] = 'R'; + out[2] = kReceiptVersion; + out[3] = 0; + std::memcpy(out + 4, ta, 6); + put32(out + 10, base); + put16(out + 14, nbits); + std::memset(out + kReceiptHeader, 0, need - kReceiptHeader); + for (uint32_t i = 0; i < nbits; ++i) + if (get_bit(base + i)) + out[kReceiptHeader + i / 8] |= static_cast(1u << (i % 8)); + return need; + } + + uint64_t late() const { + std::lock_guard lk(_mu); + return _late; + } + +private: + /* Ring addressing: index -> bit slot idx % _w, and BOTH the word and the + * bit offset derive from the slot — deriving the bit from idx % 64 is only + * equivalent when _w is a multiple of 64, and the window size is caller- + * chosen. Valid because the window invariant keeps every live index within + * [_max - _w + 1, _max]. */ + void set_bit(uint32_t idx) { + const uint32_t s = idx % _w; + _bits[s / 64] |= 1ull << (s % 64); + } + void clear_bit(uint32_t idx) { + const uint32_t s = idx % _w; + _bits[s / 64] &= ~(1ull << (s % 64)); + } + bool get_bit(uint32_t idx) const { + const uint32_t s = idx % _w; + return (_bits[s / 64] >> (s % 64)) & 1u; + } + static void put32(uint8_t *p, uint32_t v) { + p[0] = v & 0xff; + p[1] = (v >> 8) & 0xff; + p[2] = (v >> 16) & 0xff; + p[3] = (v >> 24) & 0xff; + } + static void put16(uint8_t *p, uint16_t v) { + p[0] = v & 0xff; + p[1] = (v >> 8) & 0xff; + } + + mutable std::mutex _mu; + uint32_t _w; + std::vector _bits; + bool _any = false; + uint32_t _max = 0; + uint64_t _late = 0; +}; + +/* One decoded receipt, viewing the caller's buffer (no copy). */ +struct ReceiptView { + const uint8_t *ta = nullptr; + uint32_t base = 0; + uint16_t nbits = 0; + const uint8_t *bitmap = nullptr; + bool bit(uint32_t i) const { return (bitmap[i / 8] >> (i % 8)) & 1u; } +}; + +/* Strict-prefix parse: magic + version + a bitmap fully inside the buffer. + * Trailing bytes are accepted — a received MPDU body carries the 4-byte FCS + * after the TLV (parsers deliver it; an ACK frame logs len 14 = 10 + FCS), + * and the absorb side additionally requires the 6-byte TA match, which keeps + * arbitrary payloads from parsing as receipts. */ +inline bool receipt_decode(const uint8_t *p, size_t len, ReceiptView &out) { + if (!p || len < kReceiptHeader || p[0] != 'D' || p[1] != 'R' || + p[2] != kReceiptVersion) + return false; + const uint16_t nbits = + static_cast(p[14] | (static_cast(p[15]) << 8)); + if (nbits == 0 || len < receipt_tlv_size(nbits)) + return false; + out.ta = p + 4; + out.base = static_cast(p[10]) | (static_cast(p[11]) << 8) | + (static_cast(p[12]) << 16) | + (static_cast(p[13]) << 24); + out.nbits = nbits; + out.bitmap = p + kReceiptHeader; + return true; +} + +/* Transmitter half: merge receipts into the delivered-set. + * + * `max_index` bounds the bitmap: receipts arrive OVER THE AIR, so `base` is + * attacker-influencable and an unbounded resize would let one crafted TLV + * (base near 2^32) allocate half a gigabyte. The default cap (2^26 frames = + * 8 MB of bitmap, ~7.7 h at 2.4 k fps) is far above any bench run; a peer + * whose real index space outgrows it should shard by session. */ +class ReceiptLedger { +public: + explicit ReceiptLedger(uint64_t max_index = 1ull << 26) + : _max_index(max_index) {} + + /* Absorb a receipt covering our TA. Returns the number of NEWLY-learned + * delivered indices (0 for a pure re-receipt), or -1 if the TLV does not + * parse, names a different transmitter, or claims indices beyond the + * ledger's cap. */ + long absorb(const uint8_t *tlv, size_t len, const uint8_t own_ta[6]) { + ReceiptView v; + if (!receipt_decode(tlv, len, v) || std::memcmp(v.ta, own_ta, 6) != 0) + return -1; + std::lock_guard lk(_mu); + const uint64_t top = static_cast(v.base) + v.nbits; + if (top > _max_index) + return -1; + if (top > _bits.size() * 64) + _bits.resize((top + 63) / 64, 0); + long fresh = 0; + for (uint32_t i = 0; i < v.nbits; ++i) { + if (!v.bit(i)) + continue; + const uint64_t idx = static_cast(v.base) + i; + uint64_t &word = _bits[idx / 64]; + const uint64_t m = 1ull << (idx % 64); + if (!(word & m)) { + word |= m; + ++fresh; + ++_total; + } + } + if (top > _highest_covered) + _highest_covered = top; /* exclusive: indices below this were covered */ + ++_receipts; + return fresh; + } + + bool delivered(uint64_t idx) const { + std::lock_guard lk(_mu); + return idx / 64 < _bits.size() && ((_bits[idx / 64] >> (idx % 64)) & 1u); + } + uint64_t delivered_total() const { + std::lock_guard lk(_mu); + return _total; + } + /* Exclusive upper bound of receipt coverage: every index below it has been + * inside at least one absorbed receipt's window. */ + uint64_t highest_covered() const { + std::lock_guard lk(_mu); + return _highest_covered; + } + uint64_t receipts() const { + std::lock_guard lk(_mu); + return _receipts; + } + +private: + mutable std::mutex _mu; + uint64_t _max_index; + std::vector _bits; + uint64_t _total = 0; + uint64_t _highest_covered = 0; + uint64_t _receipts = 0; +}; + +} /* namespace cell */ +} /* namespace devourer */ + +#endif /* DEVOURER_CELL_RX_RECEIPT_H */ diff --git a/tests/arq_e2e_delivery.sh b/tests/arq_e2e_delivery.sh index db80ee3c..080dd86b 100644 --- a/tests/arq_e2e_delivery.sh +++ b/tests/arq_e2e_delivery.sh @@ -44,6 +44,7 @@ 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_REPORT_N=${DRONE_REPORT_N:-1} # CCX sampling divisor (1 = every frame) +RECEIPT_MS=${RECEIPT_MS:-} # windowed RX receipts cadence (off="") 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 @@ -139,6 +140,7 @@ 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 \ + ${RECEIPT_MS:+DEVOURER_RX_RECEIPT_MS="$RECEIPT_MS"} \ ${DUT_RX_MODE:+DEVOURER_RX_MODE="$DUT_RX_MODE"} \ ${DUT_RX_MODE:+DEVOURER_RX_POOL_SPARE="$DUT_POOL_SPARE"} \ ${DUT_POOL_EXHAUST:+DEVOURER_RX_POOL_EXHAUST="$DUT_POOL_EXHAUST"} \ @@ -165,6 +167,7 @@ env DEVOURER_VID="$DRONE_VID" DEVOURER_PID="$DRONE_PID" DEVOURER_CHANNEL="$CH" \ DEVOURER_TX_RATE="$DRONE_RATE" DEVOURER_TX_PAYLOAD_BYTES="$DRONE_PAYLOAD" \ DEVOURER_TX_GAP_US="$DRONE_GAP_US" \ DEVOURER_TX_REPORT="$DRONE_REPORT_N" DEVOURER_TX_RETRY_LIMIT="$RETRY_LIMIT" \ + ${RECEIPT_MS:+DEVOURER_TX_RECEIPTS=1 DEVOURER_TX_WITH_RX=thread} \ DEVOURER_DIS_CCA="$DRONE_DIS_CCA" \ DEVOURER_TX_PWR_OFFSET_QDB="$PWR_QDB" \ DEVOURER_LOG_LEVEL=warn DEVOURER_EVENTS=stdout \ diff --git a/tests/receipt_selftest.cpp b/tests/receipt_selftest.cpp new file mode 100644 index 00000000..388cb58c --- /dev/null +++ b/tests/receipt_selftest.cpp @@ -0,0 +1,131 @@ +/* Headless guard for the windowed RX-receipt primitives + * (src/cell/RxReceipt.h): TLV encode/decode round-trip, ring-window eviction + * across large index jumps, below-window late accounting, ledger merge + * idempotence over overlapping receipts, and the strict-parse rejections + * (magic/version/length/TA) that keep arbitrary payloads from absorbing as + * receipts. Pure math — the on-air halves (injection cadence, feedback-path + * loss tolerance, frame-exact agreement with the rx.seq ledger) are + * tests/receipt_verify.py over an arq_e2e run. */ +#include +#include +#include +#include +#include +#include + +#include "cell/RxReceipt.h" + +using devourer::cell::ReceiptLedger; +using devourer::cell::ReceiptView; +using devourer::cell::ReceiptWindow; +using devourer::cell::receipt_decode; +using devourer::cell::receipt_tlv_size; + +static const uint8_t kTa[6] = {0x02, 0xaa, 0xbb, 0xcc, 0xdd, 0x01}; +static const uint8_t kOther[6] = {0x02, 0xaa, 0xbb, 0xcc, 0xdd, 0x02}; + +static std::set decode_set(const uint8_t *tlv, size_t len) { + ReceiptView v; + assert(receipt_decode(tlv, len, v)); + std::set s; + for (uint32_t i = 0; i < v.nbits; ++i) + if (v.bit(i)) + s.insert(v.base + i); + return s; +} + +int main() { + /* Round-trip: a sparse pattern survives encode/decode exactly. */ + { + ReceiptWindow w(256); + std::set fed; + for (uint32_t i : {0u, 1u, 5u, 60u, 63u, 64u, 100u, 255u}) { + w.note(i); + fed.insert(i); + } + uint8_t buf[receipt_tlv_size(256)]; + const size_t n = w.encode(kTa, buf, sizeof buf); + assert(n == receipt_tlv_size(256)); + assert(decode_set(buf, n) == fed); + } + + /* Eviction: indices that fell out of the ring are gone; slots reused by a + * big jump start clear. */ + { + ReceiptWindow w(64); + for (uint32_t i = 0; i < 64; ++i) + w.note(i); /* full window, all set */ + w.note(200); /* jump: window becomes [137..200], only 200 set */ + uint8_t buf[receipt_tlv_size(64)]; + const size_t n = w.encode(kTa, buf, sizeof buf); + assert(n == receipt_tlv_size(64)); + const auto s = decode_set(buf, n); + assert(s.size() == 1 && *s.begin() == 200); + /* Below-window arrival counts as late, never receipts. */ + assert(w.late() == 0); + w.note(3); + assert(w.late() == 1); + } + + /* Ledger: overlapping receipts merge idempotently; fresh counts only new + * indices; coverage bound advances; foreign TA and malformed TLVs refuse. */ + { + ReceiptWindow w(128); + ReceiptLedger led; + uint8_t buf[receipt_tlv_size(128)]; + for (uint32_t i = 0; i < 100; ++i) + if (i != 40 && i != 41 && i != 90) /* three losses */ + w.note(i); + size_t n = w.encode(kTa, buf, sizeof buf); + assert(led.absorb(buf, n, kTa) == 97); + assert(led.absorb(buf, n, kTa) == 0); /* pure re-receipt */ + /* Late recovery inside the window, then an overlapping re-encode. */ + w.note(90); + n = w.encode(kTa, buf, sizeof buf); + assert(led.absorb(buf, n, kTa) == 1); + assert(led.delivered_total() == 98); + assert(led.delivered(89) && led.delivered(90) && !led.delivered(40)); + assert(led.highest_covered() == 100); /* exclusive */ + assert(led.receipts() == 3); + /* Rejections: wrong TA, truncated, magic, version. Trailing bytes are + * ACCEPTED (a received body carries the FCS after the TLV). */ + assert(led.absorb(buf, n, kOther) == -1); + assert(led.absorb(buf, n - 1, kTa) == -1); + uint8_t fcs[receipt_tlv_size(128) + 4]; + std::memcpy(fcs, buf, n); + std::memset(fcs + n, 0xee, 4); + assert(led.absorb(fcs, n + 4, kTa) == 0); /* parses; nothing new */ + uint8_t bad[receipt_tlv_size(128)]; + std::memcpy(bad, buf, n); + bad[0] = 'X'; + assert(led.absorb(bad, n, kTa) == -1); + std::memcpy(bad, buf, n); + bad[2] = 99; + assert(led.absorb(bad, n, kTa) == -1); + /* Over-the-air DoS guard: a crafted base near 2^32 must refuse, not + * allocate half a gigabyte of bitmap. */ + std::memcpy(bad, buf, n); + bad[10] = bad[11] = bad[12] = bad[13] = 0xff; + assert(led.absorb(bad, n, kTa) == -1); + assert(led.delivered_total() == 98); /* untouched by the rejects */ + } + + /* Ring addressing with a window that is NOT a multiple of 64 — the word + * and bit offsets must both derive from the slot, or bits land in the + * wrong positions once idx % _w and idx % 64 diverge. */ + { + ReceiptWindow w(100); + std::set fed; + for (uint32_t i = 150; i < 250; i += 7) { /* crosses the ring seam */ + w.note(i); + fed.insert(i); + } + uint8_t buf[receipt_tlv_size(100)]; + const size_t n = w.encode(kTa, buf, sizeof buf); + assert(n != 0); + assert(decode_set(buf, n) == fed); + } + + std::puts("receipt selftest: OK"); + return 0; +} diff --git a/tests/receipt_verify.py b/tests/receipt_verify.py new file mode 100644 index 00000000..12fa5b91 --- /dev/null +++ b/tests/receipt_verify.py @@ -0,0 +1,94 @@ +#!/usr/bin/env python3 +"""Frame-exact verification of the windowed RX receipts (issue: delivery +truth at the app layer). + +Replays every receipt TLV the transmitter absorbed (tx.receipt events carry +the raw hex) through the same merge the ReceiptLedger performs, and compares +the resulting delivered-set against the RECEIVER's own ground truth — the +rx.seq ledger the arq_e2e bench already trusts frame-exactly. The comparison +domain is [0, covered): `covered` is the exclusive top of receipt coverage, +so frames the receiver saw after the last receipt flew are out of contract. + +PASS means the transmitter's receipt-derived view of "what the app received" +is identical, frame for frame, to what the app actually received. + + python3 tests/receipt_verify.py --dut /dut.jsonl --drone /drone.jsonl +""" +import argparse +import json +import sys + +HDR = 16 + + +def replay(drone_path): + got = set() + covered = 0 + receipts = 0 + for line in open(drone_path, errors="replace"): + if not line.startswith('{"ev":"tx.receipt"'): + continue + try: + ev = json.loads(line) + except Exception: + continue + tlv = bytes.fromhex(ev.get("tlv", "")) + if len(tlv) < HDR or tlv[0:2] != b"DR" or tlv[2] != 1: + print(f"BAD TLV in event stream: {ev.get('tlv', '')[:32]}...") + return None, 0, 0 + base = int.from_bytes(tlv[10:14], "little") + nbits = int.from_bytes(tlv[14:16], "little") + # Prefix-strict like receipt_decode: the captured body carries the + # 4-byte FCS after the TLV. + if len(tlv) < HDR + (nbits + 7) // 8: + print("BAD TLV length") + return None, 0, 0 + for i in range(nbits): + if tlv[HDR + i // 8] >> (i % 8) & 1: + got.add(base + i) + covered = max(covered, base + nbits) + receipts += 1 + return got, covered, receipts + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--dut", required=True) + ap.add_argument("--drone", required=True) + a = ap.parse_args() + + receipt_set, covered, receipts = replay(a.drone) + if receipt_set is None or receipts == 0: + print("VERDICT: NO-RECEIPTS — nothing to verify " + "(DEVOURER_TX_RECEIPTS + DEVOURER_RX_RECEIPT_MS both set?)") + return 3 + + truth = set() + for line in open(a.dut, errors="replace"): + if line.startswith('{"ev":"rx.seq"'): + try: + truth.add(json.loads(line)["pctr"]) + except Exception: + pass + truth_in_domain = {k for k in truth if k < covered} + + missing = truth_in_domain - receipt_set # app got it, receipts never said + phantom = receipt_set - truth_in_domain # receipts claim it, app didn't + print(f"receipts={receipts} covered=[0,{covered}) " + f"receipt_set={len(receipt_set)} truth_in_domain=" + f"{len(truth_in_domain)}") + if missing: + print(f"MISSING from receipts ({len(missing)}): " + f"{sorted(missing)[:10]}{'...' if len(missing) > 10 else ''}") + if phantom: + print(f"PHANTOM in receipts ({len(phantom)}): " + f"{sorted(phantom)[:10]}{'...' if len(phantom) > 10 else ''}") + ok = not missing and not phantom + print(f"VERDICT: {'FRAME-EXACT' if ok else 'MISMATCH'} — the " + f"transmitter's receipt-derived delivered-set " + f"{'==' if ok else '!='} the receiver's rx.seq ledger") + return 0 if ok else 1 + + +if __name__ == "__main__": + sys.exit(main())