diff --git a/AGENTS.md b/AGENTS.md index daf9a08..0865104 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -112,6 +112,57 @@ tests/compat/run-matrix.sh --tier blocking # whole tier - Both tiers (blocking and informational) gate PR CI; only the weekly upstream-drift job (unpinned HEADs) is non-blocking. +### srt-sink extended metrics + +`tests/compat/srt-sink/` is the mock SRT receiver used by the compat harness. Beyond the original 4 frozen keys (`bytes_received`, `disconnects`, `handshake_ms`, `error`), `srt-sink` now emits 6 additional keys in `result.json`: + +| Key | Source | Description | +|-----|--------|-------------| +| `ts_packets` | `ts_continuity.h` | Total 188-byte TS packets received | +| `ts_sync_errors` | `ts_continuity.h` | Packets whose first byte is not `0x47` | +| `ts_cc_errors` | `ts_continuity.h` | Continuity-counter discontinuities (excl. null PID 0x1FFF and adaptation-field `discontinuity_indicator`) | +| `pkt_rcv_loss` | `srt_bstats` `pktRcvLossTotal` | SRT-level receive loss (cumulative, summed across reconnects) | +| `pkt_rcv_drop` | `srt_bstats` `pktRcvDropTotal` | SRT-level receive drop (too-late packets) | +| `pkt_retrans` | `srt_bstats` `pktRetransTotal` | SRT-level retransmissions | + +The TS parser lives in `srt-sink/ts_continuity.h` (header-only, dependency-free). It reassembles 188-byte packets across `srt_recv` boundaries, tracks per-PID continuity counters, and excludes null PID `0x1FFF` and adaptation-only packets from CC checks. Unit tests: `srt-sink/ts_continuity_test.cpp` (registered as ctest `ts-continuity` in `srt-sink/CMakeLists.txt`). + +**`--retransmitalgo 0|1`** — new `srt-sink` flag. Sets `SRTO_RETRANSMITALGO` on the listener (pre-bind, inherited by accepted sockets). `0` = always retransmit on NAK; `1` = selective retransmit (default SRT behavior). + +**`--packetfilter `** — new `srt-sink` flag. Sets `SRTO_PACKETFILTER` on the listener (pre-bind, inherited by accepted sockets). The accepted socket's negotiated filter is written to `result.json` as `"packetfilter"` (non-empty = FEC negotiated; `""` = responder cleared the filter). + +**`--reorderfreeze 0|1`** — new `srt-sink` flag. Sets `SRTO_REORDERFREEZE` via the raw numeric opt id `(SRT_SOCKOPT)120` so it compiles against any libsrt version. Reports `reorderfreeze=on|off|unsupported` in the startup banner. + +### Profile validation A/B matrix + +`tests/compat/scenarios/profile-validation-matrix.sh` is the A/B orchestrator for the four non-FEC receive profiles. It runs paired alternating reps (baseline patched libsrt vs freeze profile) under netem reorder stress and gates on six quality clauses: + +1. `disconnects == 0` (both arms) +2. `ts_sync_errors == 0` (profile arm) +3. `ts_cc_errors <= baseline median` +4. `median goodput >= 99% baseline` +5. `p95 pkt_rcv_drop <= baseline` +6. `wire_amp <= 1.10× baseline median` (wire bytes / bytes_received) + +Registered in `matrix.yaml` as scenario `profile-validation-matrix` (tier: blocking, privileged: true). Run manually via `tests/compat/run-matrix.sh --tier blocking`. Results: all four non-FEC profiles (Balanced/Low-Latency/Resilient/Classic) PASS all six clauses; wire amplification ratios 1.054–1.078× (well under 1.10×). Evidence: `test-results/srt-receive-profiles/task-6-srt-receive-profiles.json`. + +The `reorder-stress.sh` scenario is parameterized (BITRATE_KBPS, RX_LATENCY_MS, NAKREPORT, LOSSMAXTTL, REORDERFREEZE, PROFILE_LABEL, NETEM_SEED) and now emits TS-continuity + SRT counters + `goodput_bps` + `wire_amp` into `result.json`. Default run is byte-identical to the pre-matrix behavior (Rule E). + +### FEC connect-matrix + +`tests/compat/scenarios/fec-connect-matrix.sh` proves the one-sided FEC packet-filter negotiation behavior (direct SRT loopback; SRTLA is transparent UDP so negotiation is SRT-level). Four cases: + +| Case | Caller filter | Listener filter | Result | +|------|--------------|-----------------|--------| +| (a) | FEC full config | `fec` (accept-form) | FEC negotiated — `packetfilter` non-empty | +| (b) | plain (no filter) | `fec` (accept-form) | PLAIN — responder clears filter, `packetfilter=""` | +| (c) | FEC full config | conflicting `fec,cols:20,rows:20` | HARD REJECT `SRT_REJ_FILTER` — `bytes_received=0` | +| (d) | FEC full config | no filter | ADOPTED — listener adopts caller config (informational) | + +Cases (a)/(b)/(c) are gated; (d) is informational. Registered in `matrix.yaml` as scenario `fec-connect-matrix`. See `docs/COMPATIBILITY.md §6` for the full mechanism and empirical results. + +**Key finding:** a listener with NO `packetfilter` does NOT reject a FEC caller — it adopts the caller's config (SRT `checkApplyFilterConfig` "good deal" else-branch). The genuine `SRT_REJ_FILTER` hard reject is a filter-config CONFLICT, not the absence of a filter. This is why L1 in `irl-srt-server` uses the accept-form `"fec"` and serves both FEC and non-FEC callers on the same port. + ## COMPATIBILITY TESTING The compat matrix (`tests/compat/matrix.yaml`) registers every tested sender/receiver diff --git a/docs/COMPATIBILITY.md b/docs/COMPATIBILITY.md index 5373a22..29acd3c 100644 --- a/docs/COMPATIBILITY.md +++ b/docs/COMPATIBILITY.md @@ -192,7 +192,59 @@ The CeraLive stack uses two patched libsrt forks (CERALIVE/srt for the device im --- -## 6. ENABLE_ALGO_COMPARISON Decision +## 6. SRT FEC Connect-Matrix (one-sided packet-filter negotiation) + +FEC is an **SRT-level** feature: `SRTO_PACKETFILTER=fec` is negotiated in the SRT +handshake, end-to-end between the caller and the listener. SRTLA underneath is a +transparent UDP relay and does not touch it — so a FEC stream "rides" whichever +SRT listener it terminates on. The receive-profile design uses one **fec-accept** +listener (L1, `SRTO_PACKETFILTER=fec` — just the type) for device senders. This +matrix proves that one such listener serves both FEC and non-FEC senders, and +where the negotiation actually hard-fails. + +The caller is the device (initiator); the listener is the cloud receiver +(responder). Filter strings: a full-config FEC sender is +`fec,layout:staircase,rows:10,cols:10,arq:onreq`; the fec-accept listener is just +`fec`. + +| # | Caller (sender) | Listener config | Result | Negotiated `SRTO_PACKETFILTER` on accepted socket | +|---|---|---|---|---| +| a | FEC (full config) | `fec` (fec-accept) | **Connect — FEC negotiated** | non-empty (merged config) | +| b | plain (no filter) | `fec` (fec-accept) | **Connect — PLAIN** (responder clears it per-connection) | empty `""` | +| c | FEC (full config) | conflicting `fec,…` (incompatible dims) | **HARD REJECT — `SRT_REJ_FILTER`** | n/a (no accept, 0 bytes) | +| d | FEC (full config) | *no packetfilter* (empty) | **Connect — FEC adopted** by the listener ("good deal") | non-empty (caller's config) | + +**Why case (b) means no separate FEC listener is needed.** A listener that set a +filter the caller never requested does not reject — the responder branch in +`srtcore/core.cpp` (`checkApplyFilterConfig` + the post-handshake check "agent has +configured packetfilter, but peer didn't request it") **clears** the filter for +that one connection and connects plain. So the single fec-accept L1 accepts a FEC +device (case a, full negotiation) **and** a non-FEC sender such as BELABOX +(case b, cleared per-connection). One listener, both senders — no second FEC port. + +**Where the reject boundary actually is.** The genuine `SRT_REJ_FILTER` hard +reject (case c) is a filter-config **conflict**, *not* the mere absence of a +filter. A listener with **no** packetfilter does **not** reject a FEC caller — it +takes the caller's config as a "good deal" and runs FEC anyway (case d). This +corrects the earlier mental model that a "non-FEC listener" would reject a FEC +sender: on a packet-filter-capable libsrt (≥ 1.4.0; system libsrt 1.5.x) absence +is permissive (adopt), and only an irreconcilable config closes the connection. +The one-sided config rule is documented upstream in +[`srt/docs/features/packet-filtering-and-fec.md`](https://github.com/Haivision/srt/blob/master/docs/features/packet-filtering-and-fec.md) +("one party defines the full configuration while the other only defines the +matching packet filter type … if the options specified are in conflict, the +connection will be rejected"). + +This matrix is exercised end-to-end by the +[`fec-connect-matrix`](../tests/compat/scenarios/fec-connect-matrix.sh) harness +scenario, which drives a real FEC/plain SRT caller into `srt-sink --packetfilter` +and asserts the negotiated filter the sink reads off each accepted socket +(`"packetfilter"` in the result JSON). Cases (a)+(b)+(c) gate the scenario; +case (d) is recorded as an informational observation. + +--- + +## 7. ENABLE_ALGO_COMPARISON Decision `ENABLE_ALGO_COMPARISON` is defined in `src/receiver_config.h` with a default of `1`: @@ -213,7 +265,7 @@ Changing this default requires a deliberate ADR, not a drive-by edit. --- -## 7. Maintenance Policy +## 8. Maintenance Policy ### Pin refresh When a Tier 1 or Tier 2 implementation cuts a new release, update the pin in diff --git a/tests/compat/matrix.yaml b/tests/compat/matrix.yaml index 8190aca..b577fc4 100644 --- a/tests/compat/matrix.yaml +++ b/tests/compat/matrix.yaml @@ -269,6 +269,72 @@ scenarios: precedent), end-to-end SRT `disconnects` is recorded but is NOT a pass gate for the isolation phase. Unprivileged: netem_require exits 77 (SKIP). + - name: ts-continuity-loopback + script: tests/compat/scenarios/ts-continuity-loopback.sh + requires: srt-live-transmit + skip_exit: 3 + gated_on_disconnects: false + proves: >- + Real-SRT loopback over 127.0.0.1 proving srt-sink's MPEG-TS continuity + metric is correct AND falsifiable: a clean transport stream sent raw + yields ts_sync_errors==0 and ts_cc_errors==0, while the SAME TS with one + payload-bearing packet excised yields ts_cc_errors>0. Raw byte passthrough + (srt-live-transmit, NOT ffmpeg — which re-muxes the container and would + heal the injected drop) preserves the continuity break end-to-end; the + send is paced to ~real time and null-padded to a whole SRT chunk so the + live socket neither drops nor zero-pads into spurious sync errors. + Unprivileged; SKIPs (exit 3) without srt-live-transmit. Integration + companion to the hermetic `ts-continuity` ctest (same proof, no network). + + - name: fec-connect-matrix + script: tests/compat/scenarios/fec-connect-matrix.sh + requires: srt-live-transmit + skip_exit: 3 + gated_on_disconnects: false + proves: >- + SRT one-sided packet-filter (FEC) connect negotiation against a single + fec-accept listener (srt-sink --packetfilter fec), the receiver profile + (L1) that device FEC senders ride. Reads the negotiated SRTO_PACKETFILTER + off the accepted socket from srt-sink's result JSON. (a) a full-config FEC + caller negotiates FEC: the accepted socket reports a non-empty merged + config and bytes flow. (b) a plain caller against the SAME listener + connects PLAIN: the responder clears the filter per-connection (accepted + SRTO_PACKETFILTER empty), bytes flow, NOT rejected — proving ONE fec-accept + listener serves both FEC and non-FEC senders, so no separate FEC port is + needed. (c) an irreconcilable listener filter config HARD-REJECTS with + SRT_REJ_FILTER (caller logs ERROR:FILTER, sink accepts nothing, 0 bytes). + Falsifiable: (a) and (b) share the identical listener yet must report + opposite filter results. Also records (informational, not gated) that a + no-filter listener ADOPTS a FEC caller's config (good-deal) rather than + rejecting — the reject boundary is a config CONFLICT, not the absence of a + filter. Unprivileged; SKIPs (exit 3) without srt-live-transmit. + + - name: profile-validation-matrix + script: tests/compat/scenarios/profile-validation-matrix.sh + tier: blocking + privileged: true + notes: >- + A/B validation matrix: the 4 NON-FEC receive profiles vs the patched-libsrt + baseline under cross-link reorder stress, judged on srt-sink TS-continuity + (Task 5), NOT a bytes-only proxy. Drives scenarios/reorder-stress.sh paired/ + alternating (shared per-rep netem seed) and swaps libsrt under srt-sink: the + profile arm = the reorderfreeze-1.5.5 build (opt-in SRTO_REORDERFREEZE via + srt-sink --reorderfreeze) plus the recipe flags; the baseline arm = the + patched CERALIVE/srt build (unconditional BELABOX merge, no flag). Profile + pairs (each freeze-arm vs its paired baseline at matched latency/bitrate/ + seed): Balanced freeze+NAK @1500ms, Low-Latency freeze+NAK @250ms, Resilient + freeze+NAK @3500ms, Classic freeze+NAK-off @800ms. EQUAL gate per pair: + disconnects==0, ts_sync_errors==0, ts_cc_errors<=baseline, median goodput>= + 99% baseline, p95 late-drop<=baseline, wire-amplification<=1.10x baseline. + Plus a regression-validation cell (Balanced NAK-on vs baseline @{500,1500, + 3500}ms) and a LOSSMAXTTL sweep axis {30,200,1000} to pick the cap. Blocking + verdict = all 4 non-FEC profiles PASS AND the regression cell PASSes; the + sweep + a stock-decay control arm are informational/falsifiability. FEC + scope: the Low-Latency+FEC row is gated separately (fec-connect-matrix). + Builds the two libsrt via lib/build-libsrt-matrix.sh if absent; evidence + table -> test-results/srt-receive-profiles/task-6-srt-receive-profiles.json. + Unprivileged: netem.sh `require` exits 77 (SKIP). + excluded: - name: moblink-rust repo: https://github.com/datagutt/moblink-rust diff --git a/tests/compat/scenarios/fec-connect-matrix.sh b/tests/compat/scenarios/fec-connect-matrix.sh new file mode 100755 index 0000000..6c1c617 --- /dev/null +++ b/tests/compat/scenarios/fec-connect-matrix.sh @@ -0,0 +1,255 @@ +#!/usr/bin/env bash +# +# fec-connect-matrix.sh — proves SRT's one-sided packet-filter (FEC) connect +# negotiation against a single "fec-accept" listener (SRTO_PACKETFILTER=fec), +# the receiver profile (L1) that the device's FEC senders ride. +# +# Each case runs a real SRT caller (srt-live-transmit, FEC-capable) into an +# srt-sink listener over 127.0.0.1 and reads the connection's negotiated filter +# from the sink's --result JSON ("packetfilter": the SRTO_PACKETFILTER value on +# the accepted data socket — non-empty when FEC was agreed, "" when cleared): +# +# (a) FEC caller -> fec-accept listener : FEC NEGOTIATED. +# caller sets the full config (fec,layout:staircase,rows:10,cols:10, +# arq:onreq); the listener sets just the type (fec). The accepted socket +# reports a non-empty merged config and bytes flow. +# +# (b) plain caller -> fec-accept listener : connects PLAIN, NOT rejected. +# the SAME listener with no FEC caller clears its filter for that one +# connection (responder branch, core.cpp checkApplyFilterConfig). The +# accepted socket reports an EMPTY packetfilter and bytes flow. This is +# why ONE fec-accept listener serves BOTH FEC and non-FEC (BELABOX) senders +# and no separate FEC port is needed. +# +# (c) FEC caller -> conflicting-filter listener : HARD REJECT (SRT_REJ_FILTER). +# when the listener's filter config is irreconcilable with the caller's, +# the handshake is rejected: the caller logs ERROR:FILTER and the sink +# never accepts (bytes_received == 0). This is the genuine reject boundary. +# +# (d) FEC caller -> no-filter listener : ADOPTS (informational, NOT gated). +# a listener with NO packetfilter does NOT reject a FEC caller — it takes +# the caller's config as a "good deal" and runs FEC anyway. So the reject +# boundary in (c) is a config CONFLICT, never the mere ABSENCE of a filter. +# Reported for completeness; the pass gate is (a) && (b) && (c). +# +# Falsifiability: (a) and (b) share the identical listener (--packetfilter fec) +# yet must report opposite results (non-empty vs empty), so neither can be +# hard-coded; (c) must produce a real SRT_REJ_FILTER with zero bytes. +# +# Privilege: NONE. Requires srt-live-transmit (libsrt-tools); without it the +# scenario SKIPs cleanly (exit 3), like the other capability-gated scenarios. +# +# Usage: +# fec-connect-matrix.sh [--build-dir DIR] [--duration SEC] [--keep-logs] [-h] +# --duration SEC per-case stream length (default 3). +# +# Artifacts land in tests/compat/results/fec-connect-matrix/ (gitignored); +# nothing is written outside the repo and no `../`-escaping path is used (Rule D). +# +set -uo pipefail + +SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" >/dev/null 2>&1 && pwd)" +REPO_ROOT="$(cd -- "${SCRIPT_DIR}/../../.." >/dev/null 2>&1 && pwd)" +RESULTS_DIR="${SCRIPT_DIR}/../results/fec-connect-matrix" + +log() { printf '%s\n' "$*" >&2; } +die() { printf 'fec-connect-matrix: %s\n' "$*" >&2; exit 2; } + +# --------------------------------------------------------------------------- # +# CLI # +# --------------------------------------------------------------------------- # +BUILD_DIR="${SRTLA_BUILD_DIR:-}" +KEEP_LOGS=0 +MEDIA_SEC=3 + +while [[ $# -gt 0 ]]; do + case "$1" in + --build-dir) BUILD_DIR="${2:?--build-dir needs a value}"; shift 2 ;; + --duration) MEDIA_SEC="${2:?--duration needs a value}"; shift 2 ;; + --keep-logs) KEEP_LOGS=1; shift ;; + -h|--help) sed -n '2,46p' "${BASH_SOURCE[0]}" | sed 's/^# \{0,1\}//'; exit 0 ;; + *) die "unknown argument '$1' (try --help)" ;; + esac +done +[[ "$MEDIA_SEC" =~ ^[0-9]+$ && "$MEDIA_SEC" -ge 1 ]] || die "--duration must be a positive integer" + +for tool in jq python3; do + command -v "$tool" >/dev/null 2>&1 || die "required tool '$tool' not found in PATH" +done +if ! command -v srt-live-transmit >/dev/null 2>&1; then + log "SKIP fec-connect-matrix: srt-live-transmit not found (install libsrt-tools)" + mkdir -p "$RESULTS_DIR" + printf '{"scenario":"fec-connect-matrix","skipped":true,"reason":"srt-live-transmit not installed"}\n' \ + > "${RESULTS_DIR}/result.json" + exit 3 +fi + +# --------------------------------------------------------------------------- # +# Resolve the srt-sink helper (built with -DBUILD_COMPAT_TESTS=ON). # +# --------------------------------------------------------------------------- # +resolve_build_dir() { + local candidates=() + [[ -n "$BUILD_DIR" ]] && candidates+=("$BUILD_DIR") + candidates+=("${REPO_ROOT}/build" "/tmp/srtla-build") + local d + for d in "${candidates[@]}"; do + [[ -x "${d}/tests/compat/srt-sink/srt-sink" ]] && { printf '%s' "$d"; return 0; } + done + return 1 +} +BUILD_DIR="$(resolve_build_dir)" || die \ + "srt-sink not found. Build with: + cmake -B build -DBUILD_COMPAT_TESTS=ON && cmake --build build -j" +SRT_SINK="${BUILD_DIR}/tests/compat/srt-sink/srt-sink" + +# --------------------------------------------------------------------------- # +# Constants / filter configs # +# --------------------------------------------------------------------------- # +FEC_FULL="fec,layout:staircase,rows:10,cols:10,arq:onreq" # device FEC sender +FEC_ACCEPT="fec" # L1 fec-accept form +FEC_CONFLICT="fec,layout:even,rows:20,cols:20,arq:always" # irreconcilable dims + +SRT_LATENCY_MS=300 +SINK_DURATION=$((MEDIA_SEC + 12)) # sink self-timeout guard, past the send +TX_TIMEOUT=$((MEDIA_SEC + 9)) # hard cap on a hung srt-live-transmit +SLT_EXIT_SEC=$((MEDIA_SEC + 3)) # srt-live-transmit's own since-start timer +PAYLOAD_BYTES=$((MEDIA_SEC * 130000)) # ~1 Mbit/s of bytes, paced to real time + +rm -rf "$RESULTS_DIR"; mkdir -p "$RESULTS_DIR" +PAYGEN="${RESULTS_DIR}/paygen.py" +RESULT_JSON="${RESULTS_DIR}/result.json" + +PIDS=() +cleanup() { local p; for p in "${PIDS[@]:-}"; do [[ -n "$p" ]] && kill -TERM "$p" 2>/dev/null; done; } +trap cleanup EXIT INT TERM + +# Emit PAYLOAD_BYTES of content as whole 1316-byte (7x188) chunks, paced over +# MEDIA_SEC so the SRT live socket sees a real-time stream, not a loss-inducing +# burst. Content is irrelevant here (we count bytes, not TS), so it is a fixed +# pattern; full chunks avoid a short final message. +cat > "$PAYGEN" <<'PY' +import sys, time +total = int(sys.argv[1]); dur = float(sys.argv[2]) +CH = 1316 +chunk = bytes(i % 256 for i in range(CH)) +n = max(total // CH, 1) +interval = dur / n +w = sys.stdout.buffer +t0 = time.monotonic() +for i in range(n): + w.write(chunk); w.flush() + target = t0 + (i + 1) * interval + dt = target - time.monotonic() + if dt > 0: + time.sleep(dt) +PY + +# --------------------------------------------------------------------------- # +# Run one case: srt-sink listener (optional --packetfilter) <- FEC/plain caller. # +# Sets globals: R_BYTES, R_FILTER, R_REJECT (1 if caller saw a filter reject). # +# --------------------------------------------------------------------------- # +run_case() { + local name="$1" sink_pf="$2" call_pf="$3" port="$4" + local res="${RESULTS_DIR}/${name}.json" + local sinklog="${RESULTS_DIR}/${name}-sink.log" + local calllog="${RESULTS_DIR}/${name}-call.log" + rm -f "$res" + + local sinkargs=(--port "$port" --host 127.0.0.1 --result "$res" + --latency "$SRT_LATENCY_MS" --duration "$SINK_DURATION") + [[ -n "$sink_pf" ]] && sinkargs+=(--packetfilter "$sink_pf") + "$SRT_SINK" "${sinkargs[@]}" >"$sinklog" 2>&1 & + local sp=$!; PIDS+=("$sp") + sleep 0.7 + + local curl="srt://127.0.0.1:${port}?mode=caller&transtype=live&latency=${SRT_LATENCY_MS}" + [[ -n "$call_pf" ]] && curl="${curl}&packetfilter=${call_pf}" + ( python3 "$PAYGEN" "$PAYLOAD_BYTES" "$MEDIA_SEC" \ + | timeout "$TX_TIMEOUT" srt-live-transmit -t:"$SLT_EXIT_SEC" -chunk:1316 \ + "file://con" "$curl" ) >"$calllog" 2>&1 || true + sleep 1.0 + kill -TERM "$sp" 2>/dev/null; wait "$sp" 2>/dev/null + + R_BYTES="$(jq -r '.bytes_received // -1' "$res" 2>/dev/null || echo -1)" + R_FILTER="$(jq -r '.packetfilter // ""' "$res" 2>/dev/null || echo "")" + if grep -qiE "ERROR:FILTER|Packet Filter settings error|REJECT reported from HS" "$calllog"; then + R_REJECT=1 + else + R_REJECT=0 + fi + [[ "$R_BYTES" =~ ^-?[0-9]+$ ]] || R_BYTES=-1 +} + +# --------------------------------------------------------------------------- # +# Execute the matrix. # +# --------------------------------------------------------------------------- # +log "==> fec-connect-matrix (build dir: ${BUILD_DIR})" + +run_case "case-a-fec-to-fecaccept" "$FEC_ACCEPT" "$FEC_FULL" 4871 +A_BYTES="$R_BYTES"; A_FILTER="$R_FILTER"; A_REJECT="$R_REJECT" +log " (a) FEC -> fec-accept : bytes=${A_BYTES} filter='${A_FILTER}' reject=${A_REJECT}" + +run_case "case-b-plain-to-fecaccept" "$FEC_ACCEPT" "" 4872 +B_BYTES="$R_BYTES"; B_FILTER="$R_FILTER"; B_REJECT="$R_REJECT" +log " (b) plain -> fec-accept : bytes=${B_BYTES} filter='${B_FILTER}' reject=${B_REJECT}" + +run_case "case-c-fec-to-conflict" "$FEC_CONFLICT" "$FEC_FULL" 4873 +C_BYTES="$R_BYTES"; C_FILTER="$R_FILTER"; C_REJECT="$R_REJECT" +log " (c) FEC -> conflict : bytes=${C_BYTES} filter='${C_FILTER}' reject=${C_REJECT}" + +run_case "case-d-fec-to-empty" "" "$FEC_FULL" 4874 +D_BYTES="$R_BYTES"; D_FILTER="$R_FILTER"; D_REJECT="$R_REJECT" +log " (d) FEC -> no-filter : bytes=${D_BYTES} filter='${D_FILTER}' reject=${D_REJECT} (informational)" + +# --------------------------------------------------------------------------- # +# Verdict. # +# --------------------------------------------------------------------------- # +a_ok=false +[[ "$A_BYTES" -ge 1000 && "$A_REJECT" -eq 0 && -n "$A_FILTER" && "$A_FILTER" == fec* ]] && a_ok=true +b_ok=false +[[ "$B_BYTES" -ge 1000 && "$B_REJECT" -eq 0 && -z "$B_FILTER" ]] && b_ok=true +c_ok=false +[[ "$C_BYTES" -eq 0 && "$C_REJECT" -eq 1 && -z "$C_FILTER" ]] && c_ok=true +# (d) is informational: a no-filter listener adopts the caller's FEC config. +d_adopts=false +[[ "$D_BYTES" -ge 1000 && "$D_REJECT" -eq 0 && -n "$D_FILTER" && "$D_FILTER" == fec* ]] && d_adopts=true + +pass=false +[[ "$a_ok" == true && "$b_ok" == true && "$c_ok" == true ]] && pass=true + +jq -n \ + --argjson pass "$pass" \ + --argjson a_ok "$a_ok" --argjson b_ok "$b_ok" --argjson c_ok "$c_ok" \ + --argjson d_adopts "$d_adopts" \ + --argjson a_bytes "$A_BYTES" --arg a_filter "$A_FILTER" --argjson a_reject "$A_REJECT" \ + --argjson b_bytes "$B_BYTES" --arg b_filter "$B_FILTER" --argjson b_reject "$B_REJECT" \ + --argjson c_bytes "$C_BYTES" --arg c_filter "$C_FILTER" --argjson c_reject "$C_REJECT" \ + --argjson d_bytes "$D_BYTES" --arg d_filter "$D_FILTER" --argjson d_reject "$D_REJECT" \ + --arg ts "$(date -u +%Y-%m-%dT%H:%M:%SZ)" \ + '{ + scenario:"fec-connect-matrix", pass:$pass, + criteria:{a_fec_negotiated:$a_ok, b_plain_accepted:$b_ok, c_hard_reject:$c_ok}, + cases:{ + a_fec_to_fecaccept:{bytes:$a_bytes, packetfilter:$a_filter, reject:$a_reject}, + b_plain_to_fecaccept:{bytes:$b_bytes, packetfilter:$b_filter, reject:$b_reject}, + c_fec_to_conflict:{bytes:$c_bytes, packetfilter:$c_filter, reject:$c_reject}, + d_fec_to_empty_informational:{bytes:$d_bytes, packetfilter:$d_filter, reject:$d_reject, adopts:$d_adopts} + }, + timestamp:$ts + }' > "$RESULT_JSON" + +log "" +log "================ fec-connect-matrix summary ================" +log " (a) FEC negotiated : ${a_ok} (bytes=${A_BYTES} filter='${A_FILTER}' — expect non-empty fec)" +log " (b) plain accepted : ${b_ok} (bytes=${B_BYTES} filter='${B_FILTER}' — expect empty, NOT rejected)" +log " (c) hard reject : ${c_ok} (bytes=${C_BYTES} reject=${C_REJECT} — expect SRT_REJ_FILTER, 0 bytes)" +log " (d) no-filter adopts : ${d_adopts} (bytes=${D_BYTES} filter='${D_FILTER}' — informational, not gated)" +log " result: ${RESULT_JSON}" +log "===========================================================" + +if [[ "$KEEP_LOGS" -eq 0 && "$pass" == true ]]; then + rm -f "${RESULTS_DIR}"/case-*-sink.log "${RESULTS_DIR}"/case-*-call.log \ + "${RESULTS_DIR}"/case-*.json "$PAYGEN" +fi + +if [[ "$pass" == true ]]; then log "PASS"; exit 0; else log "FAIL"; exit 1; fi diff --git a/tests/compat/scenarios/profile-validation-matrix.sh b/tests/compat/scenarios/profile-validation-matrix.sh new file mode 100755 index 0000000..d6696fb --- /dev/null +++ b/tests/compat/scenarios/profile-validation-matrix.sh @@ -0,0 +1,477 @@ +#!/usr/bin/env bash +# +# profile-validation-matrix.sh — A/B validation matrix for the SRT receive +# profiles: the 4 non-FEC profiles (Balanced / Low-Latency / Resilient / +# Classic) measured against the patched-libsrt baseline under production- +# realistic cross-link reorder stress, with an "equal" pass gate built on the +# srt-sink TS-continuity metrics (Task 5), NOT on a bytes-only proxy. +# +# WHY this exists: the receiver cutover replaces the unconditional BELABOX +# libsrt patch (decay-off + periodic-NAK-off, baked in at compile time) with the +# opt-in SRTO_REORDERFREEZE option (reorderfreeze-1.5.5 branch) plus standard +# SRTO_NAKREPORT / SRTO_LOSSMAXTTL flags. This script proves each non-FEC profile +# recipe is statistically EQUAL to the patch baseline before that cutover lands. +# +# It is the production-realistic successor to the ADR-002 A/B/C study: same +# instrument (scenarios/reorder-stress.sh), but parameterised across bitrate, +# receive-latency and LOSSMAXTTL, run paired/alternating with a fixed netem seed, +# and judged on transport-stream continuity rather than delivered byte count. +# +# THE TWO libsrt ARTIFACTS (swapped under srt-sink, never the system libsrt): +# baseline CERALIVE/srt @ 52057f6 — the 6-line unconditional BELABOX merge +# (libsrt.so.1.5.4). Behaviours (b) decay-off + (c) NAK-off are baked +# in; no socket flag is needed (and SRTO_SRTLAPATCHES is NOT defined +# on this fork — see ADR-002 §1). +# freeze CERALIVE/srt @ reorderfreeze-1.5.5 — vanilla 1.5.5 + opt-in +# SRTO_REORDERFREEZE (libsrt.so.1.5.5). The profile recipes drive it +# via srt-sink's --reorderfreeze / --nakreport / --lossmaxttl. +# Both are provided as loader prefixes (BASELINE_LIBSRT / FREEZE_LIBSRT). If a +# prefix is absent the script builds it via lib/build-libsrt-matrix.sh from +# SRT_REPO_URL @ {BASELINE_REF,FREEZE_REF} (mapping its patched slot -> baseline, +# its vanilla slot -> freeze). FEC SCOPE: this gate covers the 4 NON-FEC +# profiles only; the Low-Latency+FEC row is gated separately. +# +# THE 4 NON-FEC PROFILE RECIPES (device latency differs; receiver recipe shown): +# Balanced freeze+NAK REORDERFREEZE=1 NAKREPORT=1 @ 1500ms +# Low-Latency freeze+NAK REORDERFREEZE=1 NAKREPORT=1 @ 250ms +# Resilient freeze+NAK REORDERFREEZE=1 NAKREPORT=1 @ 3500ms +# Classic freeze+NAK-off REORDERFREEZE=1 NAKREPORT=0 @ 800ms +# (control) stock-decay REORDERFREEZE=0 NAKREPORT=1 — falsifiability arm +# +# THE "EQUAL" GATE (profile vs its paired baseline, per cell): +# 1. disconnects == 0 (both arms) 4. median goodput >= 99% baseline +# 2. ts_sync_errors == 0 (profile) 5. p95 late-drop (pkt_rcv_drop) <= baseline +# 3. ts_cc_errors <= baseline (median) 6. wire-amplification <= 1.10x baseline +# +# REGRESSION-VALIDATION CELL: Balanced (freeze, NAK-on) vs baseline at receive +# latencies {500,1500,3500}ms — the dedicated NAK-on comparison, TS-continuity as +# the signal. LOSSMAXTTL SWEEP: Balanced vs baseline at {30,200,1000} (a separate +# axis to pick the cap). The blocking verdict = all 4 non-FEC profiles PASS AND +# the regression cell PASSes; the sweep + control are informational. +# +# PRIVILEGE: needs CAP_NET_ADMIN (real root / passwordless sudo, OR mapped-root +# in a user+net namespace), gated via lib/netem.sh `require`; without it the +# script prints SKIP-PRIVILEGED and exits 77, creating no state. +# CI / local: sudo tests/compat/scenarios/profile-validation-matrix.sh +# smoke: sudo tests/compat/scenarios/profile-validation-matrix.sh --smoke +# +# Usage: +# profile-validation-matrix.sh [--build-dir DIR] [--reps N] [--duration SEC] +# [--smoke] [--keep-logs] [-h] +# --reps N paired reps per arm per cell (default 10). +# --duration SEC per-phase seconds for reorder-stress (default 58 => ~121s/rep). +# --smoke fast self-test: --reps 2 --duration 12 (~29s/rep). +# +# Evidence table -> test-results/srt-receive-profiles/task-6-srt-receive-profiles.json +# Per-run artifacts -> tests/compat/results/profile-validation-matrix/ (gitignored). +# Rule D: writes nothing above the srtla repo root. +# +set -uo pipefail + +SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" >/dev/null 2>&1 && pwd)" +REPO_ROOT="$(cd -- "${SCRIPT_DIR}/../../.." >/dev/null 2>&1 && pwd)" +RESULTS_DIR="${SCRIPT_DIR}/../results/profile-validation-matrix" +REORDER="${SCRIPT_DIR}/reorder-stress.sh" +NETEM_LIB="${SCRIPT_DIR}/../lib/netem.sh" +BUILD_MATRIX="${SCRIPT_DIR}/../lib/build-libsrt-matrix.sh" +LIBSRT_MATRIX="${REPO_ROOT}/test-results/libsrt-matrix" +EVIDENCE_DIR="${REPO_ROOT}/test-results/srt-receive-profiles" +EVIDENCE_JSON="${EVIDENCE_DIR}/task-6-srt-receive-profiles.json" +REORDER_RESULT="${SCRIPT_DIR}/../results/reorder-stress/result.json" + +log() { printf '%s\n' "$*" >&2; } +die() { printf 'profile-validation-matrix: %s\n' "$*" >&2; exit 2; } + +# --------------------------------------------------------------------------- # +# CLI / env # +# --------------------------------------------------------------------------- # +BUILD_DIR="${SRTLA_BUILD_DIR:-}" +REPS="${REPS:-10}" +PHASE_SEC="${PHASE_SEC:-58}" +BITRATE_KBPS="${BITRATE_KBPS:-8000}" +NETEM_SEED_BASE="${NETEM_SEED_BASE:-1000}" +KEEP_LOGS=0 +BASELINE_LIBSRT="${BASELINE_LIBSRT:-${LIBSRT_MATRIX}/install/patched/lib}" +FREEZE_LIBSRT="${FREEZE_LIBSRT:-${LIBSRT_MATRIX}/install/freeze/lib}" +SRT_REPO_URL="${SRT_REPO_URL:-https://github.com/CERALIVE/srt}" +BASELINE_REF="${BASELINE_REF:-52057f6846c66d4ecf5d47c9c0a2cecd281d77d6}" +FREEZE_REF="${FREEZE_REF:-reorderfreeze-1.5.5}" + +while [[ $# -gt 0 ]]; do + case "$1" in + --build-dir) BUILD_DIR="${2:?--build-dir needs a value}"; shift 2 ;; + --reps) REPS="${2:?--reps needs a value}"; shift 2 ;; + --duration) PHASE_SEC="${2:?--duration needs a value}"; shift 2 ;; + --smoke) REPS=2; PHASE_SEC=12; shift ;; + --keep-logs) KEEP_LOGS=1; shift ;; + -h|--help) sed -n '2,60p' "${BASH_SOURCE[0]}" | sed 's/^# \{0,1\}//'; exit 0 ;; + *) die "unknown argument '$1' (try --help)" ;; + esac +done +[[ "$REPS" =~ ^[0-9]+$ && "$REPS" -ge 1 ]] || die "--reps must be a positive integer" +[[ "$PHASE_SEC" =~ ^[0-9]+$ && "$PHASE_SEC" -ge 1 ]] || die "--duration must be a positive integer" + +for tool in ffmpeg jq python3 ip tc; do + command -v "$tool" >/dev/null 2>&1 || die "required tool '$tool' not found in PATH" +done +[[ -x "$REORDER" ]] || die "reorder-stress.sh not found/executable at $REORDER" + +# Capability gate via netem.sh's CLI `require` (77 == SKIP-PRIVILEGED). Run as a +# subprocess so we do not entangle netem.sh's EXIT trap with this driver. +if ! bash "$NETEM_LIB" require >/dev/null 2>&1; then + log "SKIP profile-validation-matrix: netem unavailable (need CAP_NET_ADMIN: root, sudo, or mapped-root userns)" + mkdir -p "$EVIDENCE_DIR" + jq -n --arg ts "$(date -u +%Y-%m-%dT%H:%M:%SZ)" \ + '{task:"task-6-srt-receive-profiles", skipped:true, + reason:"no CAP_NET_ADMIN / netem", timestamp:$ts}' > "$EVIDENCE_JSON" + exit 77 +fi + +# --------------------------------------------------------------------------- # +# Resolve the srtla build dir (srtla_rec / srtla_send / srt-sink). # +# --------------------------------------------------------------------------- # +resolve_build_dir() { + local candidates=() d + [[ -n "$BUILD_DIR" ]] && candidates+=("$BUILD_DIR") + candidates+=("${REPO_ROOT}/build" "/tmp/srtla-build") + for d in "${candidates[@]}"; do + if [[ -x "${d}/srtla_rec" && -x "${d}/srtla_send" \ + && -x "${d}/tests/compat/srt-sink/srt-sink" ]]; then printf '%s' "$d"; return 0; fi + done + return 1 +} +BUILD_DIR="$(resolve_build_dir)" || die \ + "no usable build dir (need srtla_rec, srtla_send, srt-sink). Build with: + cmake -B build -DBUILD_COMPAT_TESTS=ON && cmake --build build -j" + +# --------------------------------------------------------------------------- # +# Provision the two libsrt loader prefixes. If a prefix already holds a # +# libsrt.so it is used as-is (operator pre-built); otherwise both are built # +# from SRT_REPO_URL via build-libsrt-matrix.sh (its patched slot is our # +# baseline, its vanilla slot is our freeze build). # +# --------------------------------------------------------------------------- # +has_libsrt() { compgen -G "${1}/libsrt.so.*" >/dev/null 2>&1; } + +provision_libsrt() { + if has_libsrt "$BASELINE_LIBSRT" && has_libsrt "$FREEZE_LIBSRT"; then + log "==> libsrt: using pre-built prefixes" + return 0 + fi + [[ -x "$BUILD_MATRIX" ]] || die "missing libsrt prefixes and no builder at $BUILD_MATRIX" + log "==> libsrt: building baseline ($BASELINE_REF) + freeze ($FREEZE_REF) from $SRT_REPO_URL" + bash "$BUILD_MATRIX" \ + --patched-url "$SRT_REPO_URL" --patched-ref "$BASELINE_REF" \ + --vanilla-url "$SRT_REPO_URL" --vanilla-ref "$FREEZE_REF" >&2 \ + || die "build-libsrt-matrix.sh failed" + BASELINE_LIBSRT="${LIBSRT_MATRIX}/install/patched/lib" + FREEZE_LIBSRT="${LIBSRT_MATRIX}/install/vanilla/lib" + has_libsrt "$BASELINE_LIBSRT" || die "baseline libsrt missing after build" + has_libsrt "$FREEZE_LIBSRT" || die "freeze libsrt missing after build" +} +provision_libsrt +BASELINE_LIBSRT="$(cd -- "$BASELINE_LIBSRT" && pwd -P)" +FREEZE_LIBSRT="$(cd -- "$FREEZE_LIBSRT" && pwd -P)" + +log "================ profile-validation-matrix ================" +log " build dir : ${BUILD_DIR}" +log " baseline lib: ${BASELINE_LIBSRT}" +log " freeze lib : ${FREEZE_LIBSRT}" +log " reps=${REPS} phase=${PHASE_SEC}s (~$((5 + 2*PHASE_SEC))s/rep) bitrate=${BITRATE_KBPS}k seed_base=${NETEM_SEED_BASE}" +log "===========================================================" + +rm -rf "$RESULTS_DIR"; mkdir -p "$RESULTS_DIR" + +# --------------------------------------------------------------------------- # +# One reorder-stress rep -> copy its result.json to . # +# Args: out lib latency bitrate lossmaxttl nakreport reorderfreeze label seed # +# Empty lossmaxttl/nakreport/reorderfreeze => that knob is left at libsrt # +# default (the baseline arm passes all three empty: the patch is unconditional).# +# --------------------------------------------------------------------------- # +run_rep() { + local out="$1" lib="$2" lat="$3" br="$4" ttl="$5" nak="$6" frz="$7" label="$8" seed="$9" + local -a env_kv=( + "SRTLA_BUILD_DIR=${BUILD_DIR}" "SINK_LD_LIBRARY_PATH=${lib}" + "RX_LATENCY_MS=${lat}" "BITRATE_KBPS=${br}" "PROFILE_LABEL=${label}" + "NETEM_SEED=${seed}" + ) + [[ -n "$ttl" ]] && env_kv+=("LOSSMAXTTL=${ttl}") + [[ -n "$nak" ]] && env_kv+=("NAKREPORT=${nak}") + [[ -n "$frz" ]] && env_kv+=("REORDERFREEZE=${frz}") + rm -f "$REORDER_RESULT" + env "${env_kv[@]}" bash "$REORDER" --duration "$PHASE_SEC" >/dev/null 2>&1 || true + if [[ -f "$REORDER_RESULT" ]]; then cp "$REORDER_RESULT" "$out"; else printf '{}\n' > "$out"; fi +} + +# A paired cell: baseline arm + profile arm, ALTERNATED per rep with a shared +# per-rep seed so both arms meet the same netem reorder draw (paired comparison). +# Args: cell_dir latency lossmaxttl prof_nak prof_frz prof_label +run_cell() { + local dir="$1" lat="$2" ttl="$3" pnak="$4" pfrz="$5" plabel="$6" + mkdir -p "${dir}/baseline" "${dir}/profile" + local r seed + for (( r=1; r<=REPS; r++ )); do + seed=$(( NETEM_SEED_BASE + r )) + log " [$(basename "$(dirname "$dir")")/$(basename "$dir")] rep ${r}/${REPS} (seed ${seed})" + run_rep "${dir}/baseline/rep-${r}.json" "$BASELINE_LIBSRT" "$lat" "$BITRATE_KBPS" "" "" "" "baseline-patched" "$seed" + run_rep "${dir}/profile/rep-${r}.json" "$FREEZE_LIBSRT" "$lat" "$BITRATE_KBPS" "$ttl" "$pnak" "$pfrz" "$plabel" "$seed" + done +} + +# --------------------------------------------------------------------------- # +# Main matrix — the 4 non-FEC profiles, each vs the paired baseline. # +# --------------------------------------------------------------------------- # +log "==> main matrix: 4 non-FEC profiles vs baseline" +run_cell "${RESULTS_DIR}/main/balanced" 1500 30 1 1 "balanced-freeze+nak" +run_cell "${RESULTS_DIR}/main/lowlatency" 250 30 1 1 "lowlatency-freeze+nak" +run_cell "${RESULTS_DIR}/main/resilient" 3500 30 1 1 "resilient-freeze+nak" +run_cell "${RESULTS_DIR}/main/classic" 800 30 0 1 "classic-freeze+nak-off" + +# --------------------------------------------------------------------------- # +# Regression-validation cell — Balanced (freeze, NAK-on) vs baseline at the # +# three receive latencies, the dedicated NAK-on comparison. # +# --------------------------------------------------------------------------- # +log "==> regression-validation cell: balanced freeze+NAK vs baseline @ {500,1500,3500}ms" +for lat in 500 1500 3500; do + run_cell "${RESULTS_DIR}/regression/lat-${lat}" "$lat" 30 1 1 "balanced-freeze+nak" +done + +# --------------------------------------------------------------------------- # +# LOSSMAXTTL sweep — a separate axis to pick the cap (Balanced @ 1500ms). # +# --------------------------------------------------------------------------- # +log "==> LOSSMAXTTL sweep: {30,200,1000} @ 1500ms" +for ttl in 30 200 1000; do + run_cell "${RESULTS_DIR}/lossmaxttl/ttl-${ttl}" 1500 "$ttl" 1 1 "balanced-freeze+nak" +done + +# --------------------------------------------------------------------------- # +# Control — stock-decay (REORDERFREEZE=0) vs baseline @ 1500ms. Falsifiability: # +# the gate must be able to SEE the freeze matter (this arm may legitimately # +# diverge); it is reported, not part of the blocking verdict. # +# --------------------------------------------------------------------------- # +log "==> control (falsifiability): stock-decay+NAK vs baseline @ 1500ms" +run_cell "${RESULTS_DIR}/control/stock-decay" 1500 30 1 0 "stock-decay+nak" + +# --------------------------------------------------------------------------- # +# Aggregate every cell, apply the equal gate, write the evidence table. # +# --------------------------------------------------------------------------- # +mkdir -p "$EVIDENCE_DIR" +log "==> aggregating + applying the equal gate -> ${EVIDENCE_JSON}" + +python3 - "$RESULTS_DIR" "$EVIDENCE_JSON" "$BASELINE_LIBSRT" "$FREEZE_LIBSRT" \ + "$REPS" "$PHASE_SEC" "$BITRATE_KBPS" "$NETEM_SEED_BASE" "$LIBSRT_MATRIX" <<'PY' +import json, math, os, sys, glob, datetime + +(results_dir, evidence_json, baseline_lib, freeze_lib, + reps, phase_sec, bitrate, seed_base, libsrt_matrix) = sys.argv[1:10] +reps = int(reps); phase_sec = int(phase_sec); bitrate = int(bitrate) + +GOODPUT_FLOOR = 0.99 # clause 4: median goodput >= 99% baseline +WIRE_AMP_CEIL = 1.10 # clause 6: wire-amp <= 1.10x baseline + +def load_arm(arm_dir): + rows = [] + for f in sorted(glob.glob(os.path.join(arm_dir, "rep-*.json"))): + try: + with open(f) as fh: + d = json.load(fh) + except (OSError, ValueError): + d = {} + m = d.get("metrics", {}) or {} + s = d.get("sink", {}) or {} + rows.append({ + "goodput_bps": float(m.get("goodput_bps", 0) or 0), + "wire_amp": float(m.get("wire_amp", 0) or 0), + "ts_sync": int(m.get("ts_sync_errors", -1)), + "ts_cc": int(m.get("ts_cc_errors", -1)), + "pkt_drop": int(m.get("pkt_rcv_drop", 0) or 0), + "pkt_retrans": int(m.get("pkt_retrans", 0) or 0), + "bytes": int(s.get("bytes_received", 0) or 0), + "disc": int(s.get("disconnects", -1)), + }) + return rows + +def median(xs): + xs = sorted(xs) + n = len(xs) + if n == 0: + return 0.0 + mid = n // 2 + return float(xs[mid]) if n % 2 else (xs[mid - 1] + xs[mid]) / 2.0 + +def p95(xs): + xs = sorted(xs) + n = len(xs) + if n == 0: + return 0.0 + idx = min(n - 1, max(0, math.ceil(0.95 * n) - 1)) + return float(xs[idx]) + +def agg(rows): + if not rows: + return {"n": 0, "goodput_median": 0.0, "wire_amp_median": 0.0, + "ts_cc_median": 0.0, "ts_cc_max": 0, "ts_sync_max": -1, + "pkt_drop_p95": 0.0, "disc_max": -1, "pkt_retrans_median": 0.0, + "bytes_median": 0.0} + return { + "n": len(rows), + "goodput_median": median([r["goodput_bps"] for r in rows]), + "wire_amp_median": median([r["wire_amp"] for r in rows]), + "ts_cc_median": median([r["ts_cc"] for r in rows]), + "ts_cc_max": max(r["ts_cc"] for r in rows), + "ts_sync_max": max(r["ts_sync"] for r in rows), + "pkt_drop_p95": p95([r["pkt_drop"] for r in rows]), + "pkt_retrans_median": median([r["pkt_retrans"] for r in rows]), + "disc_max": max(r["disc"] for r in rows), + "bytes_median": median([r["bytes"] for r in rows]), + } + +def equal_gate(prof, base): + c1 = prof["disc_max"] == 0 and base["disc_max"] == 0 + c2 = prof["ts_sync_max"] == 0 + c3 = prof["ts_cc_median"] <= base["ts_cc_median"] + c4 = prof["goodput_median"] >= GOODPUT_FLOOR * base["goodput_median"] + c5 = prof["pkt_drop_p95"] <= base["pkt_drop_p95"] + c6 = (prof["wire_amp_median"] <= WIRE_AMP_CEIL * base["wire_amp_median"]) \ + if base["wire_amp_median"] > 0 else (prof["wire_amp_median"] == 0) + clauses = { + "disconnects_zero": c1, "ts_sync_zero": c2, "ts_cc_le_baseline": c3, + "goodput_ge_99pct": c4, "p95_late_drop_le_baseline": c5, + "wire_amp_le_110pct": c6, + } + return all(clauses.values()), clauses + +def cell(cell_dir): + base = agg(load_arm(os.path.join(cell_dir, "baseline"))) + prof = agg(load_arm(os.path.join(cell_dir, "profile"))) + passed, clauses = equal_gate(prof, base) + return {"baseline": base, "profile": prof, "pass": passed, "clauses": clauses} + +def cells_in(group): + base = os.path.join(results_dir, group) + if not os.path.isdir(base): + return {} + out = {} + for name in sorted(os.listdir(base)): + d = os.path.join(base, name) + if os.path.isdir(d): + out[name] = cell(d) + return out + +main = cells_in("main") +regression = cells_in("regression") +lossmaxttl = cells_in("lossmaxttl") +control = cells_in("control") + +main_pass = bool(main) and all(c["pass"] for c in main.values()) +regression_pass = bool(regression) and all(c["pass"] for c in regression.values()) +blocking_pass = main_pass and regression_pass + +# LOSSMAXTTL recommendation: smallest cap whose cell still passes the equal gate. +def ttl_val(name): + try: + return int(name.split("-", 1)[1]) + except (IndexError, ValueError): + return 1 << 30 +passing_ttls = sorted(int(n.split("-", 1)[1]) for n, c in lossmaxttl.items() + if c["pass"] and n.startswith("ttl-")) +recommended_lossmaxttl = passing_ttls[0] if passing_ttls else None + +# libsrt identity from the build manifest, when present. +manifest = {} +mpath = os.path.join(libsrt_matrix, "manifest.txt") +if os.path.isfile(mpath): + section = None + with open(mpath) as fh: + for line in fh: + line = line.strip() + if line.startswith("[") and line.endswith("]"): + section = line[1:-1]; manifest[section] = {} + elif section and "=" in line and not line.startswith("#"): + k, v = (x.strip() for x in line.split("=", 1)) + manifest[section][k] = v + +doc = { + "task": "task-6-srt-receive-profiles", + "title": "A/B profile validation matrix — 4 non-FEC profiles vs patched baseline", + "timestamp": datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), + "fec_scope": "non-FEC only (Balanced/Low-Latency/Resilient/Classic); " + "Low-Latency+FEC row gated separately", + "methodology": { + "instrument": "scenarios/reorder-stress.sh (cross-link reorder: 50/150ms " + "asymmetric delay + netem reorder 25% 50% phase)", + "reps_per_arm": reps, + "phase_sec": phase_sec, + "approx_sec_per_rep": 5 + 2 * phase_sec, + "bitrate_kbps": bitrate, + "pairing": "alternating baseline/profile per rep, shared per-rep netem seed", + "seed_base": int(seed_base), + "signal": "srt-sink TS-continuity + SRT loss/retrans (Task 5), NOT bytes-only", + "equal_gate": { + "disconnects": "== 0 (both arms)", + "ts_sync_errors": "== 0 (profile)", + "ts_cc_errors": "<= baseline (median)", + "goodput": ">= 99% baseline (median)", + "p95_late_drop": "<= baseline (pkt_rcv_drop p95)", + "wire_amplification": "<= 1.10x baseline (median)", + }, + }, + "libsrt": { + "baseline": {"role": "patched (unconditional BELABOX merge)", + "loader": baseline_lib, "manifest": manifest.get("patched", {})}, + "freeze": {"role": "reorderfreeze-1.5.5 (opt-in SRTO_REORDERFREEZE)", + "loader": freeze_lib, "manifest": manifest.get("vanilla", {})}, + }, + "profiles": main, + "regression_validation_cell": regression, + "lossmaxttl_sweep": lossmaxttl, + "control_stock_decay": control, + "verdict": { + "non_fec_profiles_pass": main_pass, + "regression_cell_pass": regression_pass, + "blocking_pass": blocking_pass, + "recommended_lossmaxttl": recommended_lossmaxttl, + }, +} + +with open(evidence_json, "w") as fh: + json.dump(doc, fh, indent=2) + fh.write("\n") + +def line(name, c): + g = c["profile"]["goodput_median"]; b = c["baseline"]["goodput_median"] + ratio = (g / b) if b else 0.0 + sys.stderr.write( + " %-22s %s goodput=%.0f/%.0f (%.1f%%) wire_amp=%.3f/%.3f " + "ts_cc=%.0f/%.0f p95_drop=%.0f/%.0f disc=%d/%d\n" % ( + name, "PASS" if c["pass"] else "FAIL", g, b, 100 * ratio, + c["profile"]["wire_amp_median"], c["baseline"]["wire_amp_median"], + c["profile"]["ts_cc_median"], c["baseline"]["ts_cc_median"], + c["profile"]["pkt_drop_p95"], c["baseline"]["pkt_drop_p95"], + c["profile"]["disc_max"], c["baseline"]["disc_max"])) + +sys.stderr.write("\n================ equal-gate results (profile/baseline) ================\n") +sys.stderr.write("-- main (4 non-FEC profiles) --\n") +for n, c in main.items(): line(n, c) +sys.stderr.write("-- regression-validation cell (balanced NAK-on) --\n") +for n, c in regression.items(): line(n, c) +sys.stderr.write("-- LOSSMAXTTL sweep --\n") +for n, c in lossmaxttl.items(): line(n, c) +sys.stderr.write("-- control (falsifiability; informational) --\n") +for n, c in control.items(): line(n, c) +sys.stderr.write("recommended LOSSMAXTTL cap: %s\n" % recommended_lossmaxttl) +sys.stderr.write("non-FEC profiles PASS: %s | regression cell PASS: %s\n" + % (main_pass, regression_pass)) +sys.stderr.write("BLOCKING VERDICT: %s\n" % ("PASS" if blocking_pass else "FAIL")) +sys.stderr.write("======================================================================\n") + +sys.exit(0 if blocking_pass else 1) +PY +PY_RC=$? + +if [[ "$KEEP_LOGS" -eq 0 && "$PY_RC" -eq 0 ]]; then + rm -rf "$RESULTS_DIR" +fi + +log "" +log "evidence: ${EVIDENCE_JSON}" +if [[ "$PY_RC" -eq 0 ]]; then log "PASS"; exit 0; else log "FAIL"; exit 1; fi diff --git a/tests/compat/scenarios/reorder-stress.sh b/tests/compat/scenarios/reorder-stress.sh index 69075de..2c5f47a 100755 --- a/tests/compat/scenarios/reorder-stress.sh +++ b/tests/compat/scenarios/reorder-stress.sh @@ -47,8 +47,23 @@ # # Parameterisation (env vars; empty defaults = today's behaviour, system libsrt): # SINK_LD_LIBRARY_PATH loader path so srt-sink picks a specific libsrt build -# (e.g. test-results/libsrt-matrix/install/vanilla/lib). +# (e.g. test-results/libsrt-matrix/install/freeze/lib). # SINK_EXTRA_ARGS extra srt-sink flags, e.g. "--nakreport 0 --lossmaxttl 30". +# BITRATE_KBPS media bitrate fed to ffmpeg (default 700; A/B uses 4000/8000/12000). +# RX_LATENCY_MS SRT receive-latency window on srt-sink AND the ffmpeg caller +# (default 1200; A/B sweeps 250/500/800/1500/3500). +# NAKREPORT 0|1 -> appends --nakreport (profile recipe: periodic NAK on/off). +# LOSSMAXTTL N -> appends --lossmaxttl (reorder-tolerance cap; A/B sweeps 30/200/1000). +# REORDERFREEZE 0|1 -> appends --reorderfreeze (decay-freeze; needs the freeze libsrt). +# PROFILE_LABEL free-form tag echoed into result.json so an A/B driver can attribute the run. +# NETEM_SEED fixed seed for the phase-ii reorder discipline (reproducible A/B runs; +# silently ignored on iproute2 too old for `netem ... seed`). +# +# The recipe-shorthand maps to the 4 non-FEC receive profiles like so (see the +# A/B driver scenarios/profile-validation-matrix.sh): freeze+NAK (Balanced / +# Low-Latency / Resilient) = REORDERFREEZE=1 NAKREPORT=1; freeze+NAK-off +# (Classic) = REORDERFREEZE=1 NAKREPORT=0; stock-decay+NAK (control) = +# REORDERFREEZE=0 NAKREPORT=1; baseline = the patched libsrt with no recipe flags. # # Machine-parseable summary line (always emitted on a completed run): # reorder-stress: bytes_received= disconnects= duration=s libsrt= ... @@ -90,6 +105,13 @@ KEEP_LOGS=0 PHASE_SEC=14 SINK_LD_LIBRARY_PATH="${SINK_LD_LIBRARY_PATH:-}" SINK_EXTRA_ARGS="${SINK_EXTRA_ARGS:-}" +BITRATE_KBPS="${BITRATE_KBPS:-700}" +RX_LATENCY_MS="${RX_LATENCY_MS:-1200}" +NAKREPORT="${NAKREPORT:-}" +LOSSMAXTTL="${LOSSMAXTTL:-}" +REORDERFREEZE="${REORDERFREEZE:-}" +PROFILE_LABEL="${PROFILE_LABEL:-default}" +NETEM_SEED="${NETEM_SEED:-}" while [[ $# -gt 0 ]]; do case "$1" in @@ -101,6 +123,12 @@ while [[ $# -gt 0 ]]; do esac done [[ "$PHASE_SEC" =~ ^[0-9]+$ && "$PHASE_SEC" -ge 1 ]] || die "--duration must be a positive integer" +[[ "$BITRATE_KBPS" =~ ^[0-9]+$ && "$BITRATE_KBPS" -ge 1 ]] || die "BITRATE_KBPS must be a positive integer" +[[ "$RX_LATENCY_MS" =~ ^[0-9]+$ && "$RX_LATENCY_MS" -ge 1 ]] || die "RX_LATENCY_MS must be a positive integer" +[[ -z "$NAKREPORT" || "$NAKREPORT" =~ ^[01]$ ]] || die "NAKREPORT must be 0 or 1" +[[ -z "$REORDERFREEZE" || "$REORDERFREEZE" =~ ^[01]$ ]] || die "REORDERFREEZE must be 0 or 1" +[[ -z "$LOSSMAXTTL" || "$LOSSMAXTTL" =~ ^[0-9]+$ ]] || die "LOSSMAXTTL must be a non-negative integer" +[[ -z "$NETEM_SEED" || "$NETEM_SEED" =~ ^[0-9]+$ ]] || die "NETEM_SEED must be a non-negative integer" for tool in ffmpeg jq; do command -v "$tool" >/dev/null 2>&1 || die "required tool '$tool' not found in PATH" @@ -169,8 +197,9 @@ LOCAL_SRT_PORT=6401 # SRT receive window must exceed the worst cross-link skew (150ms + reorder) # so the bonded stream rides reordering without an end-to-end disconnect; a real -# SRTLA caller (cerastream/Moblin) is tuned the same way. 1.2s >> 150ms+20ms. -SRT_LATENCY_MS=1200 +# SRTLA caller (cerastream/Moblin) is tuned the same way. Default 1.2s >> 150ms+20ms; +# the A/B driver overrides it via RX_LATENCY_MS to sweep the receive-latency axis. +SRT_LATENCY_MS="$RX_LATENCY_MS" ESTABLISH_SEC=5 MEAS_SEC=$(( ESTABLISH_SEC + 2 * PHASE_SEC )) @@ -225,6 +254,17 @@ qdisc_sent_pkts() { # dev handle(e.g. "10:") -> integer printf '%s' "${n:-0}" } +# Total egress wire BYTES the qdisc has Sent ($2 of its "Sent" line); +# read on the root "1:" prio it is the bonded forward-wire volume that the +# wire-amplification ratio (wire_bytes / bytes_received) is built from. +qdisc_sent_bytes() { # dev handle(e.g. "1:") -> integer + local n + n="$(tc -s qdisc show dev "$1" 2>/dev/null | awk -v h="$2" ' + $1=="qdisc" && $3==h {f=1; next} + f && $1=="Sent" {print $2; exit}')" + printf '%s' "${n:-0}" +} + setup_topology() { ip link set lo up 2>/dev/null || true # host-side loopback (needed under a fresh netns / unshare) @@ -271,6 +311,11 @@ SINK_RESOLVED="$(LD_LIBRARY_PATH="$SINK_LD_LIBRARY_PATH" ldd "$SRT_SINK" 2>/dev/ log " srt-sink libsrt resolves to: ${SINK_RESOLVED:-}" # shellcheck disable=SC2206 # deliberate word-split of caller-supplied sink flags SINK_ARGS=($SINK_EXTRA_ARGS) +# Recipe knobs append explicit srt-sink flags on top of SINK_EXTRA_ARGS so an A/B +# driver can express a profile (freeze+NAK, freeze+NAK-off, ...) without string-building. +[[ -n "$NAKREPORT" ]] && SINK_ARGS+=(--nakreport "$NAKREPORT") +[[ -n "$LOSSMAXTTL" ]] && SINK_ARGS+=(--lossmaxttl "$LOSSMAXTTL") +[[ -n "$REORDERFREEZE" ]] && SINK_ARGS+=(--reorderfreeze "$REORDERFREEZE") ip netns exec "$NS" env LD_LIBRARY_PATH="$SINK_LD_LIBRARY_PATH" \ "$SRT_SINK" --port "$SINK_PORT" --host 127.0.0.1 --result "$SINK_JSON" \ --latency "$SRT_LATENCY_MS" --duration $(( MEAS_SEC + 25 )) \ @@ -297,8 +342,12 @@ sleep 0.6 # latency/timeout options are in MICROSECONDS). SRT_LATENCY_US=$(( SRT_LATENCY_MS * 1000 )) SRT_OPTS="mode=caller&transtype=live&latency=${SRT_LATENCY_US}&peerlatency=${SRT_LATENCY_US}&sndbuf=24000000&timeout=30000000" +# Default (700k) keeps today's 320x240@25; HD-rate A/B runs need a larger, busier +# source so the mpeg2video encoder actually emits the requested multi-Mbps target. +FF_SIZE=320x240; FF_RATE=25 +if [[ "$BITRATE_KBPS" -ge 4000 ]]; then FF_SIZE=1280x720; FF_RATE=30; fi ffmpeg -hide_banner -loglevel warning -re \ - -f lavfi -i testsrc2=size=320x240:rate=25 -c:v mpeg2video -b:v 700k -f mpegts \ + -f lavfi -i "testsrc2=size=${FF_SIZE}:rate=${FF_RATE}" -c:v mpeg2video -b:v "${BITRATE_KBPS}k" -f mpegts \ "srt://127.0.0.1:${LOCAL_SRT_PORT}?${SRT_OPTS}" \ >"$FF_LOG" 2>&1 & FF_PID=$!; track "$FF_PID" @@ -321,10 +370,20 @@ sleep "$PHASE_SEC" # ----------------------------------------------------------------------------- # # Phase ii — layer an explicit reorder discipline onto the fast link. # # ----------------------------------------------------------------------------- # -log "==> phase ii: netem reorder 25% 50% delay ${REORDER_DELAY_MS}ms on link A for ${PHASE_SEC}s" -tc qdisc change dev "$HOSTIF" parent 1:1 handle 10: \ - netem delay "${REORDER_DELAY_MS}ms" reorder 25% 50% \ - || die "could not apply reorder discipline" +log "==> phase ii: netem reorder 25% 50% delay ${REORDER_DELAY_MS}ms on link A for ${PHASE_SEC}s${NETEM_SEED:+ (seed ${NETEM_SEED})}" +# Fixed `seed` makes the random reorder/gap draw reproducible across paired A/B +# runs; older iproute2 lacks the keyword, so fall back to the seedless form. +if [[ -n "$NETEM_SEED" ]]; then + tc qdisc change dev "$HOSTIF" parent 1:1 handle 10: \ + netem delay "${REORDER_DELAY_MS}ms" reorder 25% 50% seed "$NETEM_SEED" 2>/dev/null \ + || tc qdisc change dev "$HOSTIF" parent 1:1 handle 10: \ + netem delay "${REORDER_DELAY_MS}ms" reorder 25% 50% \ + || die "could not apply reorder discipline" +else + tc qdisc change dev "$HOSTIF" parent 1:1 handle 10: \ + netem delay "${REORDER_DELAY_MS}ms" reorder 25% 50% \ + || die "could not apply reorder discipline" +fi reorder_configured=false { printf '=== after applying reorder (phase ii start) ===\n'; tc -s qdisc show dev "$HOSTIF"; } >>"$TC_LOG" 2>&1 @@ -334,11 +393,13 @@ reorder_p0="$(qdisc_sent_pkts "$HOSTIF" "10:")" # fast-link pkts before phase sleep "$PHASE_SEC" reorder_p1="$(qdisc_sent_pkts "$HOSTIF" "10:")" # ... after band_b_pkts="$(qdisc_sent_pkts "$HOSTIF" "20:")" # slow-link total (asymmetry proof) +wire_bytes="$(qdisc_sent_bytes "$HOSTIF" "1:")" # total bonded forward-wire egress (root prio) { printf '=== after phase ii (phase ii end) ===\n'; tc -s qdisc show dev "$HOSTIF"; } >>"$TC_LOG" 2>&1 [[ "$reorder_p0" =~ ^[0-9]+$ ]] || reorder_p0=0 [[ "$reorder_p1" =~ ^[0-9]+$ ]] || reorder_p1=0 [[ "$band_b_pkts" =~ ^[0-9]+$ ]] || band_b_pkts=0 +[[ "$wire_bytes" =~ ^[0-9]+$ ]] || wire_bytes=0 reorder_pkts=$(( reorder_p1 - reorder_p0 )) [[ "$reorder_pkts" -lt 0 ]] && reorder_pkts=0 @@ -362,6 +423,23 @@ sdur_ms="$(jq -r '.duration_ms // 0' "$SINK_JSON" 2>/dev/null || echo 0)" [[ "$sdur_ms" =~ ^[0-9]+$ ]] || sdur_ms=0 duration_s=$(( sdur_ms / 1000 )) +# TS-continuity + SRT loss/retrans counters (srt-sink Task-5 additive keys). These +# are the A/B "equal" gate signal: a spurious-retransmit or late-drop profile shows +# up as ts_cc_errors / pkt_rcv_drop / pkt_retrans even when bytes_received matches. +ts_sync="$(jq -r '.ts_sync_errors // -1' "$SINK_JSON" 2>/dev/null || echo -1)" +ts_cc="$(jq -r '.ts_cc_errors // -1' "$SINK_JSON" 2>/dev/null || echo -1)" +ts_pkts="$(jq -r '.ts_packets // 0' "$SINK_JSON" 2>/dev/null || echo 0)" +pkt_loss="$(jq -r '.pkt_rcv_loss // 0' "$SINK_JSON" 2>/dev/null || echo 0)" +pkt_drop="$(jq -r '.pkt_rcv_drop // 0' "$SINK_JSON" 2>/dev/null || echo 0)" +pkt_retr="$(jq -r '.pkt_retrans // 0' "$SINK_JSON" 2>/dev/null || echo 0)" +for v in ts_sync ts_cc; do [[ "${!v}" =~ ^-?[0-9]+$ ]] || printf -v "$v" '%s' -1; done +for v in ts_pkts pkt_loss pkt_drop pkt_retr; do [[ "${!v}" =~ ^[0-9]+$ ]] || printf -v "$v" '%s' 0; done + +# Goodput (delivered B/s) and forward-wire amplification (egress / delivered) — +# the same two quantities ADR-002's pre-registered "equal" rule compares. +goodput=0; [[ "$duration_s" -gt 0 ]] && goodput=$(( bytes / duration_s )) +wire_amp="$(awk -v w="$wire_bytes" -v b="$bytes" 'BEGIN{ printf "%.4f", (b>0 ? w/b : 0) }')" + # libsrt version straight from the Task-4 srt-sink banner (proves which build ran). libsrt_ver="$(sed -n 's/^srt-sink: libsrt version \([0-9.]*\).*/\1/p' "$SINK_LOG" 2>/dev/null | head -1)" [[ -n "$libsrt_ver" ]] || libsrt_ver="unknown" @@ -392,16 +470,32 @@ jq -n \ --arg sink_libsrt_path "${SINK_RESOLVED:-system}" \ --arg sink_extra_args "${SINK_EXTRA_ARGS:-}" \ --argjson delay_a_ms "$DELAY_A_MS" --argjson delay_b_ms "$DELAY_B_MS" \ + --arg profile "$PROFILE_LABEL" \ + --argjson bitrate_kbps "$BITRATE_KBPS" --argjson rx_latency_ms "$RX_LATENCY_MS" \ + --arg nakreport "${NAKREPORT:-default}" --arg lossmaxttl "${LOSSMAXTTL:-default}" \ + --arg reorderfreeze "${REORDERFREEZE:-default}" --arg netem_seed "${NETEM_SEED:-none}" \ + --argjson ts_sync_errors "$ts_sync" --argjson ts_cc_errors "$ts_cc" \ + --argjson ts_packets "$ts_pkts" --argjson pkt_rcv_loss "$pkt_loss" \ + --argjson pkt_rcv_drop "$pkt_drop" --argjson pkt_retrans "$pkt_retr" \ + --argjson goodput_bps "$goodput" --argjson wire_bytes "$wire_bytes" \ + --argjson wire_amp "$wire_amp" \ --arg ts "$(date -u +%Y-%m-%dT%H:%M:%SZ)" \ '{ - scenario:"reorder-stress", pass:$pass, + scenario:"reorder-stress", pass:$pass, profile:$profile, criteria:{handshake_ok:$handshake_ok, bytes_ok:$bytes_ok, disconnects_ok:$disc_ok, duration_ok:$duration_ok, reorder_active:$reorder_active}, establish:{handshake:$handshake_ok, both_links_added:$both_links_added}, + config:{bitrate_kbps:$bitrate_kbps, rx_latency_ms:$rx_latency_ms, + nakreport:$nakreport, lossmaxttl:$lossmaxttl, + reorderfreeze:$reorderfreeze, netem_seed:$netem_seed}, sink:{bytes_received:$bytes, disconnects:$disconnects, duration_s:$duration_s, libsrt_version:$libsrt, libsrt_path:$sink_libsrt_path, extra_args:$sink_extra_args}, + metrics:{goodput_bps:$goodput_bps, wire_bytes:$wire_bytes, wire_amp:$wire_amp, + ts_packets:$ts_packets, ts_sync_errors:$ts_sync_errors, + ts_cc_errors:$ts_cc_errors, pkt_rcv_loss:$pkt_rcv_loss, + pkt_rcv_drop:$pkt_rcv_drop, pkt_retrans:$pkt_retrans}, reorder:{configured:$reorder_configured, phase_ii_pkts:$reorder_pkts, fast_delay_ms:$delay_a_ms, slow_delay_ms:$delay_b_ms, slow_link_pkts:$band_b_pkts}, @@ -412,12 +506,16 @@ jq -n \ # so `grep -E 'bytes_received=[0-9]+ disconnects=0'` matches on a PASS run). log "" log "reorder-stress: bytes_received=${bytes} disconnects=${disc} duration=${duration_s}s libsrt=${libsrt_ver} reorder_pkts=${reorder_pkts} slow_link_pkts=${band_b_pkts}" +log "reorder-stress[ab]: profile=${PROFILE_LABEL} goodput_bps=${goodput} wire_amp=${wire_amp} ts_sync_errors=${ts_sync} ts_cc_errors=${ts_cc} pkt_rcv_drop=${pkt_drop} pkt_retrans=${pkt_retr}" log "" log "================ reorder-stress summary ================" +log " profile=${PROFILE_LABEL} bitrate=${BITRATE_KBPS}k rx_latency=${RX_LATENCY_MS}ms" +log " recipe: nakreport=${NAKREPORT:-default} lossmaxttl=${LOSSMAXTTL:-default} reorderfreeze=${REORDERFREEZE:-default} seed=${NETEM_SEED:-none}" log " handshake_ok=${handshake_ok} (both_links_added=${both_links_added})" log " bytes_ok=${bytes_ok} (bytes=${bytes} >= 5000) disc_ok=${disc_ok} (disc=${disc})" log " duration_ok=${duration_ok} (duration=${duration_s}s >= 30)" log " reorder_active=${reorder_active} (configured=${reorder_configured} phase_ii_pkts=${reorder_pkts})" +log " equal-gate signal: goodput_bps=${goodput} wire_amp=${wire_amp} ts_sync=${ts_sync} ts_cc=${ts_cc} pkt_drop=${pkt_drop} pkt_retrans=${pkt_retr}" log " libsrt=${libsrt_ver} loader=${SINK_RESOLVED:-}" log " result: ${RESULT_JSON}" log "=======================================================" diff --git a/tests/compat/scenarios/ts-continuity-loopback.sh b/tests/compat/scenarios/ts-continuity-loopback.sh new file mode 100755 index 0000000..0e76586 --- /dev/null +++ b/tests/compat/scenarios/ts-continuity-loopback.sh @@ -0,0 +1,273 @@ +#!/usr/bin/env bash +# +# ts-continuity-loopback.sh — real-SRT loopback that proves srt-sink's MPEG-TS +# continuity metric (ts_cc_errors) is both correct on a clean stream and +# FALSIFIABLE on a damaged one. +# +# Two legs run an actual SRT caller -> srt-sink (SRT listener) transfer over +# 127.0.0.1 and read the metrics srt-sink writes to its --result JSON: +# +# CLEAN : a freshly-muxed MPEG-TS is sent raw (byte-faithful) -> the sink +# must report ts_sync_errors == 0 AND ts_cc_errors == 0. +# DROPPED : the SAME TS with exactly one payload-bearing TS packet excised +# (a real continuity break) is sent raw -> the sink must report +# ts_cc_errors > 0. +# +# WHY raw passthrough (srt-live-transmit), not ffmpeg: ffmpeg re-muxes the TS +# container even with `-c copy`, regenerating the continuity counters and so +# HEALING the injected drop. srt-live-transmit forwards the file bytes verbatim, +# preserving the discontinuity end-to-end. The payload is paced to ~real time +# (a small Python pacer) and zero-padded to a whole SRT chunk (with valid null +# TS packets, PID 0x1FFF) so the live socket neither drops "too-late" packets +# nor zero-pads a short final message into a spurious sync error. +# +# This is the integration-level companion to the hermetic unit test +# `ts-continuity-test` (ctest), which proves the same falsifiability against the +# parser directly with no network. +# +# Privilege: NONE. Requires srt-live-transmit (libsrt-tools); without it the +# scenario SKIPs cleanly (exit 3), like the other capability-gated scenarios. +# +# Usage: +# ts-continuity-loopback.sh [--build-dir DIR] [--duration SEC] [--keep-logs] [-h] +# --duration SEC per-leg media length (default 6). +# +# Artifacts land in tests/compat/results/ts-continuity-loopback/ (gitignored); +# nothing is written outside the repo and no `../`-escaping path is used (Rule D). +# +set -uo pipefail + +SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" >/dev/null 2>&1 && pwd)" +REPO_ROOT="$(cd -- "${SCRIPT_DIR}/../../.." >/dev/null 2>&1 && pwd)" +RESULTS_DIR="${SCRIPT_DIR}/../results/ts-continuity-loopback" + +log() { printf '%s\n' "$*" >&2; } +die() { printf 'ts-continuity-loopback: %s\n' "$*" >&2; exit 2; } + +# --------------------------------------------------------------------------- # +# CLI # +# --------------------------------------------------------------------------- # +BUILD_DIR="${SRTLA_BUILD_DIR:-}" +KEEP_LOGS=0 +MEDIA_SEC=6 + +while [[ $# -gt 0 ]]; do + case "$1" in + --build-dir) BUILD_DIR="${2:?--build-dir needs a value}"; shift 2 ;; + --duration) MEDIA_SEC="${2:?--duration needs a value}"; shift 2 ;; + --keep-logs) KEEP_LOGS=1; shift ;; + -h|--help) sed -n '2,40p' "${BASH_SOURCE[0]}" | sed 's/^# \{0,1\}//'; exit 0 ;; + *) die "unknown argument '$1' (try --help)" ;; + esac +done +[[ "$MEDIA_SEC" =~ ^[0-9]+$ && "$MEDIA_SEC" -ge 1 ]] || die "--duration must be a positive integer" + +# Core harness tools are hard requirements; srt-live-transmit is the gated +# capability this scenario SKIPs on (mirrors `requires:` in matrix.yaml). +for tool in ffmpeg jq python3; do + command -v "$tool" >/dev/null 2>&1 || die "required tool '$tool' not found in PATH" +done +if ! command -v srt-live-transmit >/dev/null 2>&1; then + log "SKIP ts-continuity-loopback: srt-live-transmit not found (install libsrt-tools)" + mkdir -p "$RESULTS_DIR" + printf '{"scenario":"ts-continuity-loopback","skipped":true,"reason":"srt-live-transmit not installed"}\n' \ + > "${RESULTS_DIR}/result.json" + exit 3 +fi + +# --------------------------------------------------------------------------- # +# Resolve the srt-sink helper (built with -DBUILD_COMPAT_TESTS=ON). # +# --------------------------------------------------------------------------- # +resolve_build_dir() { + local candidates=() + [[ -n "$BUILD_DIR" ]] && candidates+=("$BUILD_DIR") + candidates+=("${REPO_ROOT}/build" "/tmp/srtla-build") + local d + for d in "${candidates[@]}"; do + [[ -x "${d}/tests/compat/srt-sink/srt-sink" ]] && { printf '%s' "$d"; return 0; } + done + return 1 +} +BUILD_DIR="$(resolve_build_dir)" || die \ + "srt-sink not found. Build with: + cmake -B build -DBUILD_COMPAT_TESTS=ON && cmake --build build -j" +SRT_SINK="${BUILD_DIR}/tests/compat/srt-sink/srt-sink" + +# --------------------------------------------------------------------------- # +# Constants / artifacts # +# --------------------------------------------------------------------------- # +SINK_PORT_CLEAN=4861 +SINK_PORT_DROP=4862 +SRT_LATENCY_MS=500 +SINK_DURATION=$((MEDIA_SEC + 20)) # self-timeout guard, well past the send +TX_TIMEOUT=$((MEDIA_SEC + 12)) # hard cap on a hung srt-live-transmit +SLT_EXIT_SEC=$((MEDIA_SEC + 4)) # srt-live-transmit's own since-start timer + +rm -rf "$RESULTS_DIR"; mkdir -p "$RESULTS_DIR" +CLEAN_TS="${RESULTS_DIR}/source.ts" +CLEAN_PAD_TS="${RESULTS_DIR}/clean.ts" +DROPPED_TS="${RESULTS_DIR}/dropped.ts" +PACER="${RESULTS_DIR}/pace.py" +BUILDER="${RESULTS_DIR}/build_ts.py" +CLEAN_JSON="${RESULTS_DIR}/clean.json" +DROP_JSON="${RESULTS_DIR}/dropped.json" +RESULT_JSON="${RESULTS_DIR}/result.json" + +PIDS=() +cleanup() { local p; for p in "${PIDS[@]:-}"; do [[ -n "$p" ]] && kill -TERM "$p" 2>/dev/null; done; } +trap cleanup EXIT INT TERM + +# --------------------------------------------------------------------------- # +# Helper scripts (kept in the gitignored results dir; Rule D). # +# --------------------------------------------------------------------------- # +# Pace a file to stdout over ~DUR seconds in 1316-byte (7x188) chunks so the SRT +# live socket sees a real-time stream instead of a loss-inducing burst. +cat > "$PACER" <<'PY' +import sys, time +CH = 1316 +data = open(sys.argv[1], 'rb').read() +dur = float(sys.argv[2]) +n = (len(data) + CH - 1) // CH +interval = dur / max(n, 1) +w = sys.stdout.buffer +t0 = time.monotonic() +for i in range(n): + w.write(data[i*CH:(i+1)*CH]) + w.flush() + target = t0 + (i + 1) * interval + dt = target - time.monotonic() + if dt > 0: + time.sleep(dt) +PY + +# Build the clean (padded) and dropped (one excised payload packet, padded) TS. +# Padding uses valid null TS packets (PID 0x1FFF) so the file length is a whole +# multiple of the 1316-byte SRT chunk and the parser still sees sync byte 0x47. +cat > "$BUILDER" <<'PY' +import sys +from collections import Counter +PKT = 188 +CH = 1316 +src, clean_out, dropped_out = sys.argv[1], sys.argv[2], sys.argv[3] + +def null_pkt(): + # sync=0x47, PID=0x1FFF, afc=01 (payload), cc=0, then 0xFF stuffing + return bytes([0x47, 0x1F, 0xFF, 0x10]) + b'\xff' * 184 + +def pad(d): + while len(d) % CH != 0: + d += null_pkt() + return d + +d = open(src, 'rb').read() +counts = Counter() +for i in range(0, len(d) - PKT + 1, PKT): + if d[i] == 0x47: + pid = ((d[i + 1] & 0x1F) << 8) | d[i + 2] + if pid != 0x1FFF: + counts[pid] += 1 +if not counts: + sys.exit('no non-null PIDs in source TS') +pid = counts.most_common(1)[0][0] +idxs = [i for i in range(0, len(d) - PKT + 1, PKT) + if d[i] == 0x47 and (((d[i + 1] & 0x1F) << 8) | d[i + 2]) == pid] +drop = idxs[len(idxs) // 2] +open(clean_out, 'wb').write(pad(d)) +open(dropped_out, 'wb').write(pad(d[:drop] + d[drop + PKT:])) +print('dominant_pid=0x%04x occurrences=%d dropped_off=%d' % (pid, len(idxs), drop)) +PY + +# --------------------------------------------------------------------------- # +# Build the fixtures. # +# --------------------------------------------------------------------------- # +log "==> ts-continuity-loopback (build dir: ${BUILD_DIR})" +ffmpeg -hide_banner -loglevel error -f lavfi -i "testsrc2=size=320x240:rate=25" \ + -t "$MEDIA_SEC" -c:v mpeg2video -b:v 1M -f mpegts "$CLEAN_TS" \ + || die "ffmpeg could not generate the source TS" +fixture_info="$(python3 "$BUILDER" "$CLEAN_TS" "$CLEAN_PAD_TS" "$DROPPED_TS")" \ + || die "fixture builder failed: ${fixture_info:-}" +log " fixtures: ${fixture_info}" + +# --------------------------------------------------------------------------- # +# Run one leg: SRT caller (paced raw send) -> srt-sink. Echoes nothing; the # +# metrics are read from "$out". # +# --------------------------------------------------------------------------- # +run_leg() { + local src="$1" out="$2" port="$3" txlog="$4" sinklog="$5" + rm -f "$out" + "$SRT_SINK" --port "$port" --host 127.0.0.1 --result "$out" \ + --latency "$SRT_LATENCY_MS" --duration "$SINK_DURATION" \ + >"$sinklog" 2>&1 & + local sp=$!; PIDS+=("$sp") + sleep 0.6 + ( python3 "$PACER" "$src" "$MEDIA_SEC" \ + | timeout "$TX_TIMEOUT" srt-live-transmit -t:"$SLT_EXIT_SEC" -chunk:1316 \ + "file://con" \ + "srt://127.0.0.1:${port}?mode=caller&transtype=live&latency=${SRT_LATENCY_MS}" \ + ) >"$txlog" 2>&1 + sleep 1.5 + kill -TERM "$sp" 2>/dev/null; wait "$sp" 2>/dev/null +} + +metric() { jq -r ".$2 // -1" "$1" 2>/dev/null || echo -1; } + +log "==> CLEAN leg (port ${SINK_PORT_CLEAN})" +run_leg "$CLEAN_PAD_TS" "$CLEAN_JSON" "$SINK_PORT_CLEAN" \ + "${RESULTS_DIR}/clean-tx.log" "${RESULTS_DIR}/clean-sink.log" +clean_pkts="$(metric "$CLEAN_JSON" ts_packets)" +clean_sync="$(metric "$CLEAN_JSON" ts_sync_errors)" +clean_cc="$(metric "$CLEAN_JSON" ts_cc_errors)" + +log "==> DROPPED leg (port ${SINK_PORT_DROP})" +run_leg "$DROPPED_TS" "$DROP_JSON" "$SINK_PORT_DROP" \ + "${RESULTS_DIR}/dropped-tx.log" "${RESULTS_DIR}/dropped-sink.log" +drop_pkts="$(metric "$DROP_JSON" ts_packets)" +drop_sync="$(metric "$DROP_JSON" ts_sync_errors)" +drop_cc="$(metric "$DROP_JSON" ts_cc_errors)" + +for v in "$clean_pkts" "$clean_sync" "$clean_cc" "$drop_pkts" "$drop_sync" "$drop_cc"; do + [[ "$v" =~ ^-?[0-9]+$ ]] || die "could not read TS metrics from a result file" +done + +# --------------------------------------------------------------------------- # +# Verdict. # +# --------------------------------------------------------------------------- # +clean_ok=false +[[ "$clean_pkts" -gt 0 && "$clean_sync" -eq 0 && "$clean_cc" -eq 0 ]] && clean_ok=true +dropped_ok=false +[[ "$drop_cc" -gt 0 ]] && dropped_ok=true + +pass=false +[[ "$clean_ok" == true && "$dropped_ok" == true ]] && pass=true + +jq -n \ + --argjson pass "$pass" \ + --argjson clean_ok "$clean_ok" --argjson dropped_ok "$dropped_ok" \ + --argjson clean_pkts "$clean_pkts" --argjson clean_sync "$clean_sync" --argjson clean_cc "$clean_cc" \ + --argjson drop_pkts "$drop_pkts" --argjson drop_sync "$drop_sync" --argjson drop_cc "$drop_cc" \ + --argjson media_sec "$MEDIA_SEC" \ + --arg ts "$(date -u +%Y-%m-%dT%H:%M:%SZ)" \ + '{ + scenario:"ts-continuity-loopback", pass:$pass, + criteria:{clean_ok:$clean_ok, dropped_ok:$dropped_ok}, + clean:{ts_packets:$clean_pkts, ts_sync_errors:$clean_sync, ts_cc_errors:$clean_cc}, + dropped:{ts_packets:$drop_pkts, ts_sync_errors:$drop_sync, ts_cc_errors:$drop_cc}, + media_sec:$media_sec, timestamp:$ts + }' > "$RESULT_JSON" + +log "" +log "ts-continuity-loopback: clean_cc=${clean_cc} clean_sync=${clean_sync} dropped_cc=${drop_cc} clean_pkts=${clean_pkts} dropped_pkts=${drop_pkts}" +log "" +log "================ ts-continuity-loopback summary ================" +log " clean_ok=${clean_ok} (pkts=${clean_pkts} sync=${clean_sync} cc=${clean_cc} — expect cc=0,sync=0)" +log " dropped_ok=${dropped_ok} (pkts=${drop_pkts} sync=${drop_sync} cc=${drop_cc} — expect cc>0)" +log " result: ${RESULT_JSON}" +log "===============================================================" + +if [[ "$KEEP_LOGS" -eq 0 && "$pass" == true ]]; then + rm -f "${RESULTS_DIR}/clean-tx.log" "${RESULTS_DIR}/clean-sink.log" \ + "${RESULTS_DIR}/dropped-tx.log" "${RESULTS_DIR}/dropped-sink.log" \ + "$CLEAN_TS" "$CLEAN_PAD_TS" "$DROPPED_TS" "$PACER" "$BUILDER" +fi + +if [[ "$pass" == true ]]; then log "PASS"; exit 0; else log "FAIL"; exit 1; fi diff --git a/tests/compat/srt-sink/CMakeLists.txt b/tests/compat/srt-sink/CMakeLists.txt index 169d75b..6607adf 100644 --- a/tests/compat/srt-sink/CMakeLists.txt +++ b/tests/compat/srt-sink/CMakeLists.txt @@ -28,3 +28,14 @@ else() target_include_directories(srt-sink PRIVATE "${SRT_INCLUDE_DIR}") target_link_libraries(srt-sink PRIVATE "${SRT_LIBRARY}") endif() + +# Hermetic unit test for the MPEG-TS continuity tracker (ts_continuity.h) that +# backs srt-sink's ts_packets / ts_sync_errors / ts_cc_errors metrics. It needs +# no libsrt and no network — it injects a clean stream (cc_errors==0) and a +# stream with a dropped packet (cc_errors>0), proving the metric is falsifiable. +# Registered with ctest so the guarantee is checked on every compat-enabled +# build (enable_testing() is set by the root CMakeLists under SRTLA_BUILD_TESTS). +add_executable(ts-continuity-test ts_continuity_test.cpp) +target_compile_features(ts-continuity-test PRIVATE cxx_std_17) +target_compile_options(ts-continuity-test PRIVATE -Wall -Wextra) +add_test(NAME ts-continuity COMMAND ts-continuity-test) diff --git a/tests/compat/srt-sink/main.cpp b/tests/compat/srt-sink/main.cpp index cf5aa4e..b797192 100644 --- a/tests/compat/srt-sink/main.cpp +++ b/tests/compat/srt-sink/main.cpp @@ -1,15 +1,36 @@ // srt-sink — minimal libsrt listener for the SRTLA compatibility harness. // // Accepts a single SRT caller (the media stream relayed end-to-end through -// srtla_send -> srtla_rec), counts the application bytes it delivers, and on -// exit writes a JSON result file the harness consumes: +// srtla_send -> srtla_rec), counts the application bytes it delivers, inspects +// the MPEG-TS it carries, and on exit writes a JSON result file the harness +// consumes: // -// {"bytes_received": N, "first_byte_ms": M, "disconnects": D, "duration_ms": T} +// {"bytes_received": N, "first_byte_ms": M, "disconnects": D, "duration_ms": T, +// "ts_packets": P, "ts_sync_errors": S, "ts_cc_errors": C, +// "pkt_rcv_loss": L, "pkt_rcv_drop": D2, "pkt_retrans": R, +// "packetfilter": ""} // // bytes_received total SRT payload bytes delivered by srt_recv // first_byte_ms ms from sink start to the first delivered byte (-1 if none) // disconnects times an accepted connection broke mid-run (teardown excluded) // duration_ms ms the sink was alive (start -> exit) +// ts_packets total 188-byte MPEG-TS packets seen in the payload +// ts_sync_errors TS packets whose sync byte != 0x47 +// ts_cc_errors per-PID continuity-counter discontinuities (excludes null +// PID 0x1FFF and the adaptation-field discontinuity_indicator) +// pkt_rcv_loss SRT srt_bstats pktRcvLossTotal (cumulative, summed/conn) +// pkt_rcv_drop SRT srt_bstats pktRcvDropTotal (too-late-to-play drops) +// pkt_retrans SRT srt_bstats pktRetransTotal (retransmitted packets) +// packetfilter the SRTO_PACKETFILTER value negotiated on the accepted data +// socket (read after srt_accept). Empty when no filter is in +// effect for the connection. With --packetfilter set, this is +// how the FEC connect-matrix proves one-sided negotiation: a +// FEC caller yields a non-empty merged config; a plain caller +// yields "" because the responder clears the filter per-conn. +// +// The ts_* keys quantify transport-stream integrity (a spurious-retransmit or +// loss profile shows up as cc_errors + SRT loss/retrans counters); the existing +// keys are preserved unchanged so older harness readers keep working. // // It exits 0 on SIGTERM/SIGINT (clean teardown) or when --duration elapses, and // always flushes the JSON result first. This is a test helper, not production @@ -31,6 +52,8 @@ #include #include +#include "ts_continuity.h" + namespace { volatile sig_atomic_t g_stop = 0; @@ -44,6 +67,12 @@ uint64_t now_ms() { (static_cast(ts.tv_nsec) / 1000000ULL); } +// Referenced by numeric id, not the SRTO_REORDERFREEZE symbol: that enum is +// absent from stock libsrt headers, so the id keeps srt-sink compiling against +// any system libsrt. The set only takes effect against the freeze build; on +// vanilla/patched libsrt the call fails and is reported "unsupported", never fatal. +constexpr int kSrtoReorderFreeze = 120; + struct Options { int port = 4001; std::string host = "0.0.0.0"; @@ -52,12 +81,32 @@ struct Options { int64_t duration_ms = 0; int nakreport = -1; int lossmaxttl = -1; + int retransmitalgo = -1; + int reorderfreeze = -1; + std::string packetfilter; +}; + +// All counters the sink reports. The leading four are the original, frozen +// schema; the rest are additive MPEG-TS / SRT telemetry (Task 5). +struct Result { + uint64_t bytes_received = 0; + int64_t first_byte_ms = -1; + int disconnects = 0; + uint64_t duration_ms = 0; + uint64_t ts_packets = 0; + uint64_t ts_sync_errors = 0; + uint64_t ts_cc_errors = 0; + uint64_t pkt_rcv_loss = 0; + uint64_t pkt_rcv_drop = 0; + uint64_t pkt_retrans = 0; + std::string packetfilter; }; void usage(const char *argv0) { std::fprintf(stderr, "usage: %s --port P --result FILE [--host H] [--latency MS] " - "[--duration SEC] [--nakreport 0|1] [--lossmaxttl N]\n", + "[--duration SEC] [--nakreport 0|1] [--lossmaxttl N] " + "[--retransmitalgo 0|1] [--packetfilter STR]\n", argv0); } @@ -107,6 +156,30 @@ bool parse_args(int argc, char **argv, Options &opt) { std::fprintf(stderr, "srt-sink: --lossmaxttl must be non-negative\n"); return false; } + } else if (a == "--retransmitalgo") { + const char *v = need("--retransmitalgo"); + if (!v) return false; + opt.retransmitalgo = std::atoi(v); + if (opt.retransmitalgo != 0 && opt.retransmitalgo != 1) { + std::fprintf(stderr, "srt-sink: --retransmitalgo must be 0 or 1\n"); + return false; + } + } else if (a == "--reorderfreeze") { + const char *v = need("--reorderfreeze"); + if (!v) return false; + opt.reorderfreeze = std::atoi(v); + if (opt.reorderfreeze != 0 && opt.reorderfreeze != 1) { + std::fprintf(stderr, "srt-sink: --reorderfreeze must be 0 or 1\n"); + return false; + } + } else if (a == "--packetfilter") { + const char *v = need("--packetfilter"); + if (!v) return false; + opt.packetfilter = v; + if (opt.packetfilter.empty()) { + std::fprintf(stderr, "srt-sink: --packetfilter must be non-empty\n"); + return false; + } } else if (a == "-h" || a == "--help") { usage(argv[0]); std::exit(0); @@ -127,8 +200,9 @@ bool parse_args(int argc, char **argv, Options &opt) { } // Atomic-ish write: temp file + rename so a reader never sees a partial JSON. -bool write_result(const std::string &path, uint64_t bytes_received, - int64_t first_byte_ms, int disconnects, uint64_t duration_ms) { +// The four original keys lead and are byte-for-byte unchanged; the ts_*/pkt_* +// keys are appended (additive schema). +bool write_result(const std::string &path, const Result &r) { std::string tmp = path + ".tmp"; FILE *f = std::fopen(tmp.c_str(), "w"); if (!f) { @@ -138,8 +212,15 @@ bool write_result(const std::string &path, uint64_t bytes_received, } std::fprintf(f, "{\"bytes_received\": %" PRIu64 ", \"first_byte_ms\": %" PRId64 - ", \"disconnects\": %d, \"duration_ms\": %" PRIu64 "}\n", - bytes_received, first_byte_ms, disconnects, duration_ms); + ", \"disconnects\": %d, \"duration_ms\": %" PRIu64 + ", \"ts_packets\": %" PRIu64 ", \"ts_sync_errors\": %" PRIu64 + ", \"ts_cc_errors\": %" PRIu64 ", \"pkt_rcv_loss\": %" PRIu64 + ", \"pkt_rcv_drop\": %" PRIu64 ", \"pkt_retrans\": %" PRIu64 + ", \"packetfilter\": \"%s\"" + "}\n", + r.bytes_received, r.first_byte_ms, r.disconnects, r.duration_ms, + r.ts_packets, r.ts_sync_errors, r.ts_cc_errors, r.pkt_rcv_loss, + r.pkt_rcv_drop, r.pkt_retrans, r.packetfilter.c_str()); std::fflush(f); std::fclose(f); if (std::rename(tmp.c_str(), path.c_str()) != 0) { @@ -164,16 +245,38 @@ int main(int argc, char **argv) { signal(SIGPIPE, SIG_IGN); const uint64_t start_ms = now_ms(); - uint64_t bytes_received = 0; - int64_t first_byte_ms = -1; - int disconnects = 0; + Result res; + tscont::Tracker ts; + + // Snapshot a data socket's SRT loss/retrans counters into the running totals + // just before it is closed (cumulative *Total fields, summed across any + // reconnect). Best-effort: a fully torn-down socket may refuse srt_bstats. + auto accumulate_stats = [&](SRTSOCKET s) { + if (s == SRT_INVALID_SOCK) return; + SRT_TRACEBSTATS perf; + std::memset(&perf, 0, sizeof(perf)); + if (srt_bstats(s, &perf, 0) != 0) return; + auto nn = [](int v) -> uint64_t { + return v > 0 ? static_cast(v) : 0; + }; + res.pkt_rcv_loss += nn(perf.pktRcvLossTotal); + res.pkt_rcv_drop += nn(perf.pktRcvDropTotal); + res.pkt_retrans += nn(perf.pktRetransTotal); + }; - // Best-effort result on any early return so the harness always finds a file. - auto flush = [&]() { - write_result(opt.result_path, bytes_received, first_byte_ms, disconnects, - now_ms() - start_ms); + // Copy the live TS counters into res and write the JSON. Used by the + // early-return flush below and by the normal exit path. + auto publish = [&](uint64_t dur_ms) -> bool { + res.duration_ms = dur_ms; + res.ts_packets = ts.packets(); + res.ts_sync_errors = ts.sync_errors(); + res.ts_cc_errors = ts.cc_errors(); + return write_result(opt.result_path, res); }; + // Best-effort result on any early return so the harness always finds a file. + auto flush = [&]() { publish(now_ms() - start_ms); }; + if (srt_startup() < 0) { std::fprintf(stderr, "srt-sink: srt_startup failed: %s\n", srt_getlasterror_str()); flush(); @@ -194,13 +297,36 @@ int main(int argc, char **argv) { srt_setsockflag(listener, SRTO_RCVSYN, &no, sizeof(no)); // non-blocking accept srt_setsockflag(listener, SRTO_RCVLATENCY, &opt.latency_ms, sizeof(opt.latency_ms)); - // Apply optional libsrt evaluation flags + // Apply optional libsrt evaluation flags. These are pre-bind options on the + // listener and are inherited by the accepted data socket. if (opt.nakreport >= 0) { srt_setsockflag(listener, SRTO_NAKREPORT, &opt.nakreport, sizeof(opt.nakreport)); } if (opt.lossmaxttl >= 0) { srt_setsockflag(listener, SRTO_LOSSMAXTTL, &opt.lossmaxttl, sizeof(opt.lossmaxttl)); } + if (opt.retransmitalgo >= 0) { + srt_setsockflag(listener, SRTO_RETRANSMITALGO, &opt.retransmitalgo, + sizeof(opt.retransmitalgo)); + } + int reorderfreeze_applied = -1; + if (opt.reorderfreeze >= 0) { + int rc = srt_setsockflag(listener, static_cast(kSrtoReorderFreeze), + &opt.reorderfreeze, sizeof(opt.reorderfreeze)); + reorderfreeze_applied = (rc == 0) ? 1 : 0; + } + + if (!opt.packetfilter.empty()) { + if (srt_setsockflag(listener, SRTO_PACKETFILTER, opt.packetfilter.c_str(), + static_cast(opt.packetfilter.size())) == SRT_ERROR) { + std::fprintf(stderr, "srt-sink: SRTO_PACKETFILTER '%s' rejected: %s\n", + opt.packetfilter.c_str(), srt_getlasterror_str()); + srt_close(listener); + srt_cleanup(); + flush(); + return 1; + } + } struct sockaddr_in sa {}; sa.sin_family = AF_INET; @@ -240,11 +366,23 @@ int main(int argc, char **argv) { uint32_t minor = (srt_version >> 16) & 0xFF; uint32_t patch = (srt_version >> 8) & 0xFF; std::fprintf(stderr, "srt-sink: libsrt version %u.%u.%u\n", major, minor, patch); - std::fprintf(stderr, "srt-sink: nakreport=%s lossmaxttl=%s\n", + const char *reorderfreeze_str = + opt.reorderfreeze < 0 + ? "default" + : (reorderfreeze_applied == 1 ? (opt.reorderfreeze ? "on" : "off") + : "unsupported"); + std::fprintf(stderr, + "srt-sink: nakreport=%s lossmaxttl=%s retransmitalgo=%s " + "reorderfreeze=%s\n", opt.nakreport < 0 ? "default" : (opt.nakreport ? "on" : "off"), - opt.lossmaxttl < 0 ? "default" : std::to_string(opt.lossmaxttl).c_str()); + opt.lossmaxttl < 0 ? "default" : std::to_string(opt.lossmaxttl).c_str(), + opt.retransmitalgo < 0 ? "default" + : std::to_string(opt.retransmitalgo).c_str(), + reorderfreeze_str); std::fprintf(stderr, "srt-sink: listening on %s:%d (latency %dms)\n", opt.host.c_str(), opt.port, opt.latency_ms); + std::fprintf(stderr, "srt-sink: packetfilter=%s\n", + opt.packetfilter.empty() ? "none" : opt.packetfilter.c_str()); SRTSOCKET client = SRT_INVALID_SOCK; char buf[1500]; @@ -279,14 +417,26 @@ int main(int argc, char **argv) { srt_setsockflag(client, SRTO_RCVSYN, &cno, sizeof(cno)); int cev = SRT_EPOLL_IN | SRT_EPOLL_ERR; srt_epoll_add_usock(eid, client, &cev); - std::fprintf(stderr, "srt-sink: accepted SRT caller\n"); + + // The packet-filter config negotiated for THIS connection: non-empty + // when FEC was agreed, "" when the responder cleared it (plain caller). + char pfbuf[512]; + std::memset(pfbuf, 0, sizeof(pfbuf)); + int pflen = static_cast(sizeof(pfbuf)); + if (srt_getsockflag(client, SRTO_PACKETFILTER, pfbuf, &pflen) == 0 && + pflen > 0) { + res.packetfilter.assign(pfbuf, static_cast(pflen)); + } + std::fprintf(stderr, "srt-sink: accepted SRT caller (packetfilter='%s')\n", + res.packetfilter.c_str()); continue; } if (s != client) continue; if (what & SRT_EPOLL_ERR) { - if (!g_stop) ++disconnects; + if (!g_stop) ++res.disconnects; + accumulate_stats(client); srt_epoll_remove_usock(eid, client); srt_close(client); client = SRT_INVALID_SOCK; @@ -298,14 +448,17 @@ int main(int argc, char **argv) { for (;;) { int r = srt_recv(client, buf, sizeof(buf)); if (r > 0) { - if (first_byte_ms < 0) { - first_byte_ms = static_cast(now_ms() - start_ms); + if (res.first_byte_ms < 0) { + res.first_byte_ms = static_cast(now_ms() - start_ms); } - bytes_received += static_cast(r); + res.bytes_received += static_cast(r); + ts.feed(reinterpret_cast(buf), + static_cast(r)); continue; // drain everything ready before re-polling } if (r == 0) { - if (!g_stop) ++disconnects; + if (!g_stop) ++res.disconnects; + accumulate_stats(client); srt_epoll_remove_usock(eid, client); srt_close(client); client = SRT_INVALID_SOCK; @@ -315,7 +468,8 @@ int main(int argc, char **argv) { // r == SRT_ERROR int err = srt_getlasterror(nullptr); if (err == SRT_EASYNCRCV) break; // no more data right now - if (!g_stop) ++disconnects; + if (!g_stop) ++res.disconnects; + accumulate_stats(client); srt_epoll_remove_usock(eid, client); srt_close(client); client = SRT_INVALID_SOCK; @@ -327,18 +481,31 @@ int main(int argc, char **argv) { } } - if (client != SRT_INVALID_SOCK) srt_close(client); + if (client != SRT_INVALID_SOCK) { + accumulate_stats(client); + srt_close(client); + } srt_close(listener); srt_cleanup(); const uint64_t duration_ms = now_ms() - start_ms; + res.duration_ms = duration_ms; + res.ts_packets = ts.packets(); + res.ts_sync_errors = ts.sync_errors(); + res.ts_cc_errors = ts.cc_errors(); std::fprintf(stderr, "srt-sink: bytes=%" PRIu64 " first_byte_ms=%" PRId64 " disconnects=%d duration_ms=%" PRIu64 "\n", - bytes_received, first_byte_ms, disconnects, duration_ms); + res.bytes_received, res.first_byte_ms, res.disconnects, + duration_ms); + std::fprintf(stderr, + "srt-sink: ts_packets=%" PRIu64 " ts_sync_errors=%" PRIu64 + " ts_cc_errors=%" PRIu64 " pkt_rcv_loss=%" PRIu64 + " pkt_rcv_drop=%" PRIu64 " pkt_retrans=%" PRIu64 "\n", + res.ts_packets, res.ts_sync_errors, res.ts_cc_errors, + res.pkt_rcv_loss, res.pkt_rcv_drop, res.pkt_retrans); - if (!write_result(opt.result_path, bytes_received, first_byte_ms, disconnects, - duration_ms)) { + if (!write_result(opt.result_path, res)) { return 1; } return 0; diff --git a/tests/compat/srt-sink/ts_continuity.h b/tests/compat/srt-sink/ts_continuity.h new file mode 100644 index 0000000..3684a11 --- /dev/null +++ b/tests/compat/srt-sink/ts_continuity.h @@ -0,0 +1,125 @@ +// ts_continuity.h — MPEG-TS continuity / sync tracking for the srt-sink helper. +// +// An SRT *live* payload carries one or more contiguous 188-byte MPEG-TS +// packets. This header reassembles the byte stream across srt_recv boundaries +// (a small carry buffer) and, per PID, tracks the 4-bit continuity counter so +// the sink can report objective transport-stream health alongside its byte +// counters: +// +// packets total 188-byte TS packets observed +// sync_errors packets whose sync byte != 0x47 (mis-alignment / corruption) +// cc_errors continuity-counter discontinuities — a per-PID CC step other +// than +1 mod 16 on a payload-bearing packet — EXCLUDING the +// null PID (0x1FFF) and packets carrying the adaptation-field +// discontinuity_indicator (an intentional, signalled break). +// +// CC semantics (ISO/IEC 13818-1 §2.4.3.3): the continuity_counter only +// increments on packets that carry a payload (adaptation_field_control 01/11); +// adaptation-only packets (10) and the reserved value (00) leave it unchanged. +// We model that so a clean stream with PCR-only stuffing never false-positives. +// +// Header-only and dependency-free (no libsrt, no srtla internals) so the parser +// is unit-testable in isolation; srt-sink simply feeds it the bytes srt_recv +// delivers. A duplicate packet (same CC, payload present) is counted as a +// discontinuity here — the spec permits a single duplicate, but a measurement +// sink errs toward visibility and real encoder output does not duplicate. + +#ifndef SRTLA_COMPAT_TS_CONTINUITY_H +#define SRTLA_COMPAT_TS_CONTINUITY_H + +#include +#include + +namespace tscont { + +inline constexpr size_t TS_PACKET_SIZE = 188; +inline constexpr uint8_t TS_SYNC_BYTE = 0x47; +inline constexpr uint16_t TS_NULL_PID = 0x1FFF; // 13-bit null-packet PID +inline constexpr int TS_PID_COUNT = 8192; // 2^13 possible PIDs + +class Tracker { +public: + Tracker() { + for (int i = 0; i < TS_PID_COUNT; ++i) last_cc_[i] = -1; // -1 = unseen + } + + uint64_t packets() const { return packets_; } + uint64_t sync_errors() const { return sync_errors_; } + uint64_t cc_errors() const { return cc_errors_; } + + // Feed the raw bytes delivered by one srt_recv. Packets straddling the + // previous call are completed first via the carry buffer, so chunking is + // irrelevant to the counts. + void feed(const uint8_t *data, size_t len) { + if (data == nullptr || len == 0) return; + size_t off = 0; + + // Complete a packet left half-read at the end of the previous feed(). + if (carry_len_ > 0) { + const size_t need = TS_PACKET_SIZE - carry_len_; + const size_t take = need < len ? need : len; + for (size_t i = 0; i < take; ++i) carry_[carry_len_ + i] = data[i]; + carry_len_ += take; + off += take; + if (carry_len_ < TS_PACKET_SIZE) return; // still incomplete + parse_packet(carry_); + carry_len_ = 0; + } + + while (off + TS_PACKET_SIZE <= len) { + parse_packet(data + off); + off += TS_PACKET_SIZE; + } + + // Stash the sub-packet remainder for the next feed(). + const size_t rem = len - off; + for (size_t i = 0; i < rem; ++i) carry_[i] = data[off + i]; + carry_len_ = rem; + } + +private: + void parse_packet(const uint8_t *p) { + ++packets_; + if (p[0] != TS_SYNC_BYTE) { + ++sync_errors_; + return; // un-aligned/corrupt: PID and CC fields are untrustworthy + } + + const uint16_t pid = + static_cast((static_cast(p[1] & 0x1F) << 8) | p[2]); + if (pid == TS_NULL_PID) return; // null packets carry no continuity + + const uint8_t afc = static_cast((p[3] >> 4) & 0x03); + const bool has_payload = (afc == 0x01 || afc == 0x03); + const bool has_adaptation = (afc == 0x02 || afc == 0x03); + const uint8_t cc = static_cast(p[3] & 0x0F); + + bool discontinuity = false; + if (has_adaptation) { + const uint8_t af_len = p[4]; // adaptation_field_length + if (af_len > 0) { + discontinuity = (p[5] & 0x80) != 0; // discontinuity_indicator (bit 7) + } + } + + const int prev = last_cc_[pid]; + if (prev >= 0 && !discontinuity) { + const uint8_t expected = + has_payload ? static_cast((prev + 1) & 0x0F) + : static_cast(prev); + if (cc != expected) ++cc_errors_; + } + last_cc_[pid] = static_cast(cc); + } + + uint64_t packets_ = 0; + uint64_t sync_errors_ = 0; + uint64_t cc_errors_ = 0; + int last_cc_[TS_PID_COUNT]; + uint8_t carry_[TS_PACKET_SIZE]; + size_t carry_len_ = 0; +}; + +} // namespace tscont + +#endif // SRTLA_COMPAT_TS_CONTINUITY_H diff --git a/tests/compat/srt-sink/ts_continuity_test.cpp b/tests/compat/srt-sink/ts_continuity_test.cpp new file mode 100644 index 0000000..71eeb91 --- /dev/null +++ b/tests/compat/srt-sink/ts_continuity_test.cpp @@ -0,0 +1,158 @@ +// ts_continuity_test.cpp — hermetic unit test for the MPEG-TS continuity +// tracker that backs srt-sink's ts_packets / ts_sync_errors / ts_cc_errors +// metrics. No libsrt, no network, no ffmpeg: it synthesises transport-stream +// bytes in memory, feeds them to tscont::Tracker, and asserts the metric is +// falsifiable — a clean stream yields cc_errors==0 while a stream with a +// dropped packet yields cc_errors>0. Built under BUILD_COMPAT_TESTS and wired +// into ctest so the guarantee is checked on every compat-enabled build. + +#include "ts_continuity.h" + +#include +#include +#include + +namespace { + +int g_failures = 0; + +#define CHECK(cond, msg) \ + do { \ + if (!(cond)) { \ + std::fprintf(stderr, "FAIL: %s (%s:%d)\n", (msg), __FILE__, __LINE__); \ + ++g_failures; \ + } \ + } while (0) + +// Build one 188-byte TS packet. +// pid 13-bit packet identifier +// cc 4-bit continuity counter +// payload true -> adaptation_field_control "01" (payload only) +// false -> "10" (adaptation only; CC must NOT increment) +// discontinuity true -> set the adaptation-field discontinuity_indicator +// (forces an adaptation field to be present) +std::vector make_packet(uint16_t pid, uint8_t cc, bool payload = true, + bool discontinuity = false) { + std::vector p(tscont::TS_PACKET_SIZE, 0xFF); + p[0] = tscont::TS_SYNC_BYTE; + p[1] = static_cast((pid >> 8) & 0x1F); + p[2] = static_cast(pid & 0xFF); + + uint8_t afc; + if (discontinuity) { + afc = payload ? 0x3 : 0x2; // adaptation field required to carry the flag + } else { + afc = payload ? 0x1 : 0x2; + } + p[3] = static_cast((afc << 4) | (cc & 0x0F)); + if (afc == 0x2 || afc == 0x3) { + p[4] = (afc == 0x2) ? 183 : 1; // adaptation_field_length + p[5] = discontinuity ? 0x80 : 0x00; // discontinuity_indicator (bit 7) + } + return p; +} + +void append(std::vector &s, const std::vector &p) { + s.insert(s.end(), p.begin(), p.end()); +} + +} // namespace + +int main() { + using tscont::Tracker; + + // 1) Clean stream: one PID, CC 0..15 wrapping over 64 packets -> no errors. + { + std::vector s; + const int N = 64; + for (int i = 0; i < N; ++i) append(s, make_packet(0x100, static_cast(i & 0x0F))); + Tracker t; + t.feed(s.data(), s.size()); + CHECK(t.packets() == static_cast(N), "clean: packet count"); + CHECK(t.sync_errors() == 0, "clean: no sync errors"); + CHECK(t.cc_errors() == 0, "clean: no cc errors"); + } + + // 2) Dropped packet: omit the packet whose CC would be 5 -> >=1 cc error. + // This is the falsifiability proof: a real drop must move the metric. + { + std::vector s; + for (int i = 0; i < 10; ++i) { + if (i == 5) continue; // physically drop one packet, preserving alignment + append(s, make_packet(0x100, static_cast(i & 0x0F))); + } + Tracker t; + t.feed(s.data(), s.size()); + CHECK(t.packets() == 9, "drop: packet count"); + CHECK(t.sync_errors() == 0, "drop: no sync errors"); + CHECK(t.cc_errors() >= 1, "drop: cc error detected"); + } + + // 3) Sync error: clobber a sync byte in place (alignment preserved). + { + std::vector s; + for (int i = 0; i < 4; ++i) append(s, make_packet(0x100, static_cast(i & 0x0F))); + s[2 * tscont::TS_PACKET_SIZE] = 0x00; // 3rd packet's sync byte + Tracker t; + t.feed(s.data(), s.size()); + CHECK(t.sync_errors() >= 1, "sync: bad sync byte detected"); + } + + // 4) Null PID (0x1FFF) carries no continuity: an arbitrary CC jump is benign. + { + std::vector s; + append(s, make_packet(0x1FFF, 3)); + append(s, make_packet(0x1FFF, 9)); + Tracker t; + t.feed(s.data(), s.size()); + CHECK(t.cc_errors() == 0, "null-pid: no cc error"); + } + + // 5) discontinuity_indicator suppresses the CC-jump error. + { + std::vector s; + append(s, make_packet(0x100, 4)); + append(s, make_packet(0x100, 9, /*payload=*/true, /*discontinuity=*/true)); + Tracker t; + t.feed(s.data(), s.size()); + CHECK(t.cc_errors() == 0, "discontinuity-indicator: suppresses error"); + } + + // 6) Adaptation-only packet (no payload) must not increment CC: a repeated CC + // across a "10" packet is correct, not an error. + { + std::vector s; + append(s, make_packet(0x100, 4)); // payload, cc=4 + append(s, make_packet(0x100, 4, /*payload=*/false)); // adaptation only, cc stays 4 + append(s, make_packet(0x100, 5)); // payload, cc=5 + Tracker t; + t.feed(s.data(), s.size()); + CHECK(t.cc_errors() == 0, "adaptation-only: cc hold is not an error"); + } + + // 7) Carry buffer: feeding the SAME clean stream in ragged sub-188 chunks + // (TS packets straddle srt_recv boundaries) yields identical counts. + { + std::vector s; + const int N = 40; + for (int i = 0; i < N; ++i) append(s, make_packet(0x021, static_cast(i & 0x0F))); + Tracker t; + size_t off = 0; + const size_t chunk = 100; // deliberately not a multiple of 188 + while (off < s.size()) { + const size_t n = (off + chunk <= s.size()) ? chunk : (s.size() - off); + t.feed(s.data() + off, n); + off += n; + } + CHECK(t.packets() == static_cast(N), "carry: packet count"); + CHECK(t.sync_errors() == 0, "carry: no sync errors"); + CHECK(t.cc_errors() == 0, "carry: no cc errors"); + } + + if (g_failures == 0) { + std::printf("ts-continuity-test: all checks passed\n"); + return 0; + } + std::fprintf(stderr, "ts-continuity-test: %d check(s) failed\n", g_failures); + return 1; +}