ARQ retry-limit sweep + post-ARQ FEC dimensioning (#362 steps 1+4)

docs/scheduled-mac.md

@@ -178,6 +178,18 @@ the value the descriptors used to hardcode) — the knob, not a descriptor
 constant, is now the single source of truth for the retry limit on
 jaguar1/2/3 (inert on Kestrel and the 8814A die).
 
+Choosing the limit (`tests/arq_retry_sweep.sh`, collision regime: a ~1 k fps
+retrying unicast flood into an 8812EU duplex ground station airing
+PixelPilot-shaped feedback bursts, near-field): retries are backoff-spaced,
+so a small limit can burn entirely inside one 2–3 ms burst. Measured curve —
+limit 3: 99.72% delivered, residual 0.26%; limit 8: 99.97%, residual 0.03%
+(gaps ≤3, a K=8/N=11 FEC floor covers it); limit 16: 100.00% at +5.4%
+retry airtime (mean 0.054 retries/frame); limit 32: no further gain, +17%
+more retries than 16. Queue-time p99 is flat across limits (only the rare
+worst case doubles, then stops growing). Prefer **16** on an ARQ link, or
+**8 + a light FEC floor** where airtime is precious; the per-run residual
+gap analysis is `tests/arq_fec_dimension.py`.
+
 Responder-side capability (same setup, J3 TX as the reference soliciting
 station): **8814AU** closes the loop at retries ~0.1 (the bench responder of
 choice); **8812AU** works but degraded (97% delivery at ~7 mean retries —

tests/arq_fec_dimension.py

@@ -0,0 +1,119 @@
+#!/usr/bin/env python3
+"""Post-ARQ residual gap analysis for FEC dimensioning.
+
+For each recorded arq_e2e run directory: reconstruct the undelivered set
+(report ok=0, plus unreported-and-undelivered — conservative, slightly
+overcounts), then measure the RUN-LENGTH distribution of consecutive
+undelivered frame indices. That is the quantity wfb-style block FEC cares
+about: a (K,N) block recovers up to N-K losses per block window, so the
+residual gap-length percentiles map directly onto the N-K needed.
+
+Caveats the numbers carry: the mapping printed is SINGLE-gap coverage per
+block window (two gaps landing in one window need their sum); the ledgers are
+near-field bench data, so the residual is collision/stall structure, not
+range-fade; and the undelivered set includes frames with no report verdict at
+all (report coverage is load-dependent), which errs toward larger residuals —
+the safe direction for dimensioning.
+
+  python3 tests/arq_fec_dimension.py /tmp/arq-e2e/<run> [<run>...]
+"""
+import argparse
+import json
+from collections import Counter
+
+TAIL_GUARD = 512  # mirror arq_e2e_analyze: stream-end truncation window
+
+
+def undelivered(rundir):
+    dut = set()
+    with open(f"{rundir}/dut.jsonl", errors="replace") as f:
+        for line in f:
+            if line.startswith('{"ev":"rx.seq"'):
+                try:
+                    dut.add(json.loads(line)["pctr"])
+                except Exception:
+                    pass
+    rep = {}
+    prev = None
+    r = 0
+    with open(f"{rundir}/drone.jsonl", errors="replace") as f:
+        for line in f:
+            if not line.startswith('{"ev":"tx.report"'):
+                continue
+            try:
+                ev = json.loads(line)
+            except Exception:
+                continue
+            t = ev.get("tag")
+            if t is None:
+                continue
+            if prev is not None:
+                r += (t - prev) % 256
+            prev = t
+            rep[r] = bool(ev.get("ok"))
+    if not rep or not dut:
+        raise SystemExit(
+            f"{rundir}: {'no tagged tx.report events' if not rep else ''}"
+            f"{' and ' if not rep and not dut else ''}"
+            f"{'no rx.seq ledger' if not dut else ''} — an empty ledger would "
+            f"count every frame as undelivered (J1-format reports carry no "
+            f"tag; this tool needs a halmac TX side)")
+    hi = max(max(dut, default=0), r)
+    lo_cut = min(dut) if dut else 0  # drone frames before the DUT RX was up
+    hi_cut = hi - TAIL_GUARD
+    miss = [k for k in range(lo_cut, hi_cut)
+            if k not in dut and rep.get(k) is not True]
+    gaps = []
+    run = 0
+    prev_k = None
+    for k in miss:
+        if prev_k is not None and k == prev_k + 1:
+            run += 1
+        else:
+            if run:
+                gaps.append(run)
+            run = 1
+        prev_k = k
+    if run:
+        gaps.append(run)
+    return gaps, len(miss), max(1, hi_cut - lo_cut)
+
+
+def pct(sorted_g, p):
+    """Nearest-rank percentile: ceil(p*n)-1, 0-based. int(p*n) would be
+    biased one rank upward (P50 of 4 elements landing on the 3rd)."""
+    if not sorted_g:
+        return 0
+    import math
+    return sorted_g[min(len(sorted_g) - 1,
+                        max(0, math.ceil(p * len(sorted_g)) - 1))]
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("runs", nargs="+")
+    ap.add_argument("--k", type=int, nargs="*", default=[8, 12],
+                    help="FEC block K values to map the residual onto")
+    a = ap.parse_args()
+    for rundir in a.runs:
+        gaps, n_miss, span = undelivered(rundir)
+        g = sorted(gaps)
+        hist = Counter(gaps)
+        top = ", ".join(f"{k}x{v}" for k, v in sorted(hist.items())[:8])
+        name = rundir.rstrip("/").split("/")[-1]
+        print(f"\n== {name}: undelivered {n_miss}/{span} "
+              f"({100.0 * n_miss / span:.2f}%), {len(g)} gaps")
+        print(f"   lengths: [{top}{', ...' if len(hist) > 8 else ''}]")
+        print(f"   P50={pct(g, .5)} P99={pct(g, .99)} "
+              f"P99.9={pct(g, .999)} max={g[-1] if g else 0}")
+        need = pct(g, .999)
+        for K in a.k:
+            if need:
+                print(f"   K={K}: N-K >= {need} to cover the P99.9 single gap "
+                      f"-> N={K + need} (rate {K / (K + need):.2f})")
+            else:
+                print(f"   K={K}: residual ~gap-free")
+
+
+if __name__ == "__main__":
+    main()

tests/arq_retry_sweep.sh

@@ -0,0 +1,73 @@
+#!/usr/bin/env bash
+#
+# arq_retry_sweep.sh — hardware-ARQ retry-limit vs delivery/airtime curve.
+#
+# Hardware retries are backoff-spaced, so a small DEVOURER_TX_RETRY_LIMIT can
+# burn entirely inside one ground-station feedback burst (2-3 ms) and drop the
+# frame, while a larger limit straddles the burst tail and delivers late.
+# This sweep runs the arq_e2e bench (collision regime: default async DUT, no
+# consumer stalls — the residual is pure burst-collision loss) once per retry
+# limit and tabulates:
+#   delivered%   — reports ok=1 / reports
+#   mean_retries — airtime-cost proxy (each retry re-airs the whole frame)
+#   drops        — reports ok=0 (retry budget exhausted)
+# and hands each run to arq_fec_dimension.py for the post-ARQ residual the
+# FEC floor must cover — the (retry_limit, K/N) pairing dataset.
+#
+#   sudo bash tests/arq_retry_sweep.sh
+#   LIMITS="3 8" CYCLES=3 sudo bash tests/arq_retry_sweep.sh
+set -u
+ROOT="$(cd "$(dirname "$0")/.." && pwd)"
+LIMITS=${LIMITS:-"3 8 16 32"}
+OUT=${OUT:-/tmp/arq-retry-sweep/$(date +%Y%m%d-%H%M%S)}
+mkdir -p "$OUT"
+[ "$(id -u)" = 0 ] || { echo "must run as root"; exit 3; }
+
+declare -A RUNDIR
+for L in $LIMITS; do
+  echo "=== retry_limit=$L"
+  # Deterministic per-limit run dir (the harness honours OUT=) — inferring
+  # "newest /tmp/arq-e2e/*" would race any concurrent bench run on the host.
+  RUNDIR[$L]="$OUT/limit_$L"
+  RETRY_LIMIT="$L" OUT="${RUNDIR[$L]}" bash "$ROOT/tests/arq_e2e_delivery.sh" \
+      >"$OUT/limit_$L.log" 2>&1 || {
+    echo "run failed (see $OUT/limit_$L.log)"; exit 1; }
+  echo "    -> ${RUNDIR[$L]}"
+done
+
+echo
+printf "%8s %10s %12s %8s %10s %12s %12s\n" \
+       limit reports "delivered%" drops mean_rtry "drops@6M:10" "drops@6M:30"
+for L in $LIMITS; do
+  D=${RUNDIR[$L]}
+  # Per-burst-phase drop breakout from the run's own per-phase report table —
+  # the limit-vs-burst-length interaction is the curve's point: a bigger burst
+  # needs more backoff-spaced retries to straddle.
+  B10=$(awk '$1=="6M:10"{s+=$8} END{print s+0}' "$D/report.txt")
+  B30=$(awk '$1=="6M:30"{s+=$8} END{print s+0}' "$D/report.txt")
+  python3 - "$D/drone.jsonl" "$L" "$B10" "$B30" <<'PYEOF'
+import json, sys
+n = ok = drops = 0
+rsum = 0
+for line in open(sys.argv[1], errors="replace"):
+    if not line.startswith('{"ev":"tx.report"'):
+        continue
+    try:
+        ev = json.loads(line)
+    except Exception:
+        continue
+    n += 1
+    rsum += ev.get("retries", 0)
+    if ev.get("ok"):
+        ok += 1
+    else:
+        drops += 1
+print(f"{sys.argv[2]:>8} {n:>10} {100.0*ok/max(1,n):>11.2f} "
+      f"{drops:>8} {rsum/max(1,n):>10.3f} {sys.argv[3]:>12} {sys.argv[4]:>12}")
+PYEOF
+done | tee "$OUT/summary.txt"
+
+echo
+python3 "$ROOT/tests/arq_fec_dimension.py" \
+    $(for L in $LIMITS; do echo "${RUNDIR[$L]}"; done) | tee -a "$OUT/summary.txt"
+echo "[sweep] logs: $OUT"