stream timing: five more per-frame bytes in addr1, a group DA every receiver delivers

docs/logging.md

@@ -88,7 +88,7 @@ Emitters: L = library, RX/TX/... = demo. Optional fields in [brackets];
 | ev | emitter | fields |
 |---|---|---|
 | `rx.pkt` | RX | n, len (first 10 + every 100th frame) |
-| `rx.frame` | RX (`DEVOURER_STREAM_OUT`), duplex | rate, len, crc, icv, rssi[2], evm[2], snr[2], seq, tsfl, bw, stbc, ldpc, sgi, paggr, ppdu, fc1 (FC flags byte; bit3 = 802.11 RETRY), sa hex, fc0 (frame-control type/subtype byte: 0x40 stream data, 0x50 timing marker, 0x80 beacon), a3 hex (addr3 verbatim, frames ≥ 24 bytes), body hex; `tx_tsf` (sender's hardware egress TSF) on beacons/probe-responses only. When addr3 decodes as the stream timing field (`src/StreamTelemetry.h`, RX only): tel=1, depth (frames behind this one on the TX host), c2s_us (capture→send_packet; cap=1 when from the producer's CAPTURE_TS stamp, else from the stdin read), and once the receiver has fitted the transmitter's clock from its egress-stamped frames: lat_us (predicted submit→hardware arrival, one-way) and c2a_us (capture→arrival, cap=1 frames). SA gate: canonical-SA with `DEVOURER_RX_AGG_SA` unset; when set, that filter ("canon"/mac/"any") selects the stream instead |
+| `rx.frame` | RX (`DEVOURER_STREAM_OUT`), duplex | rate, len, crc, icv, rssi[2], evm[2], snr[2], seq, tsfl, bw, stbc, ldpc, sgi, paggr, ppdu, fc1 (FC flags byte; bit3 = 802.11 RETRY), sa hex, fc0 (frame-control type/subtype byte: 0x40 stream data, 0x50 timing marker, 0x80 beacon), a3 hex (addr3 verbatim, frames ≥ 24 bytes), body hex; `tx_tsf` (sender's hardware egress TSF) on beacons/probe-responses only. When addr3 decodes as the stream timing field (`src/StreamTelemetry.h`, RX only): tel=1, depth (frames behind this one on the TX host), c2s_us (capture→send_packet; cap=1 when from the producer's CAPTURE_TS stamp, else from the stdin read), and once the receiver has fitted the transmitter's clock from its egress-stamped frames: lat_us (predicted submit→hardware arrival, one-way) and c2a_us (capture→arrival, cap=1 frames); when addr1 carries the extension (a `07:…` group DA: `0x03 | version<<2`): tq_us (stdin read→send for this frame), tw_us (the previous frame's send_packet wall time), ctr (TX frame counter, low byte). SA gate: canonical-SA with `DEVOURER_RX_AGG_SA` unset; when set, that filter ("canon"/mac/"any") selects the stream instead |
 | `rx.timing` | RX | the transmitter's stream timing marker (probe response, canonical SA) decoded: frames, tq_p50_us, tq_max_us (stdin read→send), tw_p50_us, tw_max_us (send_packet wall time), c2s_p50_us, c2s_max_us, depth_max, captured, tsf_pred, tx_fit_ppm, tx_fit_n, presp_stamped (this part's injected probe responses carry a hardware egress TSF), beacon (a TBTT beacon carries the clock instead), tx_async; plus this receiver's clock fit: tsfl, pairs, rx_fit_ppm, rx_resid_us, rx_resets (a transmitter restart resets its TSF; the fit starts over past 50 ms) |
 | `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] |

docs/stream-timing.md

@@ -25,6 +25,37 @@ byte-for-byte untouched and the MTU is unchanged:
 The version is chosen so the old addr3 contents (the canonical SA) decode as
 "no telemetry" rather than as garbage values.
 
+**Five more bytes ride addr1**, the DA. It stays a group address — bit 0 of
+its first byte set — so nothing on the air ACKs it and every TX descriptor
+path still derives its broadcast/multicast bit from it; the first byte is
+`0x03 | version<<2` (group, locally administered), `07` for this version, so
+the broadcast address a demo writes when it carries no extension decodes as
+"none":
+
+| bytes | field |
+|---|---|
+| 0 | group + locally administered bits, version |
+| 1–2 | stdin read → `send_packet` for this frame, 10 µs units |
+| 3–4 | the previous frame's `send_packet` wall time (kestrel-air's T_WRITE), 10 µs units |
+| 5 | the transmitter's frame counter, low byte |
+
+Whether a monitor receiver delivers a non-broadcast group DA at all was
+measured, not assumed (`tests/mcast_da_rx_check.sh`: an injector alternates
+the broadcast DA and the exact bytes the extension's encoder ships, `07:…`,
+each receiver family counts both by body tag, 15 s per cell):
+
+| receiver | broadcast | group DA | ratio |
+|---|---|---|---|
+| 8822BU (Jaguar2) | 1364 | 1367 | 1.00 |
+| 8812CU (Jaguar3) | 1385 | 1385 | 1.00 |
+| 8821AU (Jaguar1) | 1353 | 1355 | 1.00 |
+| 8832CU (Kestrel) | 1416 | 1424 | 1.01 |
+| MT7612U | 1347 | 1346 | 1.00 |
+
+The RTL8733B was not plugged and is unmeasured. The marker frame, the hop
+sync marker and the Jaguar1 clock beacon keep the broadcast DA; only the
+stream's data frames carry the extension.
+
 **A periodic marker** (`DEVOURER_STREAM_TIMING=N`, every N data frames, its
 own frame like the hop sync marker) carries the absolute pair (predicted TSF,
 host clock), the state of the transmitter's host↔TSF fit, and the window since

examples/common/stream_timing_tx.h

@@ -36,6 +36,7 @@
 class StreamTimingTx {
  public:
   using FrameTiming = devourer::stream_timing::FrameTiming;
+  using FrameTimingExt = devourer::stream_timing::FrameTimingExt;
   using TimingMarker = devourer::stream_timing::TimingMarker;
 
   StreamTimingTx(IRadio &dev, devourer::EventSink &ev, Logger &log,
@@ -96,10 +97,19 @@ class StreamTimingTx {
 
   // Right before send_packet. `read_ns` = when the record came off stdin,
   // `capture_ns` = the producer's stamp (has_capture) or read_ns. Fills the six
-  // addr3 bytes; remembers the instants for sent().
-  void stamp(uint8_t *addr3, uint64_t read_ns, uint64_t capture_ns, bool has_capture) {
+  // addr3 bytes and, when `addr1` is given, the addr1 extension (the DA
+  // becomes the 03:… group address); remembers the instants for sent().
+  void stamp(uint8_t *addr3, uint64_t read_ns, uint64_t capture_ns, bool has_capture,
+             uint8_t *addr1 = nullptr) {
     ensure_started();
     _t0 = now_ns();
+    if (addr1) {
+      FrameTimingExt x;
+      x.t_queue10 = devourer::stream_timing::clip10(_t0 > read_ns ? (_t0 - read_ns) / 1000 : 0);
+      x.t_write_prev10 = devourer::stream_timing::clip10(_last_write_us);
+      x.ctr = static_cast<uint8_t>(_frames);
+      x.encode(addr1);
+    }
     _read_ns = read_ns;
     _capture_ns = capture_ns;
     _has_capture = has_capture;
@@ -129,6 +139,7 @@ class StreamTimingTx {
     const uint64_t tq = _t0 > _read_ns ? (_t0 - _read_ns) / 1000 : 0;
     const uint64_t tw = (t1 - _t0) / 1000;
     const uint64_t c2s = _t0 > _capture_ns ? (_t0 - _capture_ns) / 1000 : 0;
+    _last_write_us = tw;
     _window.add(tq, tw, c2s, _has_capture, _depth);
     ++_frames;
   }
@@ -224,7 +235,7 @@ class StreamTimingTx {
   devourer::stream_timing::TimingWindow _window;
   std::vector<uint8_t> _buf;
   uint64_t _frames = 0, _last_marker_frame = UINT64_MAX;
-  uint64_t _t0 = 0, _read_ns = 0, _capture_ns = 0;
+  uint64_t _t0 = 0, _read_ns = 0, _capture_ns = 0, _last_write_us = 0;
   bool _has_capture = false;
   uint32_t _depth = 0;
   bool _pending_capture = false, _capture_seen = false;

examples/duplex/main.cpp

@@ -403,7 +403,8 @@ static void tx_thread(TxArgs args) {
         std::lock_guard<std::mutex> lr(g_rt_mu);
         timing.maybe_marker(g_radiotap);
       }
-      timing.stamp(tx_buf.data() + addr3_off, read_ns, capture_ns, has_capture);
+      timing.stamp(tx_buf.data() + addr3_off, read_ns, capture_ns, has_capture,
+                   tx_buf.data() + addr3_off - 12 /* addr1 */);
       ok = args.rtl->send_packet(tx_buf.data(), tx_buf.size());
       timing.sent(ok);
     }

examples/rx/main.cpp

@@ -1393,6 +1393,13 @@ static void packetProcessor(const Packet &packet) {
               .f("depth", ft.depth)
               .f("c2s_us", ft.c2s10 * 10)
               .f("cap", ft.has_capture ? 1 : 0);
+          /* addr1 extension: per-frame stdin-read→send and the previous
+           * frame's send_packet wall time, plus the TX frame counter. */
+          devourer::stream_timing::FrameTimingExt fx;
+          if (devourer::stream_timing::FrameTimingExt::decode(packet.Data.data() + 4, fx))
+            ev.f("tq_us", fx.t_queue10 * 10)
+                .f("tw_us", fx.t_write_prev10 * 10)
+                .f("ctr", fx.ctr);
           devourer::stream_timing::FrameLatency fl;
           if (g_rt_hw_tsfl && sa_canon && !corrupted && g_rt_fit.ready() &&
               devourer::stream_timing::latency(

examples/streamtx/main.cpp

@@ -497,7 +497,8 @@ int main(int argc, char **argv) {
     tx_buf.insert(tx_buf.end(), dot11.begin(), dot11.end());
     tx_buf.insert(tx_buf.end(), psdu.begin(), psdu.end());
     /* addr3 is the per-frame telemetry field (the body is untouched). */
-    timing.stamp(tx_buf.data() + addr3_off, read_ns, capture_ns, has_capture);
+    timing.stamp(tx_buf.data() + addr3_off, read_ns, capture_ns, has_capture,
+                 tx_buf.data() + addr3_off - 12 /* addr1 */);
     bool ok = rtlDevice->send_packet(tx_buf.data(), tx_buf.size());
     timing.sent(ok);
     ++tx_count;

examples/svctx/main.cpp

@@ -261,7 +261,8 @@ int main(int argc, char** argv) {
       frame.insert(frame.end(), nal.begin() + off, nal.begin() + off + n);
       timing.maybe_marker(rt);
       timing.stamp(frame.data() + rt.size() + 16, read_ns,
-                   first ? capture_ns : read_ns, first && has_capture);
+                   first ? capture_ns : read_ns, first && has_capture,
+                   frame.data() + rt.size() + 4 /* addr1 */);
       timing.sent(rtlDevice->send_packet(frame.data(), frame.size()));
       first = false;
       if (gap_us > 0)

src/StreamTelemetry.h

@@ -4,7 +4,10 @@
  * Two carriers, both pure wire codecs here (no device dependency):
  *
  *  1. FrameTiming — six bytes in the 802.11 header's addr3 (BSSID) of every
- *     stream frame. The stream demos fill addr3 with the canonical SA today and
+ *     stream frame, and FrameTimingExt — five more in addr1 (the DA), which
+ *     stays a group address so nothing ACKs it (byte 0 = 0x03 | version<<2;
+ *     broadcast decodes as "none"). Which receivers deliver such a DA, and
+ *     the numbers, are in docs/stream-timing.md. The stream demos fill addr3 with the canonical SA today and
  *     no receiver reads it (every consumer keys on addr2 and slices the body at
  *     +24), so the FEC bodies stay byte-for-byte untouched and the MTU is
  *     unchanged. It carries what kestrel-air puts in its slice header: the
@@ -113,6 +116,38 @@ struct FrameTiming {
   }
 };
 
+/* ---- per-frame extension: addr1 ------------------------------------------ */
+
+/* addr1 byte 0: bit0 group (must stay set — a unicast DA would solicit an
+ * ACK), bit1 locally administered, bits 7..2 the version. Broadcast ff reads
+ * as version 63 and is rejected. */
+inline constexpr uint8_t kExtVersion = 1;
+
+struct FrameTimingExt {
+  uint16_t t_queue10 = 0;      /* stdin read → send_packet, 10 µs, clipped */
+  uint16_t t_write_prev10 = 0; /* the PREVIOUS frame's send_packet wall time, 10 µs
+                                * (kestrel-air's T_WRITE); 0 on the first frame */
+  uint8_t ctr = 0;             /* the transmitter's frame counter, low byte */
+
+  static constexpr size_t kSize = 6;
+
+  void encode(uint8_t out[kSize]) const {
+    out[0] = static_cast<uint8_t>(0x03 | (kExtVersion << 2));
+    out[1] = static_cast<uint8_t>(t_queue10);
+    out[2] = static_cast<uint8_t>(t_queue10 >> 8);
+    out[3] = static_cast<uint8_t>(t_write_prev10);
+    out[4] = static_cast<uint8_t>(t_write_prev10 >> 8);
+    out[5] = ctr;
+  }
+  static bool decode(const uint8_t in[kSize], FrameTimingExt &e) {
+    if ((in[0] & 0x03) != 0x03 || (in[0] >> 2) != kExtVersion) return false;
+    e.t_queue10 = static_cast<uint16_t>(in[1] | (in[2] << 8));
+    e.t_write_prev10 = static_cast<uint16_t>(in[3] | (in[4] << 8));
+    e.ctr = in[5];
+    return true;
+  }
+};
+
 /* Per-frame latency on the receiver. `remote_arrival_us` is this frame's
  * hardware arrival (tsfl) mapped onto the transmitter's TSF by the receiver's
  * fit. Returns false when the frame carries no TSF. On success `submit_to_air`

tests/README.md

@@ -600,7 +600,10 @@ every 10th record; the receiver must see that step on ~10% of frames), `hop`
 (slot hopping; the clock survives retunes where the part stamps injected
 frames), `corrupt` (CRC-failed frames kept and never fed to the fit), `svctx`
 (replay), `svctx-live` (`svctx --live` fed by `gen_svc_nals.py --capture-ts
---pace-us`, the same producer-delay check per NAL), `duplex`. `tests/stream_timing_analyze.py` is the per-capture summary and the
+--pace-us`, the same producer-delay check per NAL), `duplex`. `tests/mcast_da_rx_check.sh` is the gate behind the
+addr1 extension: an injector alternates the broadcast DA and the group DA the
+extension's encoder ships (`07:…`) and each receiver family (the MT7612U with `MT7612U_FW_DIR`) must deliver
+both at parity. `tests/stream_timing_analyze.py` is the per-capture summary and the
 checks; `PACE_US` slows the producer for a part that cannot take 500 fps (the
 8812EU on 5 GHz). Which parts stamp an injected probe response is measured by
 `tests/probe_resp_egress_tsf_check.sh` (`HWBEACON=1` for the hardware

tests/mcast_da_rx_check.sh

@@ -0,0 +1,87 @@
+#!/usr/bin/env bash
+# mcast_da_rx_check.sh — gate for devourer#474 (telemetry in addr1): does each
+# receiver family's monitor RX deliver a stream frame whose DA is a
+# non-broadcast GROUP address — the very bytes FrameTimingExt ships in addr1
+# (07:…, src/StreamTelemetry.h) — as readily as the broadcast one? One injector (tests/mcast_da_tx.cpp) alternates the two; each witness
+# family in turn runs rxdemo with DEVOURER_STREAM_OUT=1 and the body tag says
+# which was which. Verdict per RX family: PASS when the group-DA count is at
+# least 90% of the broadcast count (and both are non-trivial), FAIL otherwise,
+# NONE when the pair decoded nothing (link problem, not an answer). Exit 77
+# when no cell could be measured at all — a skipped matrix is not a pass.
+#
+#   sudo -v && tests/mcast_da_rx_check.sh
+#   RX_CELLS="0x0bda:0xb812 0x2357:0x0120" tests/mcast_da_rx_check.sh
+# TX is the 8812CU unless the RX cell IS the 8812CU, then the 8822BU injects.
+# The MT7612U cell needs MT7612U_FW_DIR (decompressed mt7662 firmware).
+set -u
+ROOT="$(cd "$(dirname "$0")/.." && pwd)"; cd "$ROOT"
+CH="${CH:-6}"; SECS="${SECS:-15}"
+RX_CELLS="${RX_CELLS:-0x0bda:0xb812 0x0bda:0xc812 0x2357:0x0120 0x35bc:0x0101 0x0e8d:0x7612}"
+OUT="${OUT:-/tmp/devourer-mcast-da}"
+MT7612U_FW_DIR="${MT7612U_FW_DIR:-}"
+
+plugged() { lsusb -d "$(printf '%04x:%04x' "$1" "$2")" >/dev/null 2>&1; }
+cleanup() {
+    sudo pkill -INT -x mcast_da_tx 2>/dev/null || true
+    sudo pkill -INT -x rxdemo 2>/dev/null || true
+    sleep 1
+    sudo pkill -x mcast_da_tx 2>/dev/null || true
+    sudo pkill -x rxdemo 2>/dev/null || true
+}
+trap cleanup EXIT INT TERM
+
+echo "== build =="
+cmake --build build -j --target rxdemo devourer >/dev/null || exit 1
+if [ ! -x build/mcast_da_tx ] || [ tests/mcast_da_tx.cpp -nt build/mcast_da_tx ]; then
+    g++ -std=c++20 -O2 -Isrc -Iexamples/common tests/mcast_da_tx.cpp \
+        examples/common/env_config.cpp build/libdevourer.a \
+        $(pkg-config --cflags --libs libusb-1.0) -lpthread -o build/mcast_da_tx || exit 1
+fi
+mkdir -p "$OUT"; rm -f "${OUT:?}"/*.log "${OUT:?}"/*.jsonl
+fails=0; measured=0
+for cell in $RX_CELLS; do
+    rv=${cell%%:*}; rp=${cell#*:}
+    tag=$(printf '%04x_%04x' "$rv" "$rp")
+    if ! plugged "$rv" "$rp"; then echo "SKIP: RX $rv:$rp not plugged"; continue; fi
+    tv=0x0bda; tp=0xc812
+    [ "$rp" = "0xc812" ] && tp=0xb812
+    plugged "$tv" "$tp" || { echo "SKIP: TX $tv:$tp not plugged for RX $cell"; continue; }
+    extra=()
+    if [ "$rv" = "0x0e8d" ]; then
+        [ -n "$MT7612U_FW_DIR" ] || { echo "SKIP: MT7612U cell needs MT7612U_FW_DIR"; continue; }
+        extra=(DEVOURER_MT7612U_FW_DIR="$MT7612U_FW_DIR")
+    fi
+    echo "== RX $rv:$rp  <-  TX $tv:$tp on ch$CH =="
+    sudo env DEVOURER_VID="$rv" DEVOURER_PID="$rp" DEVOURER_CHANNEL="$CH" \
+        DEVOURER_STREAM_OUT=1 DEVOURER_LOG_LEVEL=warn DEVOURER_EVENTS=stdout "${extra[@]}" \
+        timeout $((SECS + 25)) ./build/rxdemo >"$OUT/rx_$tag.jsonl" 2>"$OUT/rx_$tag.log" &
+    sleep 8
+    sudo env DEVOURER_VID="$tv" DEVOURER_PID="$tp" DEVOURER_CHANNEL="$CH" DEVOURER_LOG_LEVEL=warn \
+        ./build/mcast_da_tx "$SECS" >"$OUT/tx_$tag.jsonl" 2>"$OUT/tx_$tag.log"
+    sleep 2; cleanup
+    measured=$((measured + 1))
+    python3 - "$OUT/rx_$tag.jsonl" "$OUT/tx_$tag.jsonl" "$cell" <<'PY' || fails=$((fails + 1))
+import json, sys
+sys.path.insert(0, "tests")
+from devourer_events import iter_events
+ff = g = 0
+for e in iter_events(open(sys.argv[1]), ev="rx.frame"):
+    if e.get("sa") != "57427505d600": continue
+    body = bytes.fromhex(e["body"])
+    if body.startswith(b"DAff"): ff += 1
+    elif body.startswith(b"DAgp"): g += 1
+tx = list(iter_events(open(sys.argv[2]), ev="mcast.tx"))
+sent = tx[-1] if tx else {}
+cell = sys.argv[3]
+if ff + g < 50:
+    print(f"NONE  rx={cell} ff={ff} group={g} (tx sent {sent.get('sent')})"); sys.exit(1)
+ratio = g / ff if ff else float("inf")
+verdict = "PASS" if ff >= 25 and ratio >= 0.9 else "FAIL"
+print(f"{verdict}  rx={cell} broadcast={ff} group={g} group/broadcast={ratio:.2f} "
+      f"(tx ok_ff={sent.get('ok_ff')} ok_gp={sent.get('ok_gp')})")
+sys.exit(0 if verdict == "PASS" else 1)
+PY
+done
+if [ "$measured" -eq 0 ]; then echo "SKIP: no receiver cell could be measured"; exit 77; fi
+echo "== done: $measured cell(s) measured, $fails non-PASS; logs in $OUT =="
+exit "$fails"