diff --git a/docker/guest-ingest.Dockerfile b/docker/guest-ingest.Dockerfile index 1354b04..a6ebebf 100644 --- a/docker/guest-ingest.Dockerfile +++ b/docker/guest-ingest.Dockerfile @@ -18,8 +18,13 @@ # docker build -f docker/guest-ingest.Dockerfile -t mxl-guest-ingest:local . # Run (quickstart wires this): one container per guest slot, SRT in → MXL flow. -# Base pinned by digest for reproducibility (docs/VERSIONS.md). Override to track -# upstream: docker build --build-arg BASE=ghcr.io/cbcrc/test-generator:latest ... +# Base pinned by digest for reproducibility (docs/VERSIONS.md). This digest bakes +# in MXL SDK v1.1.0 (released 2026-09-09; Flow + Fabric API) — the gst-mxl-rs +# contract contribution_core._restamp relies on (mxlsink maps pts->grain index via +# one shared clock offset D and has NO backward-index guard, so the writer requires +# monotonic PTS from us). Verify with the commands in docs/VERSIONS.md before a +# cold-clone proof; a base refresh could silently change the SDK version. +# Override to track upstream: docker build --build-arg BASE=ghcr.io/cbcrc/test-generator:latest ... ARG BASE=ghcr.io/cbcrc/test-generator@sha256:09cad0981475095ab948ca51511d4fbdc0521e2a23632d50abaf14fc3847cd92 FROM ${BASE} @@ -29,6 +34,23 @@ RUN apt-get update -qq \ && apt-get install -y -qq --no-install-recommends gstreamer1.0-libav \ && rm -rf /var/lib/apt/lists/* +# OPT-IN: closed-caption / ANC (ST-2038) data-flow support for CaptionFileAdapter +# (Tier 3.1, docs/ROADMAP-FROM-SPECS-2026-10.md). Needs `ccconverter` +# (gstreamer1.0-plugins-bad) + the `rsclosedcaption` plugin (cctost2038anc / +# tttocea608) from gst-plugins-rs >= 0.14, which the stock cbcrc base lacks. OFF by +# default so the HW-proven image is byte-identical; enable for a caption build with: +# docker build --build-arg WITH_CAPTIONS=1 -f docker/guest-ingest.Dockerfile ... +# NOTE: rsclosedcaption is not in Ubuntu apt; the upstream devcontainer builds it from +# source (rust/gst-mxl-rs/README.md). This arg installs plugins-bad and documents the +# remaining gst-plugins-rs source build — it is NOT yet HW-verified end to end. +ARG WITH_CAPTIONS=0 +RUN if [ "$WITH_CAPTIONS" = "1" ]; then \ + apt-get update -qq \ + && apt-get install -y -qq --no-install-recommends gstreamer1.0-plugins-bad \ + && rm -rf /var/lib/apt/lists/* ; \ + echo "WITH_CAPTIONS=1: plugins-bad (ccconverter) installed; rsclosedcaption (gst-plugins-rs >=0.14) must still be built from source — see rust/gst-mxl-rs/README.md" ; \ + fi + # the contribution seam (core + adapters + entrypoints, flat so imports resolve from /opt/seam) # guest_audio.py = the A/V guest's AUDIO leg (v0.3); facility.py lets both legs resolve # flow UUIDs from config/facility.json (copied below) with the baked-in fallback. diff --git a/docs/ADR-001-nmos-control-plane.md b/docs/ADR-001-nmos-control-plane.md new file mode 100644 index 0000000..24c1e35 --- /dev/null +++ b/docs/ADR-001-nmos-control-plane.md @@ -0,0 +1,99 @@ + + + +# ADR-001: NMOS owns connection control; we own the media engine + +- **Status:** Accepted +- **Date:** 2026-10-08 +- **Decision owners:** mxl-cloud-production-demo maintainers +- **Supersedes / relates to:** `docs/ARCHITECTURE-CORE-AND-SKINS.md`, + `docs/ROADMAP-FROM-SPECS-2026-10.md`, `docs/FINDINGS.md` §6 (reader-wedge) + +## Context + +Two signals from people who define this space prompted a deliberate architectural +decision rather than a drift: + +1. **cbcrc/mxl-hands-on#34** (our report: "readers wedge silently when a source flow is + recreated; status APIs still say running"). The cbcrc maintainer closed it as + wontfix-for-the-test-apps: *"The right solution for this will be to embrace NMOS when it + is going to be available."* He pointed at `garethsb/mxl-nmos-c-example`. +2. **`garethsb/mxl-nmos-c-example`** (Gareth Sylvester-Bradley, NVIDIA/AMWA) — a producer + Node + consumer Node where the **consumer opens MXL flows only after an IS-05 activation**, + carrying `{mxl_flow_id, mxl_domain_id}` in the IS-05 `transport_params`, with disable/ + re-enable. It is a *connection* demo: it contains no switcher — no selector, keyer, mixer, + restamp, or contribution ingress. + +The tempting but wrong framing is "build the switcher with NMOS **instead of** what we're +doing." That treats a control protocol and a media application as alternatives. They are +different layers. NvNmos + Gareth's example give you zero switching; our engine is not a +connection router. The question is not either/or — it is *which layer owns what*. + +## Decision + +**Build and own the media engine. Delegate connection control to NMOS. Do not reinvent the +data plane or the control protocol.** + +Layer ownership: + +| Layer | What it is | Owner | +|---|---|---| +| Data plane | grains in shared memory (`mxlsink`/`mxlsrc`) | **MXL SDK** — consume, never reinvent | +| **Media engine** | select / cut / key / mix / restamp / conform; program & preview | **This project** — the differentiated, unowned work | +| **Contribution ingress** | SRT/RTSP guests → MXL, restamped onto the local grain clock (IN-005) | **This project** | +| **Connection control** | discover flows; route Sender→Receiver; re-open on recreation | **NMOS** (IS-04 discovery, IS-05 connection) | +| Timed metadata | ANC / captions / SCTE as event flows | **MXL v1.2** (dmf-mxl#327) — consume when it lands | + +Corollaries: +- **The media engine is the product.** No one else in the ecosystem is building the open + software switcher; the connection demos and the SDK stop at the boundaries of our layer. + This is our positioning and we invest here. +- **Connection control converges to NMOS.** Our REST cut API and the `/api/mxl/repair` + cascade are the *operational* form of what IS-05 standardizes. We promote NMOS from a + side "shim" to the **primary** routing/connection interface over time. +- **The reader-wedge (#34) proper fix is NMOS re-activation, not stat-polling.** A reader + re-opens a recreated flow through an IS-05 (re)activation — the pattern Gareth's consumer + demonstrates — rather than holding a stale handle. We adopt this once a controller is in + the loop; until then the repair cascade remains the pragmatic patch. + +## What we explicitly will NOT do + +- **Not** reimplement the MXL shared-memory data plane. +- **Not** hand-roll a bespoke connection/routing protocol when IS-04/IS-05 exist. (This is + exactly what #34's closure warned against; we take the hint.) +- **Not** throw away the media engine to "adopt NMOS." NMOS has no media engine to adopt. +- **Not** hard-depend on NvNmos / `nvnmosd`. Gareth's repo and `gst-nmos-rs` are C-API + *references*; our node stays a lightweight BCP-007-03 implementation. We copy the **wire + format** (standard IS-05), not the daemon. + +## Convergence steps (how we get from here to the decision) + +Current state: `tools/nmos_node.py` does IS-04 discovery + an IS-05 **`connection/v1.2`** +shim whose receiver `/staged` activation drives a real selector cut (PR #37). + +Convergence toward the reference (steps 1–4 ✅ done on PR #37, step 5 pending): +1. ✅ **Carry flow identity in `transport_params`**, not just IS-04 tags: + `transport_params: [{ "mxl_flow_id": "", "mxl_domain_id": "" }]`. +2. ✅ **Connection API bumped to v1.2** (the version `connect.sh` in the reference speaks); + device control = `sr-ctrl/v1.2`. +3. ✅ **Sender/Receiver disable + re-enable** (IS-05 `master_enable` false→true; disable + activates without forcing a program cut). +4. ✅ **Domain UUID advertised** as `mxl_domain_id` (resolved from `domain_def.json`'s `id`, + then env, then the legacy string). +5. ⬜ **Reader re-open via re-activation** (the #34 fix) — wire `flow_stabilizer` / the repair + path to re-drive IS-05 activation on flow recreation, once a controller drives activations. + +None of these touch the media engine; they are all in the control-plane layer, which is the +point of this ADR. + +## Consequences + +- **Positive:** we invest effort where we are differentiated (media engine, contribution + seam) and standards-compliant where the ecosystem already converged (NMOS control). We + align with the maintainers who define the space instead of diverging from them. +- **Positive:** the reader-wedge gets an architecturally-correct fix path, not just a patch. +- **Cost:** maturing the NMOS node is real work (the 5 steps above), and full reader-reopen + needs a controller in the loop — deferred until that exists. +- **Risk accepted:** until the convergence steps land, our connection control is non-standard + (REST + repair cascade). That is a known, bounded gap with a written plan to close it — not + a drift. diff --git a/docs/CONFIG.md b/docs/CONFIG.md index 5e5d263..9570e16 100644 --- a/docs/CONFIG.md +++ b/docs/CONFIG.md @@ -62,10 +62,34 @@ Already env-driven before this change (unchanged): `MXL_GUEST_HOST`, `MXL_GUEST_ |---|---|---| | `MXL_SELFHEAL` | `scripts/quickstart.sh` — start `tools/mxl-selfheal.sh --watch` (selector + relay drift recovery) | **unset / off**. Set `MXL_SELFHEAL=1` to enable. | | `MXL_SELFHEAL_INTERVAL` | `tools/mxl-selfheal.sh` — seconds between `--watch` passes | `15` | +| `MXL_INGRESS_DIR` | `tools/contribution_core.py` — directory for the IN-005 ingress registry (one `.json` per ingest: source caps/provenance + the restamp's locked offset, running err, grain step, hard-relock count). Read by `tools/ingress-soak.sh`. | `/tmp/mxl-ingress`. Set **empty** to disable (no file written). | Self-heal is **opt-in**. Desired-state reporting (`docs/DESIRED-STATE-v0.md`) stays report-only and is separate from this watcher. +## IN-005 ingress registry + soak verdict + +Each contribution ingest writes an **IN-005 ingress record** (`$MXL_INGRESS_DIR/.json`) +as the restamp locks and runs — the signal's negotiated source caps (provenance) and the +timing adjustments applied (locked offset, running wall-clock `err_ms`, grain step, +hard-relock count). AMWA IN-005 ("External Signal Ingress for DMF") asks that these be +*traceable*; this is that trace. Inspect it live: + +```sh +cat /tmp/mxl-ingress/9e111e00.json +``` + +`tools/ingress-soak.sh` turns those records into an objective PASS/FAIL in the **Timing +Model's own terms** ("latency low and mostly constant over hours"): over a sampling window +it asserts frames keep advancing (essence moving, not DRIFT), `|err_ms|` stays bounded +(`ERR_BAND_MS`), `err_ms` doesn't trend away (`ERR_DRIFT_MS`), and `hard_relocks` does not +climb (no re-lock churn — the Oct-7 60fps-restamp symptom). This is the acceptance test for +the 60fps restamp HW re-verify and any guest soak: + +```sh +DURATION=300 INTERVAL=10 tools/ingress-soak.sh guest1 # 5-min soak, PASS/FAIL + exit code +``` + ## Security-related switches (see [SECURITY.md](../SECURITY.md)) | Variable | Used by | Default | diff --git a/docs/ROADMAP-FROM-SPECS-2026-10.md b/docs/ROADMAP-FROM-SPECS-2026-10.md new file mode 100644 index 0000000..7cba517 --- /dev/null +++ b/docs/ROADMAP-FROM-SPECS-2026-10.md @@ -0,0 +1,168 @@ + + + +# Switcher roadmap from the Oct 2026 MXL spec/SDK review + +This plan is the output of a source-level review (Oct 7 2026) of the MXL specs, the +`gst-mxl-rs` v1.1.0 plugin source, the AMWA DMF increments, and the dmf-mxl **v1.2** +roadmap (GitHub milestone #9), compared against this repo. It records *what we should +build/change on the switcher and why*, grouped by dependency. + +Sources reviewed: +- `gst-mxl-rs` @ **v1.1.0** — `README.md`, `src/clock.rs`, `src/mxlsink/render_continuous.rs` +- MXL docs — Timing Model, Fabrics +- AMWA increments — IN-001..IN-006 (esp. **IN-005** external-signal ingress) +- dmf-mxl **v1.2** milestone — Timed Data (#327 / PR #720, children #731 #730) and + device-hosted/CUDA grains (#594/#751/#752) +- `tools/mxl-fabrics-demo/demo.cpp` — the real Fabrics C API + +--- + +## The one finding that reframes our restamp + +`gst-mxl-rs` `mxlsink` is **not** a dumb writer. At the source level: + +- **`clock.rs`** — each pipeline samples ONE shared constant `D = mxl_now − + pipeline_clock_now` and shares it across all MXL elements via a GstContext handshake. +- **`render_continuous.rs`** — the sink computes `mxl_ts = buffer.pts + base_time + D` + then `index = timestamp_to_index(mxl_ts, rate)`. **There is no writer-side + backward-index guard**: a backwards PTS → a lower index → a write *behind* the + committed head → the writer rejects it → `-5`. The writer **relies on the caller + presenting monotonic PTS.** + +Consequences for `tools/contribution_core.py::_restamp`: +1. Our monotonic clamp (added Oct 6 for the `-5` loop) is **load-bearing and correct — + keep it.** It does exactly the job the SDK leaves to the writer. +2. Our restamp **also re-derives its own offset** against `pipe.get_clock()`, and the + sink then adds its own `D` (same clock class) on top — the PTS is offset-corrected + twice. They mostly cancel into a constant (that's why it works), but it's fragile. +3. The servo and its diagnostics are hardcoded to 30 fps (`FRAME_NS = 33_333_333`), + which is why a 60 fps source reports `err=-41s / fps=173`. + +--- + +## Tier 1 — Ship now (fixes real bugs, no upstream dependency) + +### 1.1 — Rework the restamp to be rate-aware (the 60 fps bug) ⭐ IN PROGRESS +- **Why:** the open follow-up from Oct 6; now scoped by the `clock.rs` model above. +- **Do:** derive the monotonic frame step from the *flow's* grain rate instead of the + 30 fps `FRAME_NS` constant; keep the conform stage's `videorate → framerate=30/1` + decimation so the grain cadence stays canonical; fix the diag `fps` so it reflects + the committed (post-decimation) cadence, not the raw source rate. Keep the monotonic + clamp untouched. +- **Guardrails:** `tests/test_contribution_core.py` pins the golden launch strings + (`framerate=30/1`, exactly one `videorate`) and the monotonic re-lock — do not change + those. This is a probe-logic change, not a pipeline-shape change. +- **HW:** verify on the camera at the lab (not the phone — it drops <90s); confirm a + 60 fps source yields a sane `fps`/`err` and 0 `-5`. + +### 1.2 — Pin & verify MXL SDK v1.1.0 (half-done) +- Recorded v1.1.0 in `docs/VERSIONS.md` + `docker/guest-ingest.Dockerfile`. Remaining: + run the `gst-inspect`/`strings libmxl.so` verify commands against the pinned + `test-generator` base on the next HW box to confirm (not assert) the version, and + review the two upstream seam commits (`getFlowReader` flow-id check; `gst-mxl-rs` + wait-for-missing-flow + audio-reader-spin) before relying on them. + +### 1.3 — Audio monotonic floor = sample-duration (latent, low-pri) +- **Why:** `render_continuous.rs` indexes audio per sample; `_restamp_audio` uses the + 33 ms video `FRAME_NS` as its floor (1600 samples of slack). Not a bug today (#36 + HW-proven) — a lip-sync-precision cleanup, best folded into 1.1's HW session. + +--- + +## Tier 2 — Align now (cheap spec/positioning wins) + +### 2.1 — IN-005 ingress registry ✅ DONE +- **Why:** IN-005 wants ingress timing adjustments *traceable* via a registry; we only + `print()` `cadence offset locked / re-synced`. +- **Shipped:** `contribution_core` writes `$MXL_INGRESS_DIR/.json` (default + `/tmp/mxl-ingress`) on lock / re-lock / hard-relock / diag — source caps (provenance) + + locked offset, running `err_ms`, grain step, hard-relock count. Atomic write, stdlib-only, + opt-out via empty `MXL_INGRESS_DIR`. Tests + `docs/CONFIG.md` §"IN-005 ingress registry". + +### 2.2 — Timing-Model drift criterion as the soak pass/fail ✅ DONE +- **Why:** the Timing Model says latency must stay "low and mostly constant over hours." + `mxl-info` exposes no transfer latency directly, so derive it from the ingress records. +- **Shipped:** `tools/ingress-soak.sh` samples the IN-005 records over a window and renders + PASS/FAIL in the spec's own terms — frames advancing (not DRIFT), `|err_ms|` bounded, + `err_ms` not trending, `hard_relocks` not climbing (no re-lock churn). This is the + explicit acceptance test for the 60fps restamp HW re-verify. + +### 2.3 — IN-001 / IN-004 gap-note refresh +- **Why:** IN-004 (Flow-Connection Phase-2 control API) is now WIP and extends IN-001. + Map our v1 `/api/mxl/*` control surface to IN-004's direction in + `docs/IN-001-GAP-ANALYSIS.md` to keep the "conformant control API" positioning current. + +--- + +## Tier 3 — Track & prep for v1.2 + +### 3.1 — ANC / closed-captions data flow 🟡 CODE DONE, HW-verify owed +- **Why:** v1.1.0's `gst-mxl-rs` already has a worked, tested + `meta/x-st-2038 ↔ video/smpte291` CEA-608 round-trip. De-risked path to captions / + SCTE through the switcher today, and the exact thing v1.2 Timed Data generalizes. +- **Shipped (no-VM):** + - `contribution_core` gained a **third essence: `data`** — a conform branch that enforces + `CANON_DATA_CAPS = meta/x-st-2038,alignment=frame` + queue → `mxlsink` (writes the + `video/smpte291` data flow). No restamp (forced `timing_policy=preserve`; ANC grains are + frame-aligned). `group_hint` knows `:Data`. + - `CaptionFileAdapter` (adapters.py) mirrors the v1.1.0 README producer chain + `filesrc ! subparse ! tttocea608 ! ccconverter ! closedcaption/x-cea-608 ! cctost2038anc`. + - `docker/guest-ingest.Dockerfile` `--build-arg WITH_CAPTIONS=1` (OFF by default so the + HW-proven image is byte-identical) installs `plugins-bad` (`ccconverter`); it documents + that `rsclosedcaption` (gst-plugins-rs ≥ 0.14) must still be **built from source**. + - Tests: data launch string, adapter chain + preserve policy, `:Data` group hint. +- **Still owed (needs HW):** build `rsclosedcaption` into the caption image, then run the + round-trip (subtitle → data flow → `st2038anctocc` decode) on the lab box and confirm a + reader recovers the captions. Batch with the next HW session. + +### 3.2 — Timed Data (event flows) readiness ✅ DONE (tracking + design) +- **Why:** PR #720 (`MXL_DATA_FORMAT_EVENT`) is review-complete and near merge — + a registry-typed, strongly-timestamped home for SCTE-104/35, tally, ADM metadata; + #731 makes it fabric-portable. Our guest-ingest is a natural first consumer. +- **Shipped:** `docs/TIMED-DATA-READINESS.md` — the PR #720 API surface, v1.1 data flow + (3.1) vs v1.2 event flow mapping, an `essence='event'` guest-sidecar sketch reusing the + seam + IN-005 registry, and the blocking deps (#720 merge, #730 bindings, MXL URN + registry scheme). Tracking-only until upstream lands. + +### 3.3 — IS-05 connection shim on our NMOS node ✅ DONE +- **Why:** `tools/nmos_node.py` was IS-04 discovery-only; its docstring called IS-05 "the + next build." NVIDIA's `gst-nmos-rs` confirms IS-05 is tractable but needs `nvnmosd` and + is NOT our MXL transport — blueprint only. Built ourselves on our BCP-007-03 node. +- **Shipped:** `/x-nmos/connection/v1.2/` Connection API in `nmos_node.py` (flow identity in + `transport_params`, disable/re-enable, domain-UUID resolution — see ADR-001). Receivers are + the switcher's input slots, so **PATCH `single/receivers//staged` with + `activate_immediate` → a real selector cut** (`POST /api/mxl/input {slot:N}` on the + facility) and promotes staged→active. Senders read-only (MXL has no transport file — + BCP-007-03). Device advertises an `sr-ctrl/v1.2` control. Mutation gated by + `MXL_CONTROL_TOKEN` (same as the facility). 8 tests drive the real server end-to-end + (tree, staged read, activate→cut, route-by-transport_params, disable, stage-without-activate, + sender-405). Scheduled activation → 501 (immediate only). This turns the node from *discoverable* into + *routable* by a standard NMOS controller. + +### 3.4 — Watch #232 (GStreamer-plugin alpha support) for the keyer +- Backlog item, not ours to build — alpha in the gst plugin benefits the html5-keyer + path. Track only. + +--- + +## Fabrics (separate lane — reference, no switcher work implied) + +`tools/mxl-fabrics-demo/demo.cpp` clarified the real Fabrics API: +**INITIATOR = sender** (holds an MXL *reader*, pushes grains via +`mxlFabricsInitiatorTransferGrain/Samples`); **TARGET = receiver** (holds an MXL +*writer*, exposes buffers, emits a base64 `TargetInfo`). Provider auto-select is +**EFA > VERBS > TCP > SHM**. This confirms our existing proxy/NAT approach +(`tools/patch-target-ip.py` rewrites the sockaddr inside the base64 `TargetInfo`) is +correct. Any future "fan a guest out to a second node" work rides #731 (Timed Data over +fabric) once it lands. + +--- + +## Sequencing + +1. **Tier 1.1 + 1.2 + 1.3** in one HW session (restamp rework is the headline; verify + SDK + fix the audio floor while the rig is up). +2. **Tier 2.1 + 2.2** (ingress registry + soak criterion) — small, alongside/after. +3. **Tier 3.1 (ANC/CC)** as the next real feature. +4. **Tier 3.2 / 3.3** tracked; act when #720 merges / when IS-05 is prioritized. diff --git a/docs/TIMED-DATA-READINESS.md b/docs/TIMED-DATA-READINESS.md new file mode 100644 index 0000000..53b5931 --- /dev/null +++ b/docs/TIMED-DATA-READINESS.md @@ -0,0 +1,89 @@ + + + +# Timed Data (MXL v1.2 event flows) — switcher readiness note + +**Status:** tracking + design sketch. No code yet; the upstream feature is not merged. +This is Tier 3.2 of `docs/ROADMAP-FROM-SPECS-2026-10.md` — get ahead of the v1.2 +"Timed Data" feature so we are the reference consumer the day it lands. + +## What's coming upstream + +AMWA **IN-... / dmf-mxl #327** "Exchange of Timed Data through MXL" adds a new flow +format for **non-periodic, strongly-timestamped, variable-size, self-describing** grains — +the shared-memory analog of SMPTE **ST 2110-41** (FMX). Implementation is **PR #720** +(`dmf-mxl/mxl`, review-complete, near merge as of Oct 2026), with two carved-out child +tasks: **#731** (mxl-fabrics transport) and **#730** (Rust bindings). + +### The v1.2 API surface (from PR #720, for when we build against it) + +New flow format `MXL_DATA_FORMAT_EVENT` (`urn:x-nmos:format:event`). A **double-ended +queue where the index does NOT imply a timestamp** — the timestamp lives in the event +struct, not in the ring position (the key departure from video/audio flows). + +- Writer: `mxlFlowWriterOpenEvent` → edit `mxlEventInfo` + payload → `mxlFlowWriterCommitEvent` + (or `...CancelEvent`). One event open at a time. +- Reader: `mxlFlowReaderGetEvent(timeoutNs, …)` / `...GetEventNonBlocking`. One producer, + N independent readers (each its own cursor/instance). +- `mxlEventInfo` (512 B, v1): `timestamp` (TAI ns, **must be non-decreasing**), + `registryType` + `dataItemType[256]` (SMPTE ST 2110-41 Admin Register, or the DMF MXL + URN scheme `x-mxl:…` — the MXL scheme is still TODO upstream), `eventSize`/`offset`/`complete` + (manual fragmentation; interleaving logical events is forbidden), `index` (read-assigned). +- Capacity from grain_rate × history (default 200 ms), clamp 2..65536. +- Ships `mxl-data-probe` + ST 2110-41 ADM/DIT test pcaps. + +## How it maps onto this switcher + +We already have the **v1.1 path** for the same payloads: the `data` essence shipped in +Tier 3.1 (`CaptionFileAdapter` → `meta/x-st-2038 ↔ video/smpte291` data flow via +`mxlsink`). That carries CEA-608/708 captions, and could carry SCTE-104/35, tally, ADM +metadata — frame-aligned, inside an ST-2038 wrapper. + +Event flows are the **v1.2 path** for the *same information* but modelled natively: +aperiodic, per-event timestamped, type-registered — no ST-2038 framing needed, and +(via #731) fabric-portable to a second host. The two are complementary: + +| | v1.1 data flow (shipped, Tier 3.1) | v1.2 event flow (#327/#720, pending) | +|---|---|---| +| MXL format | `video/smpte291` (ST-2038 ANC) | `MXL_DATA_FORMAT_EVENT` | +| Cadence | frame-aligned (periodic grid) | aperiodic, event-timestamped | +| Typing | ANC DID/SDID inside 2038 | DIT + external registry (SMPTE / MXL URN) | +| GStreamer | `cctost2038anc`/`st2038anctocc` exist today | no gst element yet (C API first) | +| Our seam | `essence='data'` adapter + core branch | a new `essence='event'` sidecar (below) | + +## Design sketch: a guest event-flow sidecar + +When #720 lands, the smallest useful consumer is a **sidecar alongside a guest's A/V** +that publishes/reads an event flow carrying that guest's timed data (captions from an ASR +feed, SCTE triggers, tally). It reuses the seam's shape: + +- A new `essence='event'` in `contribution_core` — but event flows do **not** go through + GStreamer `mxlsink` (no gst element yet), so this is NOT a launch-string branch. Instead a + small Python loop using the v1.2 **Rust/C bindings** (#730): open the event writer, and on + each incoming datum set `timestamp` (our local TAI, same clock the restamp locks), + `registryType`/`dataItemType`, copy the payload, commit. +- Timestamp discipline is the natural fit for our existing model: events are stamped on the + **same local clock** the A/V restamp locks to, so a caption event and its video grain + share one time base — exactly the lip-sync invariant the A/V seam already maintains. +- The IN-005 ingress registry (Tier 2.1) extends cleanly: an event-flow ingest writes the + same `/tmp/mxl-ingress/.json` record (provenance = registry/DIT, timing = the + per-event timestamps), so `ingress-soak.sh` can watch it too. + +## Blocking dependencies before we build + +1. **#720 merged** + a tagged release that includes it (we pin the SDK by base-image + digest — see `docs/VERSIONS.md`; the caption/event build is gated behind an opt-in arg). +2. **#730 (Rust bindings)** — our tools are Python-over-gst or Python-over-bindings; the + event API is C/Rust first. Until bindings exist, a prototype would be a small C or Rust + helper, not a gst pipeline. +3. The **MXL URN registry scheme** (TODO in #720) if we want DMF-native typing rather than + borrowing the SMPTE 2110-41 register. + +## Next actions (tracking only) + +- [ ] Watch dmf-mxl **#720 / #731 / #730** for merge; re-read the final API before building. +- [ ] When bindings land, prototype the `essence='event'` sidecar above against a guest. +- [ ] Decide registry: SMPTE 2110-41 Admin Register vs. the DMF MXL URN scheme. + +Ties: `docs/ROADMAP-FROM-SPECS-2026-10.md` (Tier 3.1 data flow = the v1.1 sibling), +`docs/CONTRIBUTION-SEAM.md` (the adapter/essence model this extends). diff --git a/docs/VERSIONS.md b/docs/VERSIONS.md index e73a56c..5bc1f9e 100644 --- a/docs/VERSIONS.md +++ b/docs/VERSIONS.md @@ -24,6 +24,41 @@ These values live in one place in the code: the `PIN_*` map at the top of `scripts/quickstart.sh`. `docker/guest-ingest.Dockerfile` pins its base (`test-generator`) via a build arg that defaults to the same digest. +## MXL SDK version (inside the base image) + +The MXL SDK (`libmxl.so` + the `mxlsink`/`mxlsrc` GStreamer plugins) is **not +built by this repo** — it ships *inside* the pinned `ghcr.io/cbcrc/test-generator` +base digest above, which `docker/guest-ingest.Dockerfile` builds `FROM`. So the +effective SDK version is whatever that digest baked in: + +| Component | Pinned to | Expected version | +|---|---|---| +| MXL SDK (Flow + Fabric API) | the `test-generator` base digest `sha256:09cad09…` | **v1.1.0** (released 2026-09-09; Fabric API) | +| GStreamer | — | 1.24.x | + +Why this matters: a future base-digest refresh could silently change the SDK +version. The Flow API `mxlsink` behaviour we depend on — one shared clock offset +`D` sampled per pipeline, `index = timestamp_to_index(pts + base_time + D)`, and +**no writer-side backward-index guard** (the writer requires the caller to present +monotonic PTS; that is exactly what `contribution_core._restamp`'s monotonic clamp +provides) — is the v1.1.0 `gst-mxl-rs` contract. Pin and verify it. + +**Verify the SDK version in the pinned base before a cold-clone proof:** + +```bash +IMG=ghcr.io/cbcrc/test-generator@sha256:09cad0981475095ab948ca51511d4fbdc0521e2a23632d50abaf14fc3847cd92 +# plugins present + their version +docker run --rm --entrypoint sh "$IMG" -c 'gst-inspect-1.0 mxlsink | grep -iE "Version|Filename"; gst-inspect-1.0 mxlsrc >/dev/null && echo mxlsrc=OK' +# libmxl SONAME / version strings +docker run --rm --entrypoint sh "$IMG" -c 'f=$(find / -name "libmxl*.so*" 2>/dev/null | head -1); echo "$f"; strings "$f" 2>/dev/null | grep -iE "^1\.[01]\.[0-9]+$|v1\.[01]" | head' +``` + +If that reports anything other than v1.1.0, do **not** treat a green run as +v1.1.0-conformant — update this table and re-run the HW proof. (The standalone +fabric build on the GCP/Azure boxes already pins the SDK explicitly: +`git clone dmf-mxl/mxl && git checkout v1.1.0` — see +`docs/JONAS-FABRIC-HANDOFF.md`.) + ## Bleeding-edge mode (opt-in) To test against the current upstream `:latest` instead of the pinned digests — diff --git a/docs/changelog.html b/docs/changelog.html index eccecd3..af6181a 100644 --- a/docs/changelog.html +++ b/docs/changelog.html @@ -34,7 +34,18 @@

Running log

What's new in MXL

A dated log of Media eXchange Layer milestones — in the ecosystem and in this lab. New entries at the top. Suggest one by opening an issue on GitHub.

-

Office Hours GlobalLast updated 2026-10-01

+

Office Hours GlobalLast updated 2026-10-09

+ +

Oct 2026The whole switcher, on one $400 box

+

We moved the entire facility off cloud VMs and onto a single second-hand dual-Xeon Supermicro bought for about $400 — and ran a live multi-camera show from it. Everything that used to need three cloud instances now fits in one commodity box: all source ingest, the shared-memory domain, the cut (input-selector), the H.264 encoder, WebRTC output, and the control plane. Four live, switchable sources at once — a test pattern, a Makito hardware encoder, a PTZ camera, and a phone streaming SRT over Wi-Fi — all cuttable from a browser. Notes from the build:

+
    +
  • Commodity silicon is enough. MXL needs AVX2, not AVX-512 — so a cheap older Xeon runs the full stack. No GPU, no specialty NIC, no cloud bill.
    • +
  • Remote TD, anywhere. The control surface ("Skin") is served over a secure tunnel, so a technical director on any network can watch program and cut the show from a browser — the box stays behind the firewall.
    • +
  • Reboot-proof. A self-heal step brings the whole program chain back in the right order after a cold boot — wire the selector, cut to a live source, re-publish program, then verify it's actually advancing. Proven across repeated full reboots with zero hands on keyboard.
    • +
+ +

Oct 2026Correcting the record on EFA — it was the hardware, not the proxy

+

Earlier we reported that RDMA-over-EFA fabric transport failed in the cloud. The real cause turned out to be the instance type: our test ran on AWS Nitro v3 (no remote-write RDMA). Re-running on a Nitro v4 instance with EFA-direct, the RDMA target ran clean — no errors. The earlier "EFA failed" finding was an instance-type mismatch, not a bug in the proxy. The TCP fabric path remains the portable default; EFA is a real option on the right hardware.

Oct 2026AMWA publishes IN-001 — a control model for MXL

AMWA published IN-001, "API Requirements – Control of the Media eXchange Layer (MXL)" — the draft that starts defining how an external controller manages MXL readers and writers (MXL itself defines media exchange, not control). It's an open Increment soliciting public review. We wrote up what it is and mapped our own switcher's operations against it. Read it →

diff --git a/docs/index.html b/docs/index.html index 975a3ce..1067ee0 100644 --- a/docs/index.html +++ b/docs/index.html @@ -85,7 +85,7 @@

OPEN SOURCE / SOFTWARE-DEFINED LIVE PRODUCTION

What if the switcher
was shared memory?

-

A working exploration of a live production control room built around the Media eXchange Layer: real cameras, remote guests, playout, layouts, graphics, audio, multiview, recording and standard control — assembled as software on general-purpose cloud VMs.

+

A working exploration of a live production control room built around the Media eXchange Layer: real cameras, remote guests, playout, layouts, graphics, audio, multiview, recording and standard control — assembled as software that runs anywhere from a $400 commodity box to the cloud.

See it in action Explore the source @@ -225,6 +225,7 @@

A switcher becomes a set of replaceable media functions.

10.5 GBLive video moved through the browser — HLS segments and scrub thumbnails from the running facility4.9 GB HLS · 3.0 GB frames
18Days on air before the live lab went dark and this archive took over — self-standing, no VMsSep 13 → Sep 30
1080p3010-bit v210 inside the shared-memory domain, encoded once for every viewerENCODE ONCE
+
$400The whole facility now runs on one second-hand Supermicro — ingest, the cut, encode and control in a single commodity boxAVX2 · NO GPU · REMOTE TD
diff --git a/tests/test_contribution_core.py b/tests/test_contribution_core.py index 569ac35..18e8a90 100644 --- a/tests/test_contribution_core.py +++ b/tests/test_contribution_core.py @@ -10,6 +10,7 @@ """ import contribution_core as cc from adapters import ( + CaptionFileAdapter, MakitoAdapter, RtspCamAdapter, SrtGuestAdapter, @@ -26,10 +27,13 @@ def test_margin_is_exactly_two_grains(): assert abs(cc.MARGIN_NS - 2 * one_grain_ns) < 1_000_000 # ~2 grains, rounding slack -def test_resync_requires_sustained_drift(): - # A momentary hiccup must NOT yank the offset — re-sync only after N consecutive. - assert cc.RESYNC_COUNT >= 2 - assert cc.RESYNC_NS > cc.MARGIN_NS +def test_no_dead_slew_servo_constants(): + # The restamp is a local wall-clock grid + drop-ahead-of-realtime (HW Oct-8), NOT a slew + # servo. The old SLEW_GAIN/SLEW_MAX_NS/HARD_RELOCK_NS constants were removed when the grid + # replaced the servo; guard against them creeping back as dead config that misleads tuning. + for dead in ("SLEW_GAIN", "SLEW_MAX_NS", "HARD_RELOCK_NS"): + assert not hasattr(cc, dead), f"{dead} is dead after the grid rewrite — remove it" + assert cc.FRAME_NS == 33_333_333 # still the grain-step fallback (30fps) def test_canon_caps_are_v210_1080p30_progressive(): @@ -250,25 +254,274 @@ def test_persistent_init_builds_tail_then_source_leg(monkeypatch): assert "mxlsink" not in _Gst.last_bins[1] -def test_monotonic_relock_never_rewinds_the_flow(monkeypatch): - """The core of the R1 fix: on a leg rebuild the new source PTS restarts near 0, - so the restamp must RE-LOCK the offset forward of the last grain already written — - never rewind the flow (a rewind is exactly the selector's 'grain too early' read).""" +def test_restamp_relock_never_rewinds_the_flow(monkeypatch): + """R1 invariant, re-verified against the ACTUAL grid restamp (not the removed offset + clamp): on a persistent-flow leg rebuild the flow already has grains up to last_mapped; + the next emitted grain must land STRICTLY AFTER last_mapped even when wall-clock `now` is + far BEHIND it (fresh-session clock 0.5s in, flow already 10s long). A rewind is exactly + the selector's 'grain too early' read + mxlsink's -5 (no backward-index guard). + + This drives _restamp directly; the grid's `max(last_mapped + grain_ns, now + MARGIN)` is + what must provide the guarantee now.""" + import contribution_core as _cc monkeypatch.setenv("MXL_INGEST_PERSISTENT", "1") core = cc.ContributionCore(_persist_adapter(), repair_url="none") s = core.state - # simulate: we already wrote up to last_mapped, then a rebuild cleared the lock s["last_mapped"] = 10_000_000_000 # 10s of flow already written - s["offset"] = None + s["offset"] = None # a rebuild cleared the lock s["relock"] = True - # the new leg's first frame arrives with a small PTS (fresh session) and a clock - # time that would map it BACKWARD under a naive now-based lock - new_leg_pts = 0 - now = 500_000_000 # clock only 0.5s in (well before last_mapped) - # replicate the clamp the probe applies on re-lock - offset = now - new_leg_pts + cc.MARGIN_NS - min_pts = s["last_mapped"] + cc.FRAME_NS - if new_leg_pts + offset < min_pts: - offset = min_pts - new_leg_pts - mapped = new_leg_pts + offset - assert mapped >= s["last_mapped"] + cc.FRAME_NS # strictly forward — no rewind + clock = _ProbeClock(500_000_000) # clock only 0.5s in — WELL behind last_mapped + pad = _FakePad(_FakeCaps(30, 1)) + monkeypatch.setattr(core.pipe, "get_clock", lambda: clock) + monkeypatch.setattr(core.pipe, "get_base_time", lambda: 0) + # the new leg's first frame arrives with a small PTS (fresh session) + mapped = _drive_restamp(core, pad, clock, [0])[0] + assert mapped is not None, "first relock frame should be emitted, not dropped" + assert mapped >= s["last_mapped"] # strictly forward — the flow never rewinds + # (and 'relock' was consumed — a once-per-lock trigger) + assert core.state.get("relock") is False + + +# ── _grain_ns: the monotonic step must be RATE-DERIVED, not a 30fps constant ────── +# Oct 7 2026 fix: a 60fps source's floor was using FRAME_NS (33.3ms), a step 2x too +# large, which surfaced as the Oct-6 `err=-41s / fps=173`. The step must come from the +# pad's NEGOTIATED framerate. These use duck-typed caps/pad — no real GStreamer needed. +class _FakeStructure: + def __init__(self, num, den): + self._fr = (num, den) + + def get_fraction(self, name): + assert name == "framerate" + if self._fr is None: + return (False, 0, 0) + return (True, self._fr[0], self._fr[1]) + + +class _FakeCaps: + def __init__(self, num=None, den=1): + self._st = _FakeStructure(num, den) if num is not None else None + + def get_size(self): + return 1 if self._st is not None else 0 + + def get_structure(self, i): + return self._st + + def to_string(self): + if self._st is None: + return "video/x-raw" + n, d = self._st._fr + return f"video/x-raw, format=(string)v210, framerate=(fraction){n}/{d}" + + +class _FakePad: + def __init__(self, caps): + self._caps = caps + + def get_current_caps(self): + return self._caps + + +def test_grain_ns_is_rate_derived(): + core = cc.ContributionCore(_persist_adapter(), repair_url="none") + # 30fps (the canonical conform rate) == FRAME_NS + assert core._grain_ns(_FakePad(_FakeCaps(30, 1))) == cc.FRAME_NS + # 60fps => half the step (the exact bug: floor was 2x too large) + assert core._grain_ns(_FakePad(_FakeCaps(60, 1))) == round(1_000_000_000 / 60) + # 59.94 (60000/1001) => correct fractional grain + assert core._grain_ns(_FakePad(_FakeCaps(60000, 1001))) == round(1001 * 1_000_000_000 / 60000) + # 50fps (EU) => 20ms + assert core._grain_ns(_FakePad(_FakeCaps(50, 1))) == 20_000_000 + + +def test_grain_ns_falls_back_to_frame_ns_without_framerate(): + core = cc.ContributionCore(_persist_adapter(), repair_url="none") + assert core._grain_ns(_FakePad(_FakeCaps(None))) == cc.FRAME_NS # empty caps + assert core._grain_ns(_FakePad(None)) == cc.FRAME_NS # no caps yet + + +# ── IN-005 ingress registry: the restamp must leave a TRACEABLE per-flow record ─── +def test_ingress_record_captures_provenance_and_timing(tmp_path, monkeypatch): + import json + monkeypatch.setenv("MXL_INGRESS_DIR", str(tmp_path)) + monkeypatch.setenv("MXL_GUEST_TRANSPORT", "srt-listen") + core = cc.ContributionCore(_persist_adapter(), repair_url="none") + core.state["offset"] = 66_000_000 + core.state["n"] = 1200 + core._write_ingress_record(_FakePad(_FakeCaps(30, 1)), event="diag", + err_ns=-4_700_000, grain_ns=cc.FRAME_NS) + files = list(tmp_path.glob("*.json")) + assert len(files) == 1 + rec = json.loads(files[0].read_text()) + # provenance (IN-005: identify the signal + its source timing) + assert rec["transport"] == "srt-listen" + assert rec["source_caps"] and "framerate=(fraction)30/1" in rec["source_caps"] + assert rec["essence"] == "video" + # timing adjustments (IN-005: traceable offsets) + assert rec["offset_ms"] == 66.0 + assert rec["err_ms"] == -4.7 + assert rec["grain_ns"] == cc.FRAME_NS + assert rec["event"] == "diag" and rec["frames"] == 1200 + + +def test_ingress_record_disabled_by_empty_dir(tmp_path, monkeypatch): + # An adopter can opt OUT (no file written) by clearing MXL_INGRESS_DIR. + monkeypatch.setenv("MXL_INGRESS_DIR", "") + core = cc.ContributionCore(_persist_adapter(), repair_url="none") + assert core._ingress_path is None + core._write_ingress_record(_FakePad(_FakeCaps(30, 1)), event="locked") # must not raise + assert list(tmp_path.glob("*.json")) == [] + + +# ── Data essence: ANC / closed-caption (video/smpte291) data flow (Tier 3.1) ────── +def test_caption_adapter_is_data_essence_and_preserve(): + a = CaptionFileAdapter(srt_file="show.srt", flow_id="d" * 36, label="Captions") + assert a.essence == "data" + assert a.timing_policy == "preserve" # ST-2038 grains are frame-aligned; no restamp + assert a.group_hint == "Captions:Data" # group_hint knows the data essence + + +def test_caption_adapter_source_fragment_is_the_st2038_chain(): + a = CaptionFileAdapter(srt_file="show.srt", flow_id="d" * 36) + frag = a.source_fragment() + # mirrors the gst-mxl-rs v1.1.0 README producer chain + for tok in ("filesrc location=show.srt", "subparse", "tttocea608", + "ccconverter", "closedcaption/x-cea-608", "cctost2038anc", + "meta/x-st-2038"): + assert tok in frag, frag + assert "30000/1001" in frag # README default / NTSC 608 framerate + + +def test_data_launch_is_st2038_to_mxlsink_no_video_conform(): + a = CaptionFileAdapter(srt_file="show.srt", flow_id="d" * 36, label="CC") + launch = _core(a).pipe._launch + assert "meta/x-st-2038,alignment=frame" in launch # canonical data caps enforced + assert "mxlsink" in launch + # a data flow must NOT get the video conform transforms + assert "videorate" not in launch + assert "videoconvert" not in launch + assert "format=v210" not in launch + assert "audioconvert" not in launch + + +# ── _restamp probe: drive it directly (HW Oct-8 caught a KeyError the launch-string ── +# tests never would — the state dict had no 'last_mapped' key). These fake a pad/buffer/ +# clock and call the probe, asserting: no crash, monotonic PTS on the local grid, and +# that ahead-of-realtime burst frames are DROPPED rather than marching PTS forward. +class _ProbeBuf: + def __init__(self, pts): + self.pts = pts + self.duration = 0 + def get_buffer(self): + return self + + +class _ProbeClock: + def __init__(self, t=0): + self.t = t + def get_time(self): + return self.t + + +def _drive_restamp(core, pad, clock, pts_list): + """Feed a sequence of source PTS through _restamp; return the mapped PTS list + (None where the buffer was dropped). Advances the clock one grain per CALL.""" + import contribution_core as _cc + out = [] + for pts in pts_list: + buf = _ProbeBuf(pts) + + class _Info: + def get_buffer(_s): + return buf + ret = core._restamp(pad, _Info()) + out.append(None if ret == _cc.Gst.PadProbeReturn.DROP else buf.pts) + clock.t += _cc.FRAME_NS # wall-clock advances one grain per frame (realtime 30fps) + return out + + +def test_restamp_probe_no_keyerror_and_monotonic(monkeypatch): + core = cc.ContributionCore(_persist_adapter(), repair_url="none") + clock = _ProbeClock(10_000_000_000) # arbitrary non-zero "now" + pad = _FakePad(_FakeCaps(30, 1)) # 30fps negotiated caps (post-videorate) + monkeypatch.setattr(core.pipe, "get_clock", lambda: clock) + monkeypatch.setattr(core.pipe, "get_base_time", lambda: 0) + # realtime-paced source: PTS advances one grain per frame + pts = [i * cc.FRAME_NS for i in range(10)] + mapped = _drive_restamp(core, pad, clock, pts) + kept = [m for m in mapped if m is not None] + assert kept, "all frames dropped — grid never emitted" + # strictly monotonic, one grain apart (the local grid) + for a, b in zip(kept, kept[1:]): + assert b > a, (a, b) + assert b - a == cc.FRAME_NS + + +def test_restamp_probe_drops_ahead_of_realtime_burst(monkeypatch): + core = cc.ContributionCore(_persist_adapter(), repair_url="none") + clock = _ProbeClock(10_000_000_000) + pad = _FakePad(_FakeCaps(30, 1)) + monkeypatch.setattr(core.pipe, "get_clock", lambda: clock) + monkeypatch.setattr(core.pipe, "get_base_time", lambda: 0) + import contribution_core as _cc + + # lock on the first frame + _drive_restamp(core, pad, clock, [0]) + # now a BURST: 5 buffers arrive with NO wall-clock advance (freeze the clock) + burst_dropped = 0 + for i in range(1, 6): + buf = _ProbeBuf(i * cc.FRAME_NS) + + class _Info: + def get_buffer(_s): + return buf + if core._restamp(pad, _Info()) == _cc.Gst.PadProbeReturn.DROP: + burst_dropped += 1 + # clock does NOT advance — simulates a burst faster than realtime + assert burst_dropped >= 1, "burst frames should be dropped, not stamped ahead of now" + assert core.state.get("dropped", 0) >= 1 + + +# ── _restamp_audio: monotonic PTS guard (HW Oct-8: audio -5 restart loop) ───────── +# The bug: next_pts advances by buf.duration per buffer; on bursty fan-out delivery it races +# ahead, then the >500ms re-anchor jumped it BACK to now+MARGIN → non-monotonic PTS → mxlsink +# -5 → restart loop. The fix floors the re-anchor at last_mapped+quantum (never rewind). +def _drive_audio(core, pad, clock, buffers): + """Feed (pts_ignored, duration_ns) audio buffers through _restamp_audio; return mapped PTS. + The caller advances clock.t between buffers to simulate burst vs realtime.""" + out = [] + for dur in buffers: + b = _ProbeBuf(0) + b.duration = dur + + class _Info: + def get_buffer(_s): + return b + core._restamp_audio(pad, _Info()) + out.append(b.pts) + return out + + +def test_restamp_audio_pts_strictly_monotonic_through_burst_and_reanchor(monkeypatch): + core = cc.ContributionCore(_persist_adapter(), repair_url="none") + clock = _ProbeClock(10_000_000_000) + pad = _FakePad(_FakeCaps(48000, 1)) # audio caps (framerate irrelevant for audio path) + monkeypatch.setattr(core.pipe, "get_clock", lambda: clock) + monkeypatch.setattr(core.pipe, "get_base_time", lambda: 0) + AAC = 21_333_333 # ~1024 samples @48k ≈ 21.3ms per AAC frame + + out = [] + # 1) BURST: 60 AAC frames arrive with the clock FROZEN (jitter-buffer flush) → next_pts + # races ~1.28s ahead of `now`. (Exactly the HW case: the fan-out flushes a burst.) + out += _drive_audio(core, pad, clock, [AAC] * 60) + # 2) resume realtime pacing: the clock now advances ~one AAC frame per buffer. Because + # next_pts raced ahead, |now+MARGIN - next_pts| > 500ms on the NEXT frame → the OLD code + # re-anchored next_pts BACK to now+MARGIN (a REWIND → mxlsink -5). The fix floors it. + for _ in range(8): + clock.t += AAC + out += _drive_audio(core, pad, clock, [AAC]) + + assert all(x is not None for x in out) + # THE invariant the -5 violated: PTS must be strictly non-decreasing throughout. + for a, b in zip(out, out[1:]): + assert b >= a, f"audio PTS rewound {a} -> {b} (would be mxlsink -5)" diff --git a/tests/test_nmos_is05.py b/tests/test_nmos_is05.py new file mode 100644 index 0000000..c9ca300 --- /dev/null +++ b/tests/test_nmos_is05.py @@ -0,0 +1,221 @@ +"""IS-05 Connection API shim on the NMOS node (Tier 3.3). + +Drives the real stdlib HTTP server on a loopback port, with the 'facility' stubbed by +a second local server that records the cut. Asserts: the connection tree is exposed, +a receiver's /staged is readable, and a PATCH with activate_immediate both (a) POSTs +/api/mxl/input {slot:N} to the facility and (b) promotes staged -> active. + +No GStreamer, no real MXL domain — nmos_node is stdlib-only. +""" +import json +import sys +import threading +import time +import urllib.request +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "tools")) +import nmos_node as nm # noqa: E402 + + +def _free_port(): + import socket + s = socket.socket() + s.bind(("127.0.0.1", 0)) + p = s.getsockname()[1] + s.close() + return p + + +class _FacilityStub(BaseHTTPRequestHandler): + cuts = [] # class-level record of {slot:...} bodies POSTed to /api/mxl/input + + def log_message(self, *a): + pass + + def do_POST(self): + n = int(self.headers.get("Content-Length", 0)) + body = json.loads(self.rfile.read(n) or b"{}") + if self.path.rstrip("/") == "/api/mxl/input": + _FacilityStub.cuts.append(body) + self.send_response(200) + self.end_headers() + self.wfile.write(b'{"ok":true}') + else: + self.send_response(404) + self.end_headers() + + def do_GET(self): + # the node's poll_facility hits /api/mxl/status; return an empty-ish state + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(b'{"input":null,"pvw":null,"slots":[]}') + + +def _get(url): + with urllib.request.urlopen(url, timeout=5) as r: + return r.status, json.loads(r.read()) + + +def _patch(url, body): + req = urllib.request.Request(url, data=json.dumps(body).encode(), + headers={"Content-Type": "application/json"}, + method="PATCH") + with urllib.request.urlopen(req, timeout=5) as r: + return r.status, json.loads(r.read()) + + +def _start_node(): + """Boot a node + facility stub; return (base_url, facility_stub_server, node_server).""" + _FacilityStub.cuts = [] + fport = _free_port() + fac = ThreadingHTTPServer(("127.0.0.1", fport), _FacilityStub) + threading.Thread(target=fac.serve_forever, daemon=True).start() + + nport = _free_port() + + class Args: + port = nport + href = f"http://127.0.0.1:{nport}/" + registry = None + facility = f"http://127.0.0.1:{fport}" + + nm.MODEL = nm.Model(Args()) + node = ThreadingHTTPServer(("127.0.0.1", nport), nm.Handler) + threading.Thread(target=node.serve_forever, daemon=True).start() + time.sleep(0.2) + return f"http://127.0.0.1:{nport}", fac, node + + +def test_connection_api_tree_and_receivers_listed(): + base, fac, node = _start_node() + try: + code, root = _get(base + "/x-nmos") + assert code == 200 and "connection/" in root + code, single = _get(base + "/x-nmos/connection/v1.2/single") + assert "receivers/" in single and "senders/" in single + code, recvs = _get(base + "/x-nmos/connection/v1.2/single/receivers/") + assert code == 200 and len(recvs) == len(nm.SLOT_TO_SENDER) + finally: + node.shutdown(); fac.shutdown() + + +def test_receiver_staged_readable_and_defaults_disabled(): + base, fac, node = _start_node() + try: + rid = nm.u5("receiver", "slot2") + code, staged = _get(base + f"/x-nmos/connection/v1.2/single/receivers/{rid}/staged") + assert code == 200 + assert staged["master_enable"] is False + assert staged["activation"]["mode"] is None + finally: + node.shutdown(); fac.shutdown() + + +def test_patch_activate_immediate_cuts_program_to_that_slot(): + base, fac, node = _start_node() + try: + rid = nm.u5("receiver", "slot3") # slot 3 = cam2 + sid = nm.u5("sender", "cam2") + url = base + f"/x-nmos/connection/v1.2/single/receivers/{rid}/staged" + code, resp = _patch(url, { + "master_enable": True, + "sender_id": sid, + "activation": {"mode": "activate_immediate"}, + }) + assert code == 200, resp + # the shim translated the activation into a real selector cut to slot 3 + assert _FacilityStub.cuts == [{"slot": 3}], _FacilityStub.cuts + # and promoted staged -> active + code, active = _get(base + f"/x-nmos/connection/v1.2/single/receivers/{rid}/active") + assert active["master_enable"] is True + assert active["activation"]["mode"] == "activate_immediate" + assert active["activation"]["activation_time"] is not None + finally: + node.shutdown(); fac.shutdown() + + +def test_patch_staging_without_activation_does_not_cut(): + base, fac, node = _start_node() + try: + rid = nm.u5("receiver", "slot1") + url = base + f"/x-nmos/connection/v1.2/single/receivers/{rid}/staged" + code, resp = _patch(url, {"master_enable": True, "sender_id": nm.u5("sender", "playout")}) + assert code == 200 + assert _FacilityStub.cuts == [] # staged only — no activation, no cut + code, active = _get(base + f"/x-nmos/connection/v1.2/single/receivers/{rid}/active") + assert active["master_enable"] is False # active unchanged until activated + finally: + node.shutdown(); fac.shutdown() + + +def test_sender_staged_is_read_only(): + base, fac, node = _start_node() + try: + sid = nm.u5("sender", "cam") + req = urllib.request.Request( + base + f"/x-nmos/connection/v1.2/single/senders/{sid}/staged", + data=b"{}", headers={"Content-Type": "application/json"}, method="PATCH") + try: + urllib.request.urlopen(req, timeout=5) + assert False, "sender PATCH should be rejected" + except urllib.error.HTTPError as e: + assert e.code == 405 + finally: + node.shutdown(); fac.shutdown() + + +# ── ADR-001 convergence: MXL flow identity travels in transport_params (v1.2 shape) ── +def test_sender_active_carries_mxl_flow_and_domain_id(): + # A controller reads the Sender's active transport_params to learn what to route. + base, fac, node = _start_node() + try: + sid = nm.u5("sender", "guest1") + code, active = _get(base + f"/x-nmos/connection/v1.2/single/senders/{sid}/active") + tp = active["transport_params"][0] + assert tp["mxl_flow_id"] == nm.SENDER_FLOWS["guest1"][0] # the real MXL flow uuid + assert "mxl_domain_id" in tp # domain id present (may be str/uuid) + finally: + node.shutdown(); fac.shutdown() + + +def test_patch_routes_by_transport_params_flow_id(): + # The standard IS-05 flow: copy the sender's transport_params onto the receiver, activate. + base, fac, node = _start_node() + try: + rid = nm.u5("receiver", "slot4") # slot 4 = guest1 + flow_id = nm.SENDER_FLOWS["guest1"][0] + url = base + f"/x-nmos/connection/v1.2/single/receivers/{rid}/staged" + code, resp = _patch(url, { + "master_enable": True, + "transport_params": [{"mxl_flow_id": flow_id, "mxl_domain_id": nm.MXL_DOMAIN_ID}], + "activation": {"mode": "activate_immediate"}, + }) + assert code == 200, resp + assert _FacilityStub.cuts == [{"slot": 4}], _FacilityStub.cuts # cut to the guest1 slot + # active reflects the routed flow id + code, active = _get(base + f"/x-nmos/connection/v1.2/single/receivers/{rid}/active") + assert active["transport_params"][0]["mxl_flow_id"] == flow_id + finally: + node.shutdown(); fac.shutdown() + + +def test_patch_disable_is_first_class_and_does_not_cut(): + # IS-05 disable (master_enable=false) must activate cleanly WITHOUT forcing a program cut. + base, fac, node = _start_node() + try: + rid = nm.u5("receiver", "slot5") + url = base + f"/x-nmos/connection/v1.2/single/receivers/{rid}/staged" + code, resp = _patch(url, { + "master_enable": False, + "activation": {"mode": "activate_immediate"}, + }) + assert code == 200, resp + assert _FacilityStub.cuts == [] # disable never cuts program + code, active = _get(base + f"/x-nmos/connection/v1.2/single/receivers/{rid}/active") + assert active["master_enable"] is False + assert active["activation"]["activation_time"] is not None # but it DID activate + finally: + node.shutdown(); fac.shutdown() diff --git a/tests/test_no_hardcoded_hosts.py b/tests/test_no_hardcoded_hosts.py index c054f51..726a057 100644 --- a/tests/test_no_hardcoded_hosts.py +++ b/tests/test_no_hardcoded_hosts.py @@ -35,7 +35,6 @@ "tools/audio_pgm.py": {"prodbots.com"}, # MXL_BACKEND_URL default "tools/layout_pgm.py": {"prodbots.com"}, # MXL_BACKEND_URL default "tools/mxl_multiview.py": {"prodbots.com"}, # MXL_BACKEND_URL default (+ docstring) - "tools/nmos_node.py": {"prodbots.com"}, # MXL_BACKEND_URL default (+ usage) "tools/contribution_core.py": {"prodbots.com"}, # comment example for MXL_REPAIR_URL "tools/mv_encode.py": {"10.0.0.4", "cochran.cloud"}, # MXL_VM1_IP default (+ docstring) "tools/guest-leg-doctor.sh": {"10.0.0.4", "10.0.0.5", "prodbots.com"}, # MXL_VM1_IP/VM2_IP/BACKEND defaults diff --git a/tools/adapters.py b/tools/adapters.py index 28e80b7..70dcd6a 100644 --- a/tools/adapters.py +++ b/tools/adapters.py @@ -323,3 +323,46 @@ def zoomiso_adapters(source_flow_ids, slot_flow_ids, mode='per_participant', label=f'{label_prefix} {i+1}', domain=domain, timing_policy=timing_policy) for i in range(n)] + + +# -- Data / ANC: closed captions as a video/smpte291 MXL data flow (gst-mxl-rs v1.1.0) -- +# The seam's first DATA-essence source. mxlsink turns `meta/x-st-2038,alignment=frame` +# into a `video/smpte291` data flow; this is the path for CEA-608/708 captions, and the +# on-ramp to v1.2 "Timed Data" event flows (dmf-mxl #327). Pipeline mirrors the official +# round-trip example in rust/gst-mxl-rs/README.md @ v1.1.0 (producer side). +# +# REQUIRES (not in the stock cbcrc base image): the `rsclosedcaption` plugin from +# gst-plugins-rs >= 0.14 (cctost2038anc / tttocea608) + `ccconverter` from +# gstreamer1.0-plugins-bad. docker/guest-ingest.Dockerfile must add these before this +# adapter can run on HW (tracked in docs/ROADMAP-FROM-SPECS-2026-10.md Tier 3.1). +class CaptionFileAdapter(SourceAdapter): + """CEA-608 closed captions from a subtitle (.srt) file, wrapped as SMPTE ST 2038 + ANC and written to an MXL `video/smpte291` DATA flow. + + essence='data' routes through the core's data conform branch (just the ST-2038 caps + + a queue -> mxlsink — no video/audio restamp; timing_policy='preserve' because the + ANC wrapper frame-aligns the grains). Front end = the v1.1.0 README producer chain: + filesrc ! subparse ! tttocea608 ! ccconverter ! closedcaption/x-cea-608 ! cctost2038anc + + A real deployment would drive captions live (from an ASR/stenographer feed) rather + than a file; the file source is the reproducible first step and the upstream smoke + test. framerate defaults to 30000/1001 to match the README example and NTSC 608. + """ + essence = 'data' + needs_conform = True # the core appends the ST-2038 caps + queue + mxlsink + timing_policy = 'preserve' # ST-2038 grains carry their own frame alignment + + def __init__(self, srt_file: str, flow_id: str, label: str = 'Captions', + framerate: str = '30000/1001'): + self.srt_file = srt_file + self.flow_id = flow_id + self.label = label + self.framerate = framerate + self.description = f'CEA-608 captions -> ST-2038 data flow ({srt_file})' + + def source_fragment(self) -> str: + # Ends at `meta/x-st-2038` caps WITH the framerate the ANC wrapper needs; the core's + # data branch adds `meta/x-st-2038,alignment=frame ! queue ! mxlsink`. + return (f'filesrc location={self.srt_file} ! subparse ! tttocea608 mode=pop-on ' + f'! ccconverter ! closedcaption/x-cea-608,framerate={self.framerate} ' + f'! cctost2038anc ! meta/x-st-2038,alignment=frame,framerate={self.framerate} ') diff --git a/tools/contribution_core.py b/tools/contribution_core.py index 9ba4e69..79c50bb 100644 --- a/tools/contribution_core.py +++ b/tools/contribution_core.py @@ -12,11 +12,17 @@ scripts. A transport is expressed as a `SourceAdapter`; `ContributionCore.run()` wires it to the canonical conform + restamp + mxlsink + announce and runs the loop. -The restamp probe here is byte-for-byte the one that shipped in cam_ingest.py and -guest_ingest.py — do not "improve" it without re-reading FINDINGS §1/§6: +The restamp probe maps a remote source onto our local grain grid — do not "improve" +it without re-reading FINDINGS §1/§6 and §"Separate the output clock from the input": * MARGIN_NS = 2 grains. Bigger margins starve READERS (166ms => "too late" wedges). - * RESYNC only after RESYNC_COUNT consecutive out-of-band frames, so a momentary - network hiccup never yanks the offset. + * The output is driven by a LOCAL WALL-CLOCK GRID, not by the source PTS (HW Oct-8): + each grain is placed at `max(last_mapped + grain_ns, now + MARGIN_NS)` and any frame + that would land more than one grain ahead of wall-clock is DROPPED. This is monotonic + by construction (mxlsink has NO backward-index guard) and paced to real time. It + replaced an offset-following STEP re-sync and then a proportional SLEW servo — both + churned/ran away on a 60fps source decimated to 30 (the 1000ms SRT jitter buffer + flushes a burst ~1s ahead of the just-started wall clock, so ANY offset-follows-source + design breaks). HW-proven: clean 30fps, err bounded ~5-18ms, 0 restarts. Grain timestamps are RING ADDRESSES, not metadata: a remote source MUST be restamped onto the local clock cadence or it is not cuttable against local flows. @@ -26,6 +32,7 @@ import os import sys import threading +import time import urllib.error import urllib.request from abc import ABC, abstractmethod @@ -36,9 +43,16 @@ # --- canonical cadence constants (FINDINGS §1/§6 — load-bearing, do not tune blindly) --- MARGIN_NS = 66_000_000 # 2 grains @30fps. Bigger => reader starvation ("too late" wedges). -FRAME_NS = 33_333_333 # 1 grain @30fps — the monotonic step used on a persistent re-lock. -RESYNC_NS = 150_000_000 # out-of-band threshold before we consider re-locking the offset -RESYNC_COUNT = 45 # consecutive out-of-band frames required to actually re-sync +FRAME_NS = 33_333_333 # 1 grain @30fps — fallback grain step when caps carry no framerate. +# Gross-discontinuity threshold: if the flow is already more than this far AHEAD of wall-clock, +# the grid re-anchors (continues from last_mapped) instead of drop-guarding — distinguishes a +# reconnect/clock-step (seconds ahead → re-anchor, don't stall) from a jitter-buffer burst +# (sub-second ahead → drop to stay real-time). 2s is above the 1000ms SRT jitter burst and +# below a real leg-rebuild gap. +RELOCK_GAP_NS = 2_000_000_000 +# (The restamp drives a local wall-clock grid + drop-ahead-of-realtime; it does NOT use a +# proportional slew or a hard-relock escape hatch — those were earlier designs the grid +# replaced on HW Oct-8. See _restamp. The SLEW_* constants that lived here are gone.) # --- canonical output formats: the chain's one true grain spec, per essence --- CANON_CAPS = ('video/x-raw,format=v210,width=1920,height=1080,framerate=30/1,' @@ -47,6 +61,11 @@ # contributed audio flow drops straight into the program mix with no reconvert. CANON_AUDIO_CAPS = ('audio/x-raw,format=F32LE,layout=interleaved,rate=48000,' 'channels=2,channel-mask=(bitmask)0x3') +# Data (ANC / timed metadata) canonical grain: SMPTE ST 2038-wrapped ancillary, frame- +# aligned. mxlsink turns `meta/x-st-2038,alignment=frame` into a `video/smpte291` DATA flow +# (gst-mxl-rs v1.1.0). This is the path for CEA-608/708 captions, SCTE-104, etc. — the +# generalization of what v1.2 "Timed Data" (event flows, dmf-mxl #327) will make first-class. +CANON_DATA_CAPS = 'meta/x-st-2038,alignment=frame' DEFAULT_DOMAIN = '/mxl-domain' @@ -119,7 +138,7 @@ def is_native_mxl(self) -> bool: @property def group_hint(self) -> str: - suffix = 'Audio' if self.essence == 'audio' else 'Video' + suffix = {'audio': 'Audio', 'data': 'Data'}.get(self.essence, 'Video') return f'{self.label.replace(" ", "")}:{suffix}' @abstractmethod @@ -154,7 +173,21 @@ def __init__(self, adapter: SourceAdapter, domain: str = DEFAULT_DOMAIN, repair_url = _os.environ.get('MXL_REPAIR_URL', '') self.repair_url = None if repair_url.strip().lower() in ('', 'none') else repair_url self.diag_every = diag_every - self.state = {'offset': None, 'drift_n': 0, 'n': 0, 't0': None} + self.state = {'offset': None, 'n': 0, 't0': None} + # IN-005 ingress registry: AMWA IN-005 ("External Signal Ingress for DMF") asks that + # a workload record each external signal's provenance and the timing adjustments it + # applied, so they are TRACEABLE. We already compute all of it in _restamp; this + # persists it as a per-ingest JSON record (mirrors guest_slot_watcher's /tmp map + # pattern). Empty MXL_INGRESS_DIR disables it (default on so the quickstart writes it; + # a cloned core still writes only to /tmp, no external calls). One file per flow. + self.ingress_dir = _os.environ.get('MXL_INGRESS_DIR', '/tmp/mxl-ingress') + self._ingress_path = (os.path.join(self.ingress_dir, f'{(adapter.flow_id or "src")[:8]}.json') + if self.ingress_dir else None) + if self._ingress_path: # create the dir ONCE here, not on every record write + try: + os.makedirs(self.ingress_dir, exist_ok=True) + except Exception as _e: + print(f'ingress dir setup skipped: {_e}', flush=True) # Persistent-flow mode: adapter property OR env override (env wins when set), # so an adopter can turn it on for an existing adapter without editing it. env_persist = _os.environ.get('MXL_INGEST_PERSISTENT', '').strip().lower() @@ -193,6 +226,14 @@ def _build_launch(self) -> str: return (f'{a.source_fragment()}' f'! audioconvert ! audioresample ! {CANON_AUDIO_CAPS} ' f'! queue max-size-buffers=32 ! {sink}') + if a.essence == 'data': + # Data (ANC) conform: the adapter's source_fragment IS the conform here — it + # ends at `meta/x-st-2038,alignment=frame` (e.g. caption text -> CEA-608 -> + # cctost2038anc). We only enforce the canonical caps + a queue before mxlsink, + # which writes the video/smpte291 data flow. No restamp: ST-2038 grains are + # frame-aligned by the ANC wrapper (adapter sets timing_policy='preserve'). + return (f'{a.source_fragment()}' + f'! {CANON_DATA_CAPS} ! queue max-size-buffers=32 ! {sink}') # Video decode path: front end -> canonical conform -> mxlsink. # videorate reconciles any 30000/1001 (etc.) to exact 30/1 — WITHOUT it the # v210 capsfilter intermittently fails to negotiate (documented crash). It @@ -228,6 +269,8 @@ def _tail_fragment(self) -> str: if a.essence == 'audio': return (f'{jbuf} ! audioconvert ! audioresample ! {CANON_AUDIO_CAPS} ' f'! queue max-size-buffers=32 ! {sink}') + if a.essence == 'data': + return f'{jbuf} ! {CANON_DATA_CAPS} ! queue max-size-buffers=32 ! {sink}' return (f'{jbuf} ! videorate ! videoscale add-borders=true ! videoconvert n-threads=2 ' f'! {CANON_CAPS} ! {sink}') @@ -358,6 +401,66 @@ def _announce(self): print(f'announce err: {e} (attempt {attempt+1})', flush=True) _t.sleep(30) + def _write_ingress_record(self, pad, *, event, err_ns=None, grain_ns=None): + """Persist the IN-005 ingress record for this flow (atomic, best-effort). + + `event` is the lifecycle tag that triggered the write ('locked', 'relocked', + 'hard-relock', 'diag'). Captures signal PROVENANCE (source caps/essence/transport) + and the TIMING ADJUSTMENTS applied (locked offset, running wall-clock error, grain + step) so an operator can see — from one file — what the restamp is doing, which is + exactly what was invisible during the Oct-7 60fps churn. Never raises into the probe.""" + if not self._ingress_path: + return + try: + s = self.state + caps = pad.get_current_caps() + rec = { + 'flow_id': self.a.flow_id, + 'label': self.a.label, + 'essence': self.a.essence, + 'transport': os.environ.get('MXL_GUEST_TRANSPORT', 'n/a'), + 'timing_policy': self.a.timing_policy, + 'event': event, + 'frames': s.get('n', 0), + # timing adjustments (ns and ms for human + machine readers) + 'offset_ns': s.get('offset'), + 'offset_ms': round(s['offset'] / 1e6, 1) if s.get('offset') is not None else None, + 'err_ms': round(err_ns / 1e6, 1) if err_ns is not None else None, + 'grain_ns': grain_ns, + 'dropped': s.get('dropped', 0), # ahead-of-realtime frames dropped (burst guard) — + # the grid servo's real health signal (replaces the + # old relock count; the grid has no relock path). + # provenance: the negotiated source caps on the restamp pad + 'source_caps': caps.to_string() if caps else None, + 'mono_ns': time.monotonic_ns(), # NOT wall time — no Date dependency, just ordering + } + tmp = self._ingress_path + '.tmp' + with open(tmp, 'w') as f: + json.dump(rec, f) + os.replace(tmp, self._ingress_path) # atomic swap — a reader never sees a half-write + except Exception as e: + print(f'ingress-record write skipped: {e}', flush=True) + + def _grain_ns(self, pad): + """The monotonic grain step, derived from the NEGOTIATED output framerate on the + restamp pad — NOT the 30fps `FRAME_NS` constant. + + The probe sits on the sink pad, AFTER the conform stage's + `videorate ! ...framerate=30/1`, so for a conform source this is one grain of + CANON_CAPS (33.3ms). But the servo/clamp must not BAKE IN 30fps: a 60fps source + whose videorate hasn't fully settled can momentarily deliver buffers ~16.6ms + apart. With a hardcoded 30fps clamp those map AHEAD of wall-clock faster than the + grain clock — `mapped` races `now`, `err` runs away (the Oct-6 `err=-41s/fps=173` + on a 60fps camera). Reading the actual framerate makes the step correct for + whatever cadence the pad negotiated (30, 50, 59.94, …). Falls back to FRAME_NS if + caps carry no framerate (e.g. pre-negotiation).""" + caps = pad.get_current_caps() + if caps and caps.get_size() > 0: + ok, num, den = caps.get_structure(0).get_fraction('framerate') + if ok and num > 0: + return round(den * 1_000_000_000 / num) + return FRAME_NS + # --- the cadence-preserving restamp (verbatim; the single most load-bearing idea) --- def _restamp(self, pad, info): # Wait for caps before touching buffers — a dynamic-pad demux (tsdemux on the @@ -370,58 +473,87 @@ def _restamp(self, pad, info): if not clock or buf.pts == Gst.CLOCK_TIME_NONE: return Gst.PadProbeReturn.OK now = clock.get_time() - self.pipe.get_base_time() + grain_ns = self._grain_ns(pad) # rate-derived step (see _grain_ns); 30fps => FRAME_NS s = self.state if s['offset'] is None: + # LOCK ANCHOR — a once-per-lock trigger + IN-005 provenance only. The grid below + # does NOT use s['offset'] for pacing (HW Oct-8: the grid is driven by last_mapped + + # now, see below); offset is recorded purely as "where wall-clock sat at lock" in the + # ingress record. On a persistent-flow RE-LOCK the grid's own monotonic floor + # (max(last_mapped + grain_ns, now + MARGIN)) already guarantees the first new-leg + # grain lands after the last committed one — no offset clamp needed. s['offset'] = now - buf.pts + MARGIN_NS - # On a RE-LOCK (persistent-flow leg rebuild) the flow already has grains up to - # last_mapped; clamp the offset so the first new-leg grain lands STRICTLY after - # that (never rewind the flow — a rewind is exactly the "grain too early" read). - if s.get('relock') and s.get('last_mapped') is not None: - min_pts = s['last_mapped'] + FRAME_NS - if buf.pts + s['offset'] < min_pts: - s['offset'] = min_pts - buf.pts - print(f'cadence offset RE-LOCKED (monotonic): {s["offset"]/1e6:.0f}ms', flush=True) - s['relock'] = False - else: - print(f'cadence offset locked: {s["offset"]/1e6:.0f}ms', flush=True) - if self.a.announce_on_lock and self.repair_url: - threading.Thread(target=self._announce, daemon=True).start() - elif self.a.announce_on_lock: - print('announce skipped (no repair_url) — selector is pre-wired for this slot', flush=True) - mapped = buf.pts + s['offset'] - err = now + MARGIN_NS - mapped - if abs(err) > RESYNC_NS: - s['drift_n'] += 1 - if s['drift_n'] >= RESYNC_COUNT: - s['offset'] += err - s['drift_n'] = 0 - print(f'cadence re-synced by {err/1e6:.0f}ms', flush=True) - mapped = buf.pts + s['offset'] + relock = bool(s.get('relock') and s.get('last_mapped') is not None) + s['relock'] = False + print(f'cadence {"RE-locked" if relock else "locked"} ' + f'(grid-driven; anchor {s["offset"]/1e6:.0f}ms)', flush=True) + self._write_ingress_record(pad, event=('relocked' if relock else 'locked'), + grain_ns=grain_ns) + if not relock and self.a.announce_on_lock and self.repair_url: + threading.Thread(target=self._announce, daemon=True).start() + elif not relock and self.a.announce_on_lock: + print('announce skipped (no repair_url) — selector is pre-wired for this slot', flush=True) + # LOCAL-GRID RESTAMP (HW Oct-8: the offset-follows-source approaches all failed on a + # 60fps camera — the 1000ms SRT jitter buffer releases a BURST whose source PTS is ~1s + # ahead of the just-started wall clock, so `buf.pts + offset` lands >1s off and every + # earlier design — step re-sync, slew servo, monotonic+grain clamp — either churned or + # hard-relock-looped to death). FINDINGS §"Separate the output clock from the input" + # (v-final / flow_stabilizer) is explicit: drive output from a FIXED grain grid SLEWED + # TO THE WALL CLOCK and DROP ahead-of-realtime arrivals. So we stop following buf.pts + # entirely and drive a local grid: + # target = now + MARGIN_NS (where a live grain should sit) + # mapped = max(last_mapped + grain_ns, target) (one grain forward, never behind now) + # This is monotonic by construction (mxlsink has NO backward-index guard — clock.rs — + # so WE must guarantee it), paced to wall-clock (bursts can't race ahead: a frame that + # would land before last_mapped+grain is simply placed on the next grid slot; a frame + # far ahead is pulled back to `target`), and has no offset to drift or re-lock. grain_ns + # is rate-derived so 50/59.94 grids work too. `err` is kept purely as a DIAGNOSTIC of + # how far the raw source PTS sat from the grid (not a control input anymore).""" + target = now + MARGIN_NS + last_mapped = s.get('last_mapped') # not in the initial state dict (HW Oct-8 KeyError) + if last_mapped is None: + mapped = target + elif last_mapped - target > RELOCK_GAP_NS: + # GROSS discontinuity, not a burst: the flow is already far (> RELOCK_GAP_NS) ahead + # of wall-clock. This happens on a persistent-flow leg rebuild (10s of flow written, + # the new leg's clock starts near 0) or a wall-clock step. If we treated this as a + # burst we'd DROP every frame until `now` crawled up to last_mapped — a multi-second + # stall on reconnect (regression caught by test_restamp_relock_never_rewinds). Instead + # CONTINUE the grid from where the flow already is: one grain past last_mapped. The + # flow stays monotonic (never rewinds past the committed head) and resumes immediately. + mapped = last_mapped + grain_ns else: - s['drift_n'] = 0 - # ⚠️ Monotonicity guard (HW Oct 6 2026): a re-sync can swing the offset by a full - # second (seen `cadence re-synced by -1036ms` on bursty SRT, where the latency - # buffer delivers grains ahead of wall-clock). Writing the resulting `mapped` PTS - # unclamped pushes it BACKWARDS past the grain already committed to the flow → - # non-monotonic PTS into v210/mxlsink → "Internal data stream error (-5)" → srtsrc - # dies → restart loop. (Proven: the SAME listener pipeline WITHOUT this probe ran a - # live camera 0-error; re-enabling the unclamped resync reintroduced -5.) The - # re-lock path already forbids rewinding past last_mapped+FRAME_NS; apply the same - # floor to EVERY emitted grain so a resync can only ever nudge the cadence forward, - # never rewind the flow. Drift that genuinely needs catching up is absorbed over - # subsequent grains instead of in one flow-breaking jump. - if s.get('last_mapped') is not None and mapped <= s['last_mapped']: - mapped = s['last_mapped'] + FRAME_NS - s['offset'] = mapped - buf.pts # keep offset consistent with the clamped PTS + nxt = last_mapped + grain_ns + # DROP ahead-of-realtime arrivals: next grid slot is up to one grain beyond where + # wall-clock wants it — a burst/catch-up frame (e.g. the 1000ms SRT jitter buffer + # flushing). Stamping it would march `mapped` ahead of `now` unbounded (the Oct-8 + # -30ms/frame runaway). Drop it; the grid stays locked to wall-clock. (Only reached + # when the flow is NOT grossly ahead — i.e. a genuine burst, not a reconnect.) + if nxt > target + grain_ns: + s['dropped'] = s.get('dropped', 0) + 1 + return Gst.PadProbeReturn.DROP + mapped = max(nxt, target) + # GRID RESIDUAL (the real health signal, HW Oct-8): how far the emitted grain sits from + # where wall-clock wants it. By construction this stays within ~one grain in steady + # state; a value that grows means the grid is drifting ahead of realtime (the drop-guard + # should prevent it). NOT the old source-PTS-vs-offset gap, which was meaningless on a + # grid servo (it read -10s on HW while the flow was a perfectly healthy 30fps grid). + err = mapped - target buf.pts = mapped - s['last_mapped'] = mapped # for a monotonic re-lock on the next leg rebuild + buf.duration = grain_ns # stamp an explicit one-grain duration (mirrors the v4 servo) + s['last_mapped'] = mapped s['n'] += 1 if s['t0'] is None: s['t0'] = now if s['n'] % self.diag_every == 0: el = (now - s['t0']) / 1e9 + # fps over the FULL window since t0 (not a per-tick burst) → the committed + # cadence, ~grain rate in steady state. A persistently high value means + # videorate isn't decimating (source-rate mismatch worth investigating). fps = s['n'] / el if el > 0 else 0 - print(f"diag n={s['n']} fps={fps:.2f} err={err/1e6:.0f}ms", flush=True) + print(f"diag n={s['n']} fps={fps:.2f} err={err/1e6:.0f}ms " + f"grain={grain_ns/1e6:.1f}ms dropped={s.get('dropped', 0)}", flush=True) + self._write_ingress_record(pad, event='diag', err_ns=err, grain_ns=grain_ns) return Gst.PadProbeReturn.OK # --- the AUDIO restamp (verbatim from guest_audio.py) --- @@ -442,21 +574,36 @@ def _restamp_audio(self, pad, info): return Gst.PadProbeReturn.OK now = clock.get_time() - self.pipe.get_base_time() s = self.state - if s.get('next_pts') is None or abs(now + MARGIN_NS - s['next_pts']) > 500_000_000: - s['next_pts'] = now + MARGIN_NS - # Monotonic re-anchor on a persistent-flow leg rebuild: never place the new - # leg's audio before the last sample we already wrote (same rule as video). - if s.get('relock') and s.get('last_mapped') is not None: - s['next_pts'] = max(s['next_pts'], s['last_mapped'] + FRAME_NS) - print(f'audio cadence RE-ANCHORED (monotonic): {s["next_pts"]/1e6:.0f}ms', flush=True) - s['relock'] = False + target = now + MARGIN_NS + last = s.get('last_mapped') + # Re-anchor when we have no anchor yet, OR we've drifted far from wall-clock. The OLD + # code re-anchored to `now + MARGIN` UNCONDITIONALLY on a >500ms gap — but on bursty + # fan-out delivery `next_pts` races AHEAD (it advances by buf.duration per buffer), so + # the re-anchor JUMPED next_pts BACKWARD to now+MARGIN → non-monotonic PTS into mxlsink + # (no backward-index guard) → "streaming stopped, reason error (-5)" → restart loop + # (HW Oct-8, reproduced with a live camera's AAC through the A/V fan-out). FIX: the + # anchor floors at last_mapped + one audio quantum, so a re-anchor can only ever nudge + # FORWARD — never rewind the flow. (Mirrors the video grid's monotonic guarantee.) + if s.get('next_pts') is None or abs(target - s['next_pts']) > 500_000_000: + new_anchor = target + if last is not None: + # never place audio at/behind the last sample already written (one buffer's + # worth of headroom; buf.duration is this buffer's span) + quantum = buf.duration if buf.duration != Gst.CLOCK_TIME_NONE else FRAME_NS + new_anchor = max(new_anchor, last + quantum) + relock = bool(s.get('relock') and last is not None) + s['relock'] = False + s['next_pts'] = new_anchor if s['offset'] is None: # first lock -> announce if asked (parity w/ video) - s['offset'] = s['next_pts'] - print(f'audio cadence anchored: {s["next_pts"]/1e6:.0f}ms', flush=True) + s['offset'] = new_anchor + print(f'audio cadence anchored: {new_anchor/1e6:.0f}ms', flush=True) if self.a.announce_on_lock and self.repair_url: threading.Thread(target=self._announce, daemon=True).start() elif self.a.announce_on_lock: print('announce skipped (no repair_url) — mixer tolerates absent flows', flush=True) + else: + print(f'audio cadence {"RE-ANCHORED" if relock else "re-anchored"} (monotonic): ' + f'{new_anchor/1e6:.0f}ms', flush=True) buf.pts = s['next_pts'] s['last_mapped'] = s['next_pts'] # for a monotonic re-anchor on the next rebuild if buf.duration != Gst.CLOCK_TIME_NONE: @@ -465,6 +612,13 @@ def _restamp_audio(self, pad, info): def run(self): policy = self.a.timing_policy + # Data (ANC) grains are frame-aligned by the ST-2038 wrapper and carry their own + # timing — there is no sensible video/audio restamp for them. Force 'preserve' so + # the probe is never attached, regardless of what the adapter requested. + if self.a.essence == 'data' and policy == 'restamp': + print("essence=data: forcing timing_policy=preserve (ST-2038 grains are " + "frame-aligned; no restamp)", flush=True) + policy = 'preserve' # pick the essence-appropriate restamp probe (video offset-map vs audio # duration-accumulate). Both reference the same clock + MARGIN_NS. restamp_probe = self._restamp_audio if self.a.essence == 'audio' else self._restamp diff --git a/tools/guest_av_listen.sh b/tools/guest_av_listen.sh index c4e7c5c..4d3dc7e 100755 --- a/tools/guest_av_listen.sh +++ b/tools/guest_av_listen.sh @@ -35,13 +35,22 @@ LATENCY_MS="${4:-300}" # our srtsink callers reconnect to them. wait-for-connection=false so the pipeline comes # up and listens even before a contributor or a leg is present. while :; do + # tee branches MUST be decoupled with leaky queues (HW Oct-8): both legs carry the full TS + # (~13 Mbit/s each), so if either srtsink's SRT send buffer backs up momentarily, a PLAIN + # bounded queue fills → backpressures the tee → the OTHER leg's srtsink SRT-errors ("Error on + # SRT socket (0)"). Observed: the audio leg (:${AUDIO_PORT}) errored + its core -5-looped while + # the video leg looked fine — classic unbalanced-tee stall. leaky=downstream makes each branch + # DROP its own oldest TS packets rather than stall the tee, so one slow leg can't kill the other. + # tee allow-not-linked so a torn-down leg (core restart) doesn't error the whole fan-out. gst-launch-1.0 -e \ srtsrc uri="srt://0.0.0.0:${PUBLIC_PORT}?mode=listener&latency=${LATENCY_MS}" \ - ! tsparse set-timestamps=true ! tee name=t \ - t. ! queue ! srtsink uri="srt://127.0.0.1:${VIDEO_PORT}?mode=caller&latency=${LATENCY_MS}" \ - wait-for-connection=false \ - t. ! queue ! srtsink uri="srt://127.0.0.1:${AUDIO_PORT}?mode=caller&latency=${LATENCY_MS}" \ - wait-for-connection=false \ + ! tsparse set-timestamps=true ! tee name=t allow-not-linked=true \ + t. ! queue leaky=downstream max-size-time=400000000 max-size-buffers=0 max-size-bytes=0 \ + ! srtsink uri="srt://127.0.0.1:${VIDEO_PORT}?mode=caller&latency=${LATENCY_MS}" \ + wait-for-connection=false \ + t. ! queue leaky=downstream max-size-time=400000000 max-size-buffers=0 max-size-bytes=0 \ + ! srtsink uri="srt://127.0.0.1:${AUDIO_PORT}?mode=caller&latency=${LATENCY_MS}" \ + wait-for-connection=false \ || true echo "guest_av_listen: fan-out exited (contributor gone or leg churn) — rebuilding" >&2 sleep 2 diff --git a/tools/ingress-soak.sh b/tools/ingress-soak.sh new file mode 100755 index 0000000..f4a203b --- /dev/null +++ b/tools/ingress-soak.sh @@ -0,0 +1,150 @@ +#!/usr/bin/env sh +# ingress-soak.sh — objective PASS/FAIL soak verdict for a contribution ingest, +# in the MXL Timing Model's OWN terms. +# +# The Timing Model (dmf-mxl docs, "Timing Model") says a healthy writer keeps flow +# latency "low and mostly constant" over hours, with no runaway drift. `mxl-info` +# exposes no transfer latency directly (see docs/GCP-BUILD-PLAN.md), so we read the +# IN-005 ingress registry that contribution_core writes (MXL_INGRESS_DIR, default +# /tmp/mxl-ingress/.json) — it carries the restamp's running wall-clock error +# (err_ms = grid residual), the locked offset, the grain step, and the dropped-frame count +# (the grid servo's burst-guard activity). We sample those over a window and apply: +# +# PASS when, across the window: +# - the record keeps UPDATING (mono_ns advances) AND frames advance (essence moving) +# - |err_ms| stays within ERR_BAND_MS (bounded — "low"), over real numeric samples only +# (lock/relock records carry err_ms=null and are skipped, not counted as 0) +# - err_ms does not trend away: last-vs-first spread < ERR_DRIFT_MS ("constant") +# - dropped does not climb faster than DROP_RATE_MAX per committed frame (no burst-guard +# thrash — the grid never settling is the Oct-7 60fps regression signal) +# STALE when the record never updated across the window (can't judge — run longer). +# FAIL otherwise, naming which clause broke. +# +# This is the explicit pass/fail for the 60fps restamp HW re-verify and any guest soak. +# +# Usage: +# tools/ingress-soak.sh [flow8|path] # default: every *.json in MXL_INGRESS_DIR +# DURATION=300 INTERVAL=10 tools/ingress-soak.sh guest1 # 5-min soak, sample /10s +# +# Pure POSIX sh + the JSON records. No jq (parse like scripts/doctor.sh's jget). + +set -eu + +INGRESS_DIR="${MXL_INGRESS_DIR:-/tmp/mxl-ingress}" +DURATION="${DURATION:-120}" # soak window, seconds +INTERVAL="${INTERVAL:-10}" # sample period, seconds +ERR_BAND_MS="${ERR_BAND_MS:-250}" # |err| must stay under this (bounded = "low") +ERR_DRIFT_MS="${ERR_DRIFT_MS:-150}" # max first..last |err| spread (no runaway = "constant") +# Max dropped-frames climb PER committed frame before it's thrash. A 60->30 decimation legit +# drops ~1 per committed frame, so allow 3 (headroom); sustained higher = the burst guard +# churning (the grid never settling), which is the real 60fps regression signal. +DROP_RATE_MAX="${DROP_RATE_MAX:-3}" + +# jget — scalar value for "key" (same approach as scripts/doctor.sh). +jget() { sed -n "s/.*\"$2\" *: *\"\{0,1\}\([^,\"}]*\).*/\1/p" < true if X is a plain (optionally -) integer, not 'null'/''/decimal-only + is_int() { case "${1#-}" in ''|*[!0-9]*) return 1 ;; *) return 0 ;; esac } + while [ "$elapsed" -le "$DURATION" ]; do + body="$(cat "$rec" 2>/dev/null || true)" + err="$(jget "$body" err_ms)" + frames="$(jget "$body" frames)"; frames="${frames:-0}" + dropped="$(jget "$body" dropped)"; dropped="${dropped:-0}" + mono="$(jget "$body" mono_ns)"; mono="${mono:-0}" + # err_ms is JSON null on lock/relock records — do NOT treat null as 0 (that masks drift). + # Only fold a real numeric err into the band/drift stats. + err_i="${err%.*}" + if is_int "$err_i"; then + abs="${err_i#-}" + [ "$abs" -gt "$max_abs" ] 2>/dev/null && max_abs="$abs" + [ -z "$first_err" ] && first_err="$err_i" + last_err="$err_i"; err_samples=$((err_samples + 1)) + err_disp="${err_i}ms" + else + err_disp="(null)" + fi + is_int "$frames" || frames=0 + [ -z "$first_frames" ] && { first_frames="$frames"; first_dropped="$dropped"; first_mono="$mono"; } + last_frames="$frames"; last_dropped="$dropped"; last_mono="$mono" + samples=$((samples + 1)) + printf ' %-10s t=%3ds err=%7s frames=%-7s dropped=%s\n' "$name" "$elapsed" "$err_disp" "$frames" "$dropped" + [ "$elapsed" -ge "$DURATION" ] && break + sleep "$INTERVAL"; elapsed=$((elapsed + INTERVAL)) + done + + # ---- verdict ---- + verdict=PASS; why="" + # 0. the record must actually be UPDATING over the window (mono_ns advances). If it never + # changed, the ingest isn't writing records fast enough to judge — report STALE, not a + # false FAIL/PASS. (Run longer, or lower the ingest's diag_every.) + if [ "$samples" -ge 2 ] && [ "$last_mono" = "$first_mono" ]; then + echo "STALE $name — ingress record did not update across the ${DURATION}s window (mono_ns unchanged); can't judge. Increase DURATION or lower the ingest diag_every." + return 2 + fi + # 1. essence moving: frames must advance (only meaningful once the record is updating) + if [ "$last_frames" -le "$first_frames" ] 2>/dev/null; then + verdict=FAIL; why="$why frames-not-advancing(${first_frames}->${last_frames}=DRIFT);" + fi + # 2/3. err bounded + constant — only if we actually saw numeric err samples (lock-only + # windows carry null err and are not a pass signal on their own). + if [ "$err_samples" -ge 1 ]; then + if [ "$max_abs" -gt "$ERR_BAND_MS" ] 2>/dev/null; then + verdict=FAIL; why="$why err-out-of-band(|${max_abs}|>${ERR_BAND_MS}ms);" + fi + spread=$((last_err - first_err)); spread="${spread#-}" + if [ "$spread" -gt "$ERR_DRIFT_MS" ] 2>/dev/null; then + verdict=FAIL; why="$why err-drifting(${first_err}->${last_err}=${spread}ms>${ERR_DRIFT_MS});" + fi + else + # Every sample carried err_ms=null (lock/relock-only window): we never observed a real + # grid residual, so we CANNOT assert the flow is healthy. Report STALE, not PASS — a + # lock-only window is not a pass signal on its own. + echo "STALE $name — no numeric err_ms samples across the window (lock-only records); can't judge grid residual. Run longer so a 'diag' record is sampled." + return 2 + fi + # 4. burst guard not thrashing: dropped must not climb faster than DROP_RATE_MAX per frame. + # Steady 60->30 decimation drops ~half the frames (dropped grows ~1 per committed frame) — + # that's EXPECTED. Thrash = dropped climbing while frames barely advance. + d_gain=$((last_dropped - first_dropped)); f_gain=$((last_frames - first_frames)) + if [ "$f_gain" -gt 0 ] && [ "$d_gain" -gt $(( f_gain * DROP_RATE_MAX )) ] 2>/dev/null; then + verdict=FAIL; why="$why drop-thrash(dropped+${d_gain} vs frames+${f_gain});" + fi + [ "$verdict" = PASS ] && why=" |err|<=${max_abs}ms bounded+constant (${err_samples} err samples), frames ${first_frames}->${last_frames}, dropped +$((last_dropped - first_dropped))" + echo "$verdict $name —$why" + [ "$verdict" = PASS ] +} + +echo "ingress-soak: ${DURATION}s window, every ${INTERVAL}s (band=±${ERR_BAND_MS}ms, drift<${ERR_DRIFT_MS}ms)" +rc=0 +for rec in $(records "$@"); do + soak_one "$rec" || rc=1 +done +[ "$rc" -eq 0 ] && echo "SOAK PASS" || echo "SOAK FAIL" +exit "$rc" diff --git a/tools/nmos_node.py b/tools/nmos_node.py index 497e28b..4be6cdd 100644 --- a/tools/nmos_node.py +++ b/tools/nmos_node.py @@ -12,18 +12,30 @@ └─ per selector slot: RECEIVER (caps video/v210, subscribed to its attached sender — live PGM shown in tags) -Read-only by design (IS-04 discovery only; the IS-05 connection shim that -lets controllers ROUTE us is the next build). Runs anywhere that can see -the facility's public status API — deliberately NOT on the production -host, so it can be iterated without touching the rig. +IS-04 discovery (above) PLUS an IS-05 Connection API shim (/x-nmos/connection/v1.2): +a controller that PATCHes a RECEIVER's /staged with a sender_id + activate triggers a +real selector cut — because our receivers ARE the switcher's input slots and "program" +is just which slot is live. So `activate receiver slotN` == `cut program to slot N` +(POST /api/mxl/input on the facility). This is the piece that turns the node from +"discoverable" into "routable" by a standard NMOS controller (BCP-007-03 §connection). +Senders are read-only here (an MXL sender has no transport file to stage — BCP-007-03). + +Set --facility to a backend that exposes POST /api/mxl/input for activation to take +effect; without it, staging still works (IS-05 conformant) but activation is a no-op +cut (logged). Mutating the facility is gated the same way the facility gates it +(X-MXL-Token via MXL_CONTROL_TOKEN). Usage: nmos_node.py [--port 8021] [--href http://THIS_HOST:8021/] [--registry http://registry:8010] # enables registration+heartbeat - [--facility https://prodbots.com] + [--facility http://127.0.0.1:3100] # POST /api/mxl/input for IS-05 activation stdlib only. Peer-to-peer queryable without a registry: curl :8021/x-nmos/node/v1.3/senders/ + curl :8021/x-nmos/connection/v1.2/single/receivers/ +A controller routes by copying a Sender's active transport_params +({mxl_flow_id, mxl_domain_id}) onto a Receiver's /staged, then activating — the shape +garethsb/mxl-nmos-c-example uses. See docs/ADR-001-nmos-control-plane.md. """ import argparse import json @@ -36,9 +48,30 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer NS = uuid.UUID('6ba7b812-9dad-11d1-80b4-00c04fd430c8') # uuid5 namespace (X500) -MXL_DOMAIN_ID = 'domain_1' # from the domain's domain_def.json (BCP-007-03 - # prefers a UUID here — upgrading the def is a - # one-line change on the domain host) + +# IS-05 Connection API version. v1.2 is what real MXL controllers speak (the version used by +# garethsb/mxl-nmos-c-example's connect.sh). ADR-001: NMOS owns connection control, so we track +# the version the ecosystem converged on. Backward-compatible with v1.1 clients at the wire level. +CONN_VER = 'v1.2' + + +def _resolve_domain_id(): + """The MXL domain UUID to advertise as `mxl_domain_id` (ADR-001 convergence step 4). + Prefer the `id` in the domain's domain_def.json (BCP-007-03 wants a UUID here); fall back + to the env override, then the legacy 'domain_1' string. A controller that requires a + concrete UUID won't accept a non-UUID — same caveat as the reference example.""" + path = os.environ.get('MXL_DOMAIN', '/mxl-domain') + try: + with open(os.path.join(path, 'domain_def.json')) as f: + did = json.load(f).get('id') + if did: + return did + except Exception: + pass + return os.environ.get('MXL_DOMAIN_ID', 'domain_1') + + +MXL_DOMAIN_ID = _resolve_domain_id() # UUID if domain_def.json has one, else 'domain_1' # name -> (mxl flow uuid, human label) SENDER_FLOWS = { @@ -71,6 +104,9 @@ def __init__(self, args): self.state = {'input': None, 'pvw': None, 'live': {}} self.version = ver() self.lock = threading.Lock() + self._conn = {} # IS-05 staged/active state, keyed by receiver id + # receiver id -> slot, for IS-05 PATCH routing + self._rid_slot = {u5('receiver', f'slot{s}'): s for s in SLOT_TO_SENDER} def poll_facility(self): while True: @@ -120,8 +156,12 @@ def device(self): d = self.base('device', 'switcher', 'MXL Cloud Switcher', 'uncompressed v210 shared-memory switcher (Azure)') d['id'] = self.device_id + # Advertise the IS-05 Connection API so a controller knows we are routable. + conn_href = self.args.href.rstrip('/') + f'/x-nmos/connection/{CONN_VER}/' d.update({'type': 'urn:x-nmos:device:generic', 'node_id': self.node_id, - 'controls': [], 'senders': [u5('sender', n) for n in SENDER_FLOWS], + 'controls': [{'href': conn_href, + 'type': f'urn:x-nmos:control:sr-ctrl/{CONN_VER}'}], + 'senders': [u5('sender', n) for n in SENDER_FLOWS], 'receivers': [u5('receiver', f'slot{s}') for s in SLOT_TO_SENDER]}) return d @@ -191,6 +231,67 @@ def receivers(self): out.append(d) return out + # ---- IS-05 Connection API (BCP-007-03 §connection) --------------------- + # Our receivers are the switcher's input slots; activating one = cutting PGM to that slot. + # The MXL flow identity travels in transport_params (ADR-001 convergence step 1), matching + # garethsb/mxl-nmos-c-example: {mxl_flow_id, mxl_domain_id}. A controller copies a Sender's + # active transport_params onto a Receiver's staged, then activates — so the flow id the + # controller routes is the SENDER's MXL flow uuid, and the receiver's slot is fixed wiring. + @staticmethod + def _mxl_tparams(flow_id=None): + """One transport_params leg for an MXL receiver/sender (BCP-007-03 MXL transport).""" + return [{'mxl_flow_id': flow_id, 'mxl_domain_id': MXL_DOMAIN_ID}] + + def _blank_conn_leg(self, flow_id=None): + return {'master_enable': False, 'sender_id': None, + 'activation': {'mode': None, 'requested_time': None, 'activation_time': None}, + 'transport_params': self._mxl_tparams(flow_id)} + + def _recv_conn(self, slot): + """Lazily-initialised staged/active connection record for a receiver slot. Seeded with + the slot's wired sender flow id so a GET before any PATCH still shows a sane shape.""" + rid = u5('receiver', f'slot{slot}') + wired = SENDER_FLOWS.get(SLOT_TO_SENDER.get(slot, ''), (None,))[0] + rec = self._conn.setdefault(rid, { + 'staged': self._blank_conn_leg(wired), + 'active': self._blank_conn_leg(wired), + }) + return rid, rec + + def _ctl_headers(self): + h = {'Content-Type': 'application/json', 'User-Agent': 'mxl-nmos-node/1.0'} + tok = os.environ.get('MXL_CONTROL_TOKEN', '').strip() + if tok: + h['X-MXL-Token'] = tok + return h + + def activate_receiver(self, slot, sender_id): + """Apply an IS-05 activation: cut the switcher's program to this slot. + Returns (ok, detail). Never raises — a failed cut becomes a 500 to the + controller, not a crash.""" + # The staged sender must be one of ours (or None to disconnect). We route by + # SLOT — the sender_id is advisory here (the receiver's slot is fixed wiring). + try: + req = urllib.request.Request( + self.args.facility + '/api/mxl/input', + data=json.dumps({'slot': slot}).encode(), + headers=self._ctl_headers(), method='POST') + with urllib.request.urlopen(req, timeout=8) as r: + body = r.read()[:200].decode('utf-8', 'replace') + print(f'IS-05 activate: receiver slot{slot} -> PGM (sender {sender_id}) ' + f'=> {r.status} {body}', flush=True) + return True, body + except Exception as e: + print(f'IS-05 activate slot{slot} FAILED: {e}', flush=True) + return False, str(e) + + def conn_senders(self): + # Senders are read-only in BCP-007-03 (no transport file to stage). + return [u5('sender', n) for n in SENDER_FLOWS] + + def conn_receivers(self): + return [u5('receiver', f'slot{s}') for s in SLOT_TO_SENDER] + MODEL = None class Handler(BaseHTTPRequestHandler): @@ -210,13 +311,19 @@ def do_GET(self): p = self.path.rstrip('/') m = MODEL table = { - '/x-nmos': ['node/'], + '/x-nmos': ['node/', 'connection/'], '/x-nmos/node': ['v1.3/'], '/x-nmos/node/v1.3': ['self/', 'devices/', 'sources/', 'flows/', 'senders/', 'receivers/'], + '/x-nmos/connection': [f'{CONN_VER}/'], + f'/x-nmos/connection/{CONN_VER}': ['single/', 'bulk/'], + f'/x-nmos/connection/{CONN_VER}/single': ['senders/', 'receivers/'], } if p in table: return self.send_json(table[p]) + # ---- IS-05 Connection API (GET) ---- + if self.conn_get(p, m): + return col = {'/x-nmos/node/v1.3/devices': lambda: [m.device()], '/x-nmos/node/v1.3/sources': m.sources, '/x-nmos/node/v1.3/flows': m.flows, @@ -235,6 +342,121 @@ def do_GET(self): return self.send_json({'code': 404, 'error': 'not found'}, 404) self.send_json({'code': 404, 'error': 'not found'}, 404) + # ---- IS-05 Connection API: GET of single/{senders,receivers}/... -------- + def conn_get(self, p, m): + """Handle a GET under /x-nmos/connection//single/. Returns True if it + matched (and responded), False to fall through to the IS-04 handler.""" + S = f'/x-nmos/connection/{CONN_VER}/single/senders' + R = f'/x-nmos/connection/{CONN_VER}/single/receivers' + if p == S: + self.send_json([s + '/' for s in m.conn_senders()]); return True + if p == R: + self.send_json([r + '/' for r in m.conn_receivers()]); return True + # senders are read-only (no transport file): expose the standard sub-resources. + if p.startswith(S + '/'): + rest = p[len(S) + 1:] + sid, _, leaf = rest.partition('/') + if sid not in m.conn_senders(): + self.send_json({'code': 404, 'error': 'no such sender'}, 404); return True + if leaf == '': + self.send_json(['constraints/', 'staged/', 'active/', 'transporttype/']); return True + if leaf == 'transporttype': + self.send_json('urn:x-nmos:transport:mxl'); return True + if leaf == 'constraints': + self.send_json([{}]); return True + if leaf in ('staged', 'active'): + # Carry the sender's real MXL flow id so a controller can copy sender->receiver + # (the whole point of IS-05 — garethsb/mxl-nmos-c-example's connect.sh reads this). + fid = next((f for n, (f, _l) in SENDER_FLOWS.items() if u5('sender', n) == sid), None) + self.send_json({'master_enable': True, 'receiver_id': None, + 'activation': {'mode': None, 'requested_time': None, + 'activation_time': None}, + 'transport_params': m._mxl_tparams(fid)}); return True + self.send_json({'code': 404, 'error': 'not found'}, 404); return True + if p.startswith(R + '/'): + rest = p[len(R) + 1:] + rid, _, leaf = rest.partition('/') + if rid not in m._rid_slot: + self.send_json({'code': 404, 'error': 'no such receiver'}, 404); return True + _, rec = m._recv_conn(m._rid_slot[rid]) + if leaf == '': + self.send_json(['constraints/', 'staged/', 'active/', 'transporttype/']); return True + if leaf == 'transporttype': + self.send_json('urn:x-nmos:transport:mxl'); return True + if leaf == 'constraints': + self.send_json([{}]); return True + if leaf == 'staged': + self.send_json(rec['staged']); return True + if leaf == 'active': + self.send_json(rec['active']); return True + self.send_json({'code': 404, 'error': 'not found'}, 404); return True + return False + + # ---- IS-05 Connection API: PATCH single/receivers//staged ----------- + def do_PATCH(self): + m = MODEL + p = self.path.rstrip('/') + R = f'/x-nmos/connection/{CONN_VER}/single/receivers' + if not (p.startswith(R + '/') and p.endswith('/staged')): + return self.send_json({'code': 405, 'error': 'only receiver /staged is patchable ' + '(senders are read-only on an MXL transport)'}, 405) + rid = p[len(R) + 1:-len('/staged')] + if rid not in m._rid_slot: + return self.send_json({'code': 404, 'error': 'no such receiver'}, 404) + try: + n = int(self.headers.get('Content-Length', 0)) + patch = json.loads(self.rfile.read(n) or b'{}') + except Exception: + return self.send_json({'code': 400, 'error': 'bad JSON body'}, 400) + slot = m._rid_slot[rid] + _, rec = m._recv_conn(slot) + st = rec['staged'] + # merge the PATCH into staged (IS-05 is a partial update) + if 'master_enable' in patch: + st['master_enable'] = bool(patch['master_enable']) + if 'sender_id' in patch: + st['sender_id'] = patch['sender_id'] + if 'transport_params' in patch and patch['transport_params']: + # The controller routes by the MXL flow id in transport_params (the standard + # shape — garethsb/mxl-nmos-c-example). Merge per-leg so a partial PATCH keeps + # the fields it didn't set. + for i, leg in enumerate(patch['transport_params']): + if i < len(st['transport_params']): + st['transport_params'][i].update(leg) + else: + st['transport_params'].append(leg) + act = patch.get('activation') or {} + mode = act.get('mode') + st['activation'] = {'mode': mode, 'requested_time': act.get('requested_time'), + 'activation_time': None} + # Only immediate activation is supported (no scheduled/relative PTP here). + if mode == 'activate_immediate': + if st['master_enable']: + # ENABLE: route program to this slot. (Flow id travels in transport_params; + # the slot is our fixed wiring, so the cut is by slot — we log the flow id.) + fid = (st['transport_params'][0] or {}).get('mxl_flow_id') + ok, detail = m.activate_receiver(slot, fid) + if not ok: + return self.send_json({'code': 500, 'error': f'activation failed: {detail}'}, 500) + else: + # DISABLE (master_enable=false): IS-05 disable is first-class (ADR-001 step 3). + # Receiver stops consuming; we don't force a program change (the switcher always + # shows SOME slot), we just record the disabled state. A re-enable re-activates. + print(f'IS-05 disable: receiver slot{slot} master_enable=false (no forced cut)', + flush=True) + # promote staged -> active; stamp activation_time + st['activation']['activation_time'] = ver() + rec['active'] = json.loads(json.dumps(st)) # deep copy + rec['active']['activation']['mode'] = 'activate_immediate' + with m.lock: + m.version = ver() + return self.send_json(st, 200) + elif mode in ('activate_scheduled_absolute', 'activate_scheduled_relative'): + return self.send_json({'code': 501, 'error': 'scheduled activation not supported ' + '(immediate only on this MXL shim)'}, 501) + # staging without activation + return self.send_json(st, 200) + def register_loop(m, registry): api = registry.rstrip('/') + '/x-nmos/registration/v1.3' def post(kind, data): @@ -268,7 +490,10 @@ def main(): ap.add_argument('--port', type=int, default=8021) ap.add_argument('--href', default=None) ap.add_argument('--registry', default=None) - ap.add_argument('--facility', default=os.environ.get('MXL_BACKEND_URL', 'https://prodbots.com')) + # Portable default: the LOCAL control plane, not any specific facility (adopter PR #38's + # sweep — a cloned core must not point at someone else's backend). Override with --facility + # or MXL_BACKEND_URL for a real deployment. + ap.add_argument('--facility', default=os.environ.get('MXL_BACKEND_URL', 'http://127.0.0.1:3100')) args = ap.parse_args() if not args.href: args.href = f'http://{socket.gethostbyname(socket.gethostname())}:{args.port}/' @@ -277,8 +502,9 @@ def main(): if args.registry: threading.Thread(target=register_loop, args=(MODEL, args.registry), daemon=True).start() - print(f'IS-04 node on :{args.port} (href {args.href})' - + (f' -> registry {args.registry}' if args.registry else ' (peer-to-peer)'), + print(f'IS-04 + IS-05 node on :{args.port} (href {args.href})' + + (f' -> registry {args.registry}' if args.registry else ' (peer-to-peer)') + + f'; IS-05 activation routes to {args.facility}/api/mxl/input', flush=True) ThreadingHTTPServer(('0.0.0.0', args.port), Handler).serve_forever()