From aca88c3dbcda7351c0051a1e1cd44f2eabce9046 Mon Sep 17 00:00:00 2001 From: Blax Software Date: Fri, 10 Jul 2026 11:29:26 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20Rust=20SFU=20data=20plane=20=E2=80=94?= =?UTF-8?q?=20backend-hosted=20rooms,=20plug-and-play=20sidecar=20binary?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit rust/ becomes a standalone static binary (blax-webrtc-sfu, tokio-free std-thread loop on str0m): terminates ICE/DTLS/SRTP, forwards Opus between room peers without decoding, records per-peer Ogg/Opus (ffprobe-verified), server-side mute/kick, stats + peer_connected/peer_left usage events, all driven by PHP over a JSON-lines Unix control socket. Renegotiation rides the browser data channel (str0m chat model) — signaling only carries the initial offer/answer. PHP side: RustMediaEngine (MediaEngine contract + room-aware API) is plug-and-play — first use downloads the sha256-verified prebuilt binary (BinaryManager, RoadRunner pattern) and spawns it detached, flock-guarded (SidecarSupervisor). webrtc:install / webrtc:sidecar provided for explicit control; config gains the webrtc.sfu block. Replaces the ext-php-rs Str0mMediaEngine seam. Tests: 61 PHP (incl. a real PHP<->Rust E2E that spawns the sidecar from PHPUnit) + 7 Rust (incl. an end-to-end control-socket integration test). build-release.sh produces the static-musl release + .sha256 (verified: static-pie, 9.2MB, runs). Card #1083 (rel #1051 #1060 #1064). Co-Authored-By: Claude Fable 5 --- .gitignore | 4 +- CHANGELOG.md | 29 + README.md | 61 +- config/webrtc.php | 51 +- rust/Cargo.lock | 824 ++++++++++++++++++++++ rust/Cargo.toml | 26 +- rust/README.md | 86 ++- rust/build-release.sh | 45 ++ rust/src/control.rs | 106 +++ rust/src/lib.rs | 66 +- rust/src/main.rs | 55 ++ rust/src/proto.rs | 179 +++++ rust/src/recording.rs | 150 ++++ rust/src/server.rs | 85 +++ rust/src/sfu.rs | 807 +++++++++++++++++++++ rust/tests/control.rs | 96 +++ src/Console/Commands/InstallSfuBinary.php | 43 ++ src/Console/Commands/RunSfuSidecar.php | 52 ++ src/Contracts/MediaEngine.php | 6 +- src/Media/NullMediaEngine.php | 2 +- src/Media/Rust/BinaryManager.php | 159 +++++ src/Media/Rust/ControlClient.php | 137 ++++ src/Media/Rust/SidecarException.php | 11 + src/Media/Rust/SidecarSupervisor.php | 151 ++++ src/Media/RustMediaEngine.php | 161 +++++ src/Media/Str0mMediaEngine.php | 82 --- src/WebRtcServiceProvider.php | 19 + tests/BinaryManagerTest.php | 117 +++ tests/ControlClientTest.php | 97 +++ tests/RustMediaEngineTest.php | 130 ++++ tests/SidecarSupervisorTest.php | 107 +++ 31 files changed, 3749 insertions(+), 195 deletions(-) create mode 100644 rust/Cargo.lock create mode 100755 rust/build-release.sh create mode 100644 rust/src/control.rs create mode 100644 rust/src/main.rs create mode 100644 rust/src/proto.rs create mode 100644 rust/src/recording.rs create mode 100644 rust/src/server.rs create mode 100644 rust/src/sfu.rs create mode 100644 rust/tests/control.rs create mode 100644 src/Console/Commands/InstallSfuBinary.php create mode 100644 src/Console/Commands/RunSfuSidecar.php create mode 100644 src/Media/Rust/BinaryManager.php create mode 100644 src/Media/Rust/ControlClient.php create mode 100644 src/Media/Rust/SidecarException.php create mode 100644 src/Media/Rust/SidecarSupervisor.php create mode 100644 src/Media/RustMediaEngine.php delete mode 100644 src/Media/Str0mMediaEngine.php create mode 100644 tests/BinaryManagerTest.php create mode 100644 tests/ControlClientTest.php create mode 100644 tests/RustMediaEngineTest.php create mode 100644 tests/SidecarSupervisorTest.php diff --git a/.gitignore b/.gitignore index 6aacbc0..171c225 100644 --- a/.gitignore +++ b/.gitignore @@ -7,6 +7,6 @@ composer.lock .idea/ .vscode/ -# Rust core build artifacts +# Rust build artifacts (Cargo.lock IS committed — bin crate, reproducible releases) /rust/target/ -/rust/Cargo.lock +/rust/dist/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 9b4b1b1..69922ec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,35 @@ Blax Software's backward-compatibility guarantee. ### Added +- **Backend-hosted rooms — the Rust SFU sidecar** (`rust/`, binary + `blax-webrtc-sfu`): str0m terminates ICE/DTLS/SRTP server-side and forwards + Opus between room peers without decoding; per-peer **Ogg/Opus recording** + written from the decrypted stream; server-side **mute/kick**, `stats` + (connected/muted/usage seconds) and pushed `peer_connected`/`peer_left` + events. Driven over a JSON-lines Unix control socket. Renegotiation rides + each browser's data channel (str0m `chat` model) — signaling only carries the + initial offer/answer. +- **`Media\RustMediaEngine`** — implements `MediaEngine` (composite + `room/peer` ids) plus the room-aware API (`addPeer`/`removePeer`/`mutePeer`/ + `stats`/`drainEvents`). **Plug-and-play**: first use auto-downloads the + sha256-verified prebuilt static binary (`Rust\BinaryManager`, RoadRunner + pattern) and spawns it detached + flock-guarded (`Rust\SidecarSupervisor`). +- Commands `webrtc:install` (pre-fetch the binary) and `webrtc:sidecar` + (foreground runner for docker/supervisord; disable `auto_spawn` there). +- `config/webrtc.php` gained the `sfu` block (socket, udp/public ip+port, + auto_install/auto_spawn, download/checksum URLs, versioned install path). +- `rust/build-release.sh` builds the static-musl release binary + `.sha256`. +- Tests: 61 PHP (control protocol framing, event buffering, checksum + install/mismatch, supervisor args + a REAL PHP↔Rust E2E that spawns the + sidecar from PHPUnit) and 7 Rust (proto round-trips, Ogg/Opus writer + [ffprobe-verified], end-to-end control-socket integration). + +### Removed + +- `Media\Str0mMediaEngine` (the ext-php-rs extension seam) — replaced by + `RustMediaEngine` + the sidecar: no PHP-version-bound extension, no crash + sharing a process with PHP, works with the prebuilt-base-image deploy model. + - **Typed rooms** (`Rooms\RoomType`, derived from the room name prefix like laravel-websockets channels): `public`, `private-*` (authorized), `presence-*` (authorized + member list), `open-presence-*` (public + member diff --git a/README.md b/README.md index 388b2cb..528478b 100644 --- a/README.md +++ b/README.md @@ -124,15 +124,62 @@ rec.start(250) Recordings finalize on leave/disconnect. Swap `FileRecordingStore` for your own `RecordingStore` (S3, DB blob) and bind a `CallEventListener` (`WEBRTC_EVENTS`) to persist call records — participants, duration, and the finalized recording locator — even for calls you don't record. -### Server-terminated media (AI bridge / SFU) +### Backend-hosted rooms — the Rust SFU (recommended for production) -When the server must be *in* the media path, configure a `MediaEngine`. `NullMediaEngine` (default) is signaling-only; `Str0mMediaEngine` is backed by a Rust `str0m` core via `ext-php-rs` (see [`rust/README.md`](rust/README.md)): +Mesh is fine for two browsers; for **hosted rooms** — server in the media path, per-peer recording without client cooperation, server-side mute/kick, usage metering, room sizes beyond ~5 — switch the media engine to the bundled Rust SFU (Selective Forwarding Unit: it receives each peer's audio once and forwards it to the others *without decoding*): ```dotenv -WEBRTC_MEDIA_ENGINE="Blax\WebRtc\Media\Str0mMediaEngine" +WEBRTC_MEDIA_ENGINE="Blax\WebRtc\Media\RustMediaEngine" ``` -The engine-backed path uses `Signaling\SignalingHandler` (offer/ice/record/connect/bridge/close) + `Server\SignalingServer`; it's independent of the browser relay above and shares the same `RoomManager`. +That's the whole setup — it is **plug-and-play**. On first use the engine downloads the prebuilt, sha256-verified `blax-webrtc-sfu` static binary for your platform (the RoadRunner pattern) and spawns it detached, flock-guarded so concurrent workers race exactly one spawn. Explicit control when you want it: + +```bash +php artisan webrtc:install # pre-download the binary (deploy step) +php artisan webrtc:sidecar # run it in the FOREGROUND (docker service / + # supervisor; set WEBRTC_SFU_AUTO_SPAWN=false there) +``` + +PHP stays the control plane. Sessions look like: + +```php +use Blax\WebRtc\Media\RustMediaEngine; + +$engine = app(RustMediaEngine::class); + +// Browser sends its SDP offer over YOUR signaling (WS/HTTP) — answer it: +$answer = $engine->addPeer('lobby', $userId, $sdpOffer, recordPath: "/rec/lobby/$userId.ogg"); + +$engine->mutePeer('lobby', $userId, true); // room stops hearing them +$engine->removePeer('lobby', $userId); // kick +$engine->stats(); // rooms → peers → connected/muted/seconds +$engine->drainEvents(); // peer_connected / peer_left (usage seconds) +``` + +Recordings are standard **Ogg/Opus** (ffmpeg/browser playable), written by the sidecar from the decrypted SRTP — no browser MediaRecorder upload needed on this path. + +**Browser contract** (one thing beyond a normal WebRTC client): create a *data channel* before offering. Only the initial offer/answer passes through your signaling; when peers join later, the SFU renegotiates new audio tracks **over that data channel** directly: + +```js +const pc = new RTCPeerConnection() +pc.addTrack(mic.getAudioTracks()[0], mic) +const dc = pc.createDataChannel('signaling') +dc.onmessage = async (e) => { // SFU offers new tracks as peers join + await pc.setRemoteDescription(JSON.parse(e.data)) + const answer = await pc.createAnswer() + await pc.setLocalDescription(answer) + dc.send(JSON.stringify(answer)) +} +pc.ontrack = (e) => playRemoteAudio(e.streams[0]) +const offer = await pc.createOffer() +await pc.setLocalDescription(offer) +const { answer } = await yourSignaling.addPeer('lobby', offer.sdp) // → $engine->addPeer(...) +await pc.setRemoteDescription({ type: 'answer', sdp: answer }) +``` + +Ops notes: pin `WEBRTC_SFU_UDP_PORT` and publish it (UDP) on your container, and set `WEBRTC_SFU_PUBLIC_IP` to the address browsers can reach — it's advertised in the ICE host candidate, which is also why no TURN server is required for the common case. See the `sfu` block in `config/webrtc.php` for every knob (socket path, auto_install/auto_spawn, download/checksum URLs, log). + +The legacy raw-TCP `Signaling\SignalingHandler` + `Server\SignalingServer` path still exists for driving a `MediaEngine` without a browser relay and shares the same `RoomManager`. ### Use it on your own WebSocket (no bundled server) @@ -172,7 +219,7 @@ Membership lives in a swappable `RoomStore`. The default `ArrayRoomStore` keeps ## Status -**Working package.** Browser-to-browser (mesh) calls, typed rooms (public / private / presence / open-presence) with authorization + presence, and full per-participant **call recording** (mic streamed over the WS, stored to disk) all work today and are tested (incl. a real WebSocket round-trip). The only remaining piece is *server-terminated SRTP* media (an in-path SFU/recorder for AI bridging or huge groups) — that's the `Str0mMediaEngine` + `rust/` core, the next build. +**Working package.** Browser-to-browser (mesh) calls, typed rooms (public / private / presence / open-presence) with authorization + presence, full per-participant **call recording**, and **backend-hosted rooms on the Rust SFU sidecar** (server-terminated ICE/DTLS/SRTP, forwarding without decode, native Ogg/Opus recording, mute/kick/stats/usage events, plug-and-play binary management) — all tested, including a real PHP↔Rust round-trip that spawns the sidecar from PHPUnit. Next: the duplex AI bridge inside the SFU data plane (today `OpenAiRealtimeBridge` covers turn-based AI participants) and browser E2E automation. ## Architecture @@ -189,7 +236,9 @@ laravel-webrtc (this) │ └─ RelaySignalingHandler browser P2P calls over the loop │ └─ MediaEngine (optional) …or terminate media server-side - └─ Str0mMediaEngine Rust/str0m via ext-php-rs (rust/) + └─ RustMediaEngine backend-hosted rooms on the bundled + └─ blax-webrtc-sfu Rust SFU sidecar (str0m) — JSON control + socket, plug-and-play binary (rust/) ``` The domain layer runs on **any** transport (the bundled server, or a host WS like learn-atc's fork-per-message WebSocket). The bundled server + media engine are opt-in. diff --git a/config/webrtc.php b/config/webrtc.php index 2c54554..03de6c9 100644 --- a/config/webrtc.php +++ b/config/webrtc.php @@ -3,6 +3,7 @@ use Blax\WebRtc\Authorization\DefaultRoomAuthorizer; use Blax\WebRtc\Events\NullCallEventListener; use Blax\WebRtc\Media\NullMediaEngine; +use Blax\WebRtc\Rooms\Stores\ArrayRoomStore; return [ /* @@ -27,11 +28,55 @@ return [ | Media engine |-------------------------------------------------------------------------- | The backend that terminates media (ICE/DTLS/SRTP/RTP/Opus), records, and - | bridges to external realtime providers. NullMediaEngine = signaling only; - | Str0mMediaEngine requires the blax_webrtc Rust extension (see rust/). + | bridges to external realtime providers. NullMediaEngine = signaling only + | (peer-to-peer mesh); RustMediaEngine = BACKEND-HOSTED rooms on the Rust + | SFU sidecar (plug-and-play — see the `sfu` block below). */ 'media_engine' => env('WEBRTC_MEDIA_ENGINE', NullMediaEngine::class), + /* + |-------------------------------------------------------------------------- + | Rust SFU sidecar (RustMediaEngine) + |-------------------------------------------------------------------------- + | The `blax-webrtc-sfu` binary from this package's rust/ directory: one + | process per host that terminates ICE/DTLS/SRTP and forwards audio between + | room peers (no decoding), records Ogg/Opus per peer, and reports usage. + | + | Plug-and-play defaults: the first RustMediaEngine call DOWNLOADS the + | checksummed prebuilt binary for this platform (auto_install) and SPAWNS + | it detached, flock-guarded (auto_spawn). For supervised deployments run + | `php artisan webrtc:sidecar` as a service and disable auto_spawn. + | + | udp_port 0 = ephemeral (fine same-host); pin it (and publish it on the + | container) for production, and set public_ip to the address browsers + | can reach — it is advertised in the ICE host candidate. + */ + 'sfu' => [ + 'socket' => env('WEBRTC_SFU_SOCKET', storage_path('app/webrtc/sfu.sock')), + 'udp_ip' => env('WEBRTC_SFU_UDP_IP', '0.0.0.0'), + 'udp_port' => (int) env('WEBRTC_SFU_UDP_PORT', 0), + 'public_ip' => env('WEBRTC_SFU_PUBLIC_IP'), + + 'auto_spawn' => (bool) env('WEBRTC_SFU_AUTO_SPAWN', true), + 'auto_install' => (bool) env('WEBRTC_SFU_AUTO_INSTALL', true), + 'spawn_timeout' => (float) env('WEBRTC_SFU_SPAWN_TIMEOUT', 10), + 'log' => env('WEBRTC_SFU_LOG', storage_path('logs/webrtc-sfu.log')), + + // Binary resolution: explicit path > local rust/target build > install. + 'binary' => env('WEBRTC_SFU_BINARY'), + 'install_path' => env('WEBRTC_SFU_INSTALL_PATH', storage_path('app/webrtc/bin')), + 'version' => env('WEBRTC_SFU_VERSION'), // default: the version pinned by this package + 'download_url' => env( + 'WEBRTC_SFU_DOWNLOAD_URL', + 'https://git.blax.at/blax-software/laravel-webrtc/releases/download/{version}/blax-webrtc-sfu-{platform}', + ), + // sha256 verification is REQUIRED when set; null opts out explicitly. + 'checksum_url' => env( + 'WEBRTC_SFU_CHECKSUM_URL', + 'https://git.blax.at/blax-software/laravel-webrtc/releases/download/{version}/blax-webrtc-sfu-{platform}.sha256', + ), + ], + /* |-------------------------------------------------------------------------- | Recording (store the full call audio) @@ -67,7 +112,7 @@ return [ | worker) binds a Cache/Redis-backed RoomStore so state survives across | workers — that is how a host runs this relay on its existing WS. */ - 'room_store' => env('WEBRTC_ROOM_STORE', \Blax\WebRtc\Rooms\Stores\ArrayRoomStore::class), + 'room_store' => env('WEBRTC_ROOM_STORE', ArrayRoomStore::class), /* |-------------------------------------------------------------------------- diff --git a/rust/Cargo.lock b/rust/Cargo.lock new file mode 100644 index 0000000..8a39ecf --- /dev/null +++ b/rust/Cargo.lock @@ -0,0 +1,824 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "aho-corasick" +version = "1.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ddd31a130427c27518df266943a5308ed92d4b226cc639f5a8f1002816174301" +dependencies = [ + "memchr", +] + +[[package]] +name = "anstream" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "824a212faf96e9acacdbd09febd34438f8f711fb84e09a8916013cd7815ca28d" +dependencies = [ + "anstyle", + "anstyle-parse", + "anstyle-query", + "anstyle-wincon", + "colorchoice", + "is_terminal_polyfill", + "utf8parse", +] + +[[package]] +name = "anstyle" +version = "1.0.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000" + +[[package]] +name = "anstyle-parse" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52ce7f38b242319f7cabaa6813055467063ecdc9d355bbb4ce0c68908cd8130e" +dependencies = [ + "utf8parse", +] + +[[package]] +name = "anstyle-query" +version = "1.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" +dependencies = [ + "windows-sys", +] + +[[package]] +name = "anstyle-wincon" +version = "3.0.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" +dependencies = [ + "anstyle", + "once_cell_polyfill", + "windows-sys", +] + +[[package]] +name = "anyhow" +version = "1.0.103" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3" + +[[package]] +name = "bitflags" +version = "1.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a" + +[[package]] +name = "bitflags" +version = "2.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b4388bee8683e3d04af747c73422af53102d2bd24d9eadb6cbc100baef4b43f8" + +[[package]] +name = "blax-webrtc-sfu" +version = "0.1.0" +dependencies = [ + "anyhow", + "env_logger", + "log", + "ogg", + "serde", + "serde_json", + "str0m", +] + +[[package]] +name = "block-buffer" +version = "0.10.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" +dependencies = [ + "generic-array", +] + +[[package]] +name = "byteorder" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" + +[[package]] +name = "bytes" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" + +[[package]] +name = "cc" +version = "1.2.66" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f5d6cac793997bd970000024b2934968efe83b382de4fdcf4fcb46b6ee4ad996" +dependencies = [ + "find-msvc-tools", + "shlex", +] + +[[package]] +name = "cfg-if" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" + +[[package]] +name = "colorchoice" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" + +[[package]] +name = "combine" +version = "4.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba5a308b75df32fe02788e748662718f03fde005016435c444eea572398219fd" +dependencies = [ + "bytes", + "memchr", +] + +[[package]] +name = "cpufeatures" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" +dependencies = [ + "libc", +] + +[[package]] +name = "crc" +version = "3.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5eb8a2a1cd12ab0d987a5d5e825195d372001a4094a0376319d5a0ad71c1ba0d" +dependencies = [ + "crc-catalog", +] + +[[package]] +name = "crc-catalog" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "217698eaf96b4a3f0bc4f3662aaa55bdf913cd54d7204591faa790070c6d0853" + +[[package]] +name = "crypto-common" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" +dependencies = [ + "generic-array", + "typenum", +] + +[[package]] +name = "defmt" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e2953bfe4f93bbd20cc71198842756f77d161884c99ebbabc41d80231ded88d1" +dependencies = [ + "bitflags 1.3.2", + "defmt-macros", +] + +[[package]] +name = "defmt-macros" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bad9c72e7ca2137e0dc3813245a0d282fd6daad32fd800af018306a9169b5fe8" +dependencies = [ + "defmt-parser", + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "defmt-parser" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "10d60334b3b2e7c9d91ef8150abfb6fa4c1c39ebbcf4a81c2e346aad939fee3e" +dependencies = [ + "thiserror 2.0.18", +] + +[[package]] +name = "digest" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" +dependencies = [ + "block-buffer", + "crypto-common", + "subtle", +] + +[[package]] +name = "env_filter" +version = "2.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "900d271a03799a1ee8d1ca9b19893b48ca674a9284fefcfb85f05e74ed314217" +dependencies = [ + "log", + "regex", +] + +[[package]] +name = "env_logger" +version = "0.11.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "de671bd27a75a797dc9ae289ba1e77276e75e2026408aab65185384e2d5cd3f6" +dependencies = [ + "anstream", + "anstyle", + "env_filter", + "jiff", + "log", +] + +[[package]] +name = "fastrand" +version = "2.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f1f227452a390804cdb637b74a86990f2a7d7ba4b7d5693aac9b4dd6defd8d6" + +[[package]] +name = "find-msvc-tools" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" + +[[package]] +name = "foreign-types" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f6f339eb8adc052cd2ca78910fda869aefa38d22d5cb648e6485e4d3fc06f3b1" +dependencies = [ + "foreign-types-shared", +] + +[[package]] +name = "foreign-types-shared" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "00b0228411908ca8685dba7fc2cdd70ec9990a6e753e89b6ac91a84c40fbaf4b" + +[[package]] +name = "fxhash" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c31b6d751ae2c7f11320402d34e41349dd1016f8d5d45e48c4312bc8625af50c" +dependencies = [ + "byteorder", +] + +[[package]] +name = "generic-array" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" +dependencies = [ + "typenum", + "version_check", +] + +[[package]] +name = "getrandom" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" +dependencies = [ + "cfg-if", + "libc", + "wasi", +] + +[[package]] +name = "hmac" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e" +dependencies = [ + "digest", +] + +[[package]] +name = "is_terminal_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" + +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + +[[package]] +name = "jiff" +version = "0.2.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "961d16382652bfdd8c6f68b223b26a8c93e0d475c672f414411db31c6c5c900e" +dependencies = [ + "defmt", + "jiff-static", + "log", + "portable-atomic", + "portable-atomic-util", + "serde_core", +] + +[[package]] +name = "jiff-static" +version = "0.2.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0879bd39df99c4c5e2c6615ccc026391a423dde10532c573e6086eb94a802cc" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "libc" +version = "0.2.186" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" + +[[package]] +name = "log" +version = "0.4.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" + +[[package]] +name = "memchr" +version = "2.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" + +[[package]] +name = "ogg" +version = "0.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fdab8dcd8d4052eaacaf8fb07a3ccd9a6e26efadb42878a413c68fc4af1dee2b" +dependencies = [ + "byteorder", +] + +[[package]] +name = "once_cell" +version = "1.21.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" + +[[package]] +name = "once_cell_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" + +[[package]] +name = "openssl" +version = "0.10.81" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77823a27f0babb03091cb9ed9ef80af3b39dbc82f97e8fa530374b7dafd87a45" +dependencies = [ + "bitflags 2.13.0", + "cfg-if", + "foreign-types", + "libc", + "openssl-macros", + "openssl-sys", +] + +[[package]] +name = "openssl-macros" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a948666b637a0f465e8564c73e89d4dde00d72d4d473cc972f390fc3dcee7d9c" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "openssl-src" +version = "300.6.1+3.6.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "46eb8fb9fb3b61ce1c0f8a026c4c1a0714d3a9e138e7fbde78753ce2babc3846" +dependencies = [ + "cc", +] + +[[package]] +name = "openssl-sys" +version = "0.9.117" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b47e7e6bb2c38cd930d25a23b40fa52e068c10e85f3e03a7f5ba5aaca5713695" +dependencies = [ + "cc", + "libc", + "openssl-src", + "pkg-config", + "vcpkg", +] + +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + +[[package]] +name = "pkg-config" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e" + +[[package]] +name = "portable-atomic" +version = "1.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c33a9471896f1c69cecef8d20cbe2f7accd12527ce60845ff44c153bb2a21b49" + +[[package]] +name = "portable-atomic-util" +version = "0.2.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2a106d1259c23fac8e543272398ae0e3c0b8d33c88ed73d0cc71b0f1d902618" +dependencies = [ + "portable-atomic", +] + +[[package]] +name = "ppv-lite86" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9" +dependencies = [ + "zerocopy", +] + +[[package]] +name = "proc-macro2" +version = "1.0.106" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fd00f0bb2e90d81d1044c2b32617f68fcb9fa3bb7640c23e9c748e53fb30934" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dfbc457d0c7a0759a614551b11a6409e5951f6c7537be1f1b7682b9ae9230368" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "rand" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5ca0ecfa931c29007047d1bc58e623ab12e5590e8c7cc53200d5202b69266d8a" +dependencies = [ + "libc", + "rand_chacha", + "rand_core", +] + +[[package]] +name = "rand_chacha" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" +dependencies = [ + "ppv-lite86", + "rand_core", +] + +[[package]] +name = "rand_core" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" +dependencies = [ + "getrandom", +] + +[[package]] +name = "regex" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a0e75113e14dc5acb068cd0786884f214f1312650a3d36d269f5c4f3cdee8a2" +dependencies = [ + "aho-corasick", + "memchr", + "regex-automata", + "regex-syntax", +] + +[[package]] +name = "regex-automata" +version = "0.4.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f388202e4b80542a0921078cc23b6333bcf1409c1e3f86404cae4766a6131db" +dependencies = [ + "aho-corasick", + "memchr", + "regex-syntax", +] + +[[package]] +name = "regex-syntax" +version = "0.8.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" + +[[package]] +name = "sctp-proto" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d4dea4fe3384a24652f065296ac333c810dfd0c5b39b98a2214762c16aaadc3c" +dependencies = [ + "bytes", + "crc", + "fxhash", + "log", + "rand", + "slab", + "thiserror 1.0.69", +] + +[[package]] +name = "serde" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41d385c7d4ca58e59fc732af25c3983b67ac852c1a25000afe1175de458b67ad" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "serde_json" +version = "1.0.150" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + +[[package]] +name = "sha1" +version = "0.10.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3bf829a2d51ab4a5ddf1352d8470c140cadc8301b2ae1789db023f01cedd6ba" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", + "sha1-asm", +] + +[[package]] +name = "sha1-asm" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "286acebaf8b67c1130aedffad26f594eff0c1292389158135327d2e23aed582b" +dependencies = [ + "cc", +] + +[[package]] +name = "shlex" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" + +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + +[[package]] +name = "str0m" +version = "0.6.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f6110f29fbdda0516ae4ca52b62d802fbb6f874275db717a340ebb0ddd86e51c" +dependencies = [ + "combine", + "crc", + "fastrand", + "hmac", + "libc", + "once_cell", + "openssl", + "openssl-sys", + "sctp-proto", + "serde", + "sha1", + "thiserror 1.0.69", + "tracing", +] + +[[package]] +name = "subtle" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" + +[[package]] +name = "syn" +version = "2.0.118" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1b9ae57f904213ebb649ce6895b8a66c66f0203b9319718f69a5612a065b1422" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "thiserror" +version = "1.0.69" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52" +dependencies = [ + "thiserror-impl 1.0.69", +] + +[[package]] +name = "thiserror" +version = "2.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4288b5bcbc7920c07a1149a35cf9590a2aa808e0bc1eafaade0b80947865fbc4" +dependencies = [ + "thiserror-impl 2.0.18", +] + +[[package]] +name = "thiserror-impl" +version = "1.0.69" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebc4ee7f67670e9b64d05fa4253e753e016c6c95ff35b89b7941d6b856dec1d5" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "tracing" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" +dependencies = [ + "pin-project-lite", + "tracing-attributes", + "tracing-core", +] + +[[package]] +name = "tracing-attributes" +version = "0.1.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "tracing-core" +version = "0.1.36" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" +dependencies = [ + "once_cell", +] + +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + +[[package]] +name = "unicode-ident" +version = "1.0.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" + +[[package]] +name = "utf8parse" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" + +[[package]] +name = "vcpkg" +version = "0.2.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426" + +[[package]] +name = "version_check" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" + +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + +[[package]] +name = "zerocopy" +version = "0.8.54" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7cbbc0a705a0fd05cc3676525980d2bf5a9bc4adac6d6475209a7887cf59d19" +dependencies = [ + "zerocopy-derive", +] + +[[package]] +name = "zerocopy-derive" +version = "0.8.54" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e2e817b7b52d0c7358d3246da9d69935ebb18116b2b102b4230dac079b4862f5" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "zmij" +version = "1.0.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8848ee67ecc8aedbaf3e4122217aff892639231befc6a1b58d29fff4c2cabaa" diff --git a/rust/Cargo.toml b/rust/Cargo.toml index e6343cb..822fd14 100644 --- a/rust/Cargo.toml +++ b/rust/Cargo.toml @@ -1,26 +1,26 @@ [package] -name = "blax_webrtc" +name = "blax-webrtc-sfu" version = "0.1.0" edition = "2021" license = "MIT" -description = "Rust media core (str0m) for blax-software/laravel-webrtc, exposed to PHP via ext-php-rs." +description = "Rust SFU data plane for blax-software/laravel-webrtc: terminates ICE/DTLS/SRTP (str0m), forwards Opus RTP between room peers, records, and is driven by PHP over a JSON control socket." publish = false -# Built as a PHP extension (cdylib). `cargo php install` loads it into your PHP. -[lib] -crate-type = ["cdylib"] +[[bin]] +name = "blax-webrtc-sfu" +path = "src/main.rs" [dependencies] -# Sans-IO WebRTC (ICE/DTLS/SRTP/RTP), built for SFU + recording use cases. +# Sans-IO WebRTC (ICE/DTLS/SRTP/RTP) — the media core. str0m = "0.6" -# Zend bindings so PHP can drive the core (SDP/ICE/control across FFI). -ext-php-rs = "0.12" - -# Added as the core grows: -# opus = "..." # audio (de)coding for recording / transcoding -# hound / ogg = "..." # write recordings to disk -# tokio = "..." # the OpenAI-realtime bridge (server-side WebSocket) +serde = { version = "1", features = ["derive"] } +serde_json = "1" +anyhow = "1" +log = "0.4" +env_logger = "0.11" +ogg = "0.9" [profile.release] lto = true codegen-units = 1 +strip = true diff --git a/rust/README.md b/rust/README.md index c5a4d57..ac786a5 100644 --- a/rust/README.md +++ b/rust/README.md @@ -1,39 +1,65 @@ -# blax_webrtc — Rust media core +# blax-webrtc-sfu — the Rust data plane -The timing-critical WebRTC media path for `blax-software/laravel-webrtc`, so that -PHP never runs per-20ms real-time DSP. +The sidecar binary behind `Blax\WebRtc\Media\RustMediaEngine`: **backend-hosted +rooms**. PHP/Laravel owns the control plane (rooms, auth, signaling +orchestration, billing hooks); this process owns everything per-packet. -## Why Rust here +- **ICE** (Interactive Connectivity Establishment — finds a working network + path to each browser), **DTLS** (the UDP TLS handshake that negotiates keys) + and **SRTP** (the encrypted media packets) are terminated here via + [str0m](https://github.com/algesten/str0m), a sans-IO Rust WebRTC stack. +- **SFU forwarding** (Selective Forwarding Unit): each peer's Opus audio is + received once and forwarded to the other peers of its room **without + decoding** — one loop thread, one shared UDP socket, `rtc.accepts()` + demultiplexing. +- **Recording**: per-peer Ogg/Opus written straight from the depayloaded + stream — playable by ffmpeg/browsers, no transcoding, no client cooperation. +- **Negotiation**: the initial offer/answer arrives via the control socket + (from PHP signaling). Renegotiation as peers join rides each browser's + WebRTC **data channel** directly — PHP never relays it (see str0m's `chat` + example, which this follows). -Handling **connections** concurrently in PHP is a solved problem (Swoole, -FrankenPHP, ReactPHP). What is *not* proven is sustained per-packet media DSP -(decrypt SRTP + Opus + RTP jitter every 20ms) glitch-free in single-threaded PHP. -That work belongs in a language built for it. `str0m` is **sans-IO** (no threads -or async of its own — you feed it bytes), which makes it ideal to embed and drive -from PHP. +## Control protocol (PHP ⇄ sidecar) -- `str0m` owns ICE / DTLS / SRTP / RTP / Opus. -- `ext-php-rs` exposes a thin function surface to PHP. -- PHP (`Str0mMediaEngine`) forwards SDP/ICE/control across the FFI boundary and - calls back into Laravel for persistence (recordings, transmission rows). +JSON-lines over a Unix socket. Requests carry an `id` echoed by the reply; +notices are pushed without one. -## Build - -```bash -# one-time: cargo + the ext-php-rs helper -cargo install cargo-php - -cd rust -cargo php install --release # compiles the cdylib, installs it as the `blax_webrtc` extension -php -m | grep blax_webrtc # verify it loaded +``` +→ {"id":1,"cmd":"add_peer","room":"lobby","peer":"alice","offer":"v=0…","record":"/rec/alice.ogg"} +← {"id":1,"ok":true,"answer":"v=0…"} +→ {"id":2,"cmd":"mute_peer","room":"lobby","peer":"alice","muted":true} +→ {"id":3,"cmd":"record_start","room":"lobby","peer":"alice","path":"/rec/a.ogg"} (record_stop too) +→ {"id":4,"cmd":"remove_peer","room":"lobby","peer":"alice"} +→ {"id":5,"cmd":"stats"} ← rooms → peers → {connected, muted, seconds} +→ {"id":6,"cmd":"ping"} / {"id":7,"cmd":"shutdown"} +← {"event":"peer_connected","room":"lobby","peer":"alice"} +← {"event":"peer_left","room":"lobby","peer":"alice","seconds":42.5} ``` -Then set `WEBRTC_MEDIA_ENGINE=BlaxSoftware\LaravelWebRtc\Media\Str0mMediaEngine`. +## Build & test -## Status +```bash +cargo build && cargo test # 7 tests incl. an end-to-end control-socket run +./build-release.sh # static musl release + .sha256 into dist/ +``` -Scaffold. `src/lib.rs` documents the intended FFI functions -(`blax_webrtc_offer`, `..._add_ice_candidate`, `..._start_recording`, -`..._connect_peers`, `..._bridge`, `..._close`) that `Str0mMediaEngine` forwards -to. Implement them against `str0m`'s RTP/Frame API, then flip the PHP engine off -`NullMediaEngine`. +On NixOS: `nix-shell -p gcc pkg-config openssl perl gnumake musl --run ./build-release.sh`. + +## Releasing + +`BinaryManager::VERSION` (PHP) pins the release tag it downloads from. To ship: +build `dist/` artifacts (linux-x86_64 at minimum), attach both files to the +Forgejo release with that tag, and bump `VERSION` when the protocol changes. +PHP resolves binaries in this order: `webrtc.sfu.binary` override → a local +`target/{release,debug}` build (dev) → the downloaded install. + +## CLI + +``` +blax-webrtc-sfu --socket /run/sfu.sock --udp-ip 0.0.0.0 --udp-port 41000 --public-ip 203.0.113.9 +``` + +`--udp-port 0` (default) picks an ephemeral port — fine on one host, pin it in +production and publish it (UDP) on the container. `--public-ip` is what +browsers are told to reach in the ICE host candidate; it defaults to the +default-route interface address. diff --git a/rust/build-release.sh b/rust/build-release.sh new file mode 100755 index 0000000..2b405f5 --- /dev/null +++ b/rust/build-release.sh @@ -0,0 +1,45 @@ +#!/usr/bin/env bash +# Build the release artifacts for a laravel-webrtc release tag: +# blax-webrtc-sfu- (static musl binary on Linux) +# blax-webrtc-sfu-.sha256 (checksum BinaryManager verifies) +# +# Usage: ./build-release.sh [target-triple] +# default target: x86_64-unknown-linux-musl (fully static — runs on any +# Linux, glibc or musl, any distro; str0m's vendored-openssl builds into it). +# +# Attach both files to the Forgejo release whose tag matches +# BinaryManager::VERSION — that is where webrtc:install / auto_install look. +set -euo pipefail +cd "$(dirname "$0")" + +TARGET="${1:-x86_64-unknown-linux-musl}" + +case "$TARGET" in + x86_64-unknown-linux-*) PLATFORM="linux-x86_64" ;; + aarch64-unknown-linux-*) PLATFORM="linux-aarch64" ;; + x86_64-apple-darwin) PLATFORM="darwin-x86_64" ;; + aarch64-apple-darwin) PLATFORM="darwin-aarch64" ;; + *) echo "unmapped target triple: $TARGET" >&2; exit 1 ;; +esac + +if command -v rustup >/dev/null; then + rustup target add "$TARGET" >/dev/null +fi + +# musl static linking needs a musl C toolchain for vendored OpenSSL. +if [[ "$TARGET" == *musl* ]] && ! command -v musl-gcc >/dev/null; then + echo "musl-gcc not found — install musl-tools (debian) / musl (nix: nix-shell -p musl gcc pkg-config perl gnumake)" >&2 +fi + +cargo build --release --target "$TARGET" + +mkdir -p dist +OUT="dist/blax-webrtc-sfu-$PLATFORM" +cp "target/$TARGET/release/blax-webrtc-sfu" "$OUT" +(cd dist && sha256sum "$(basename "$OUT")" > "$(basename "$OUT").sha256") + +echo +echo "Release artifacts:" +ls -la dist/ +echo +echo "Upload both files to the release tagged $(grep -m1 '^version' Cargo.toml | cut -d'"' -f2 | sed 's/^/v/')." diff --git a/rust/src/control.rs b/rust/src/control.rs new file mode 100644 index 0000000..ea4ac9b --- /dev/null +++ b/rust/src/control.rs @@ -0,0 +1,106 @@ +//! Unix-socket control plane: JSON-lines requests in, replies + pushed events out. +//! +//! Each connection gets a reader thread (parses requests, forwards them to the +//! SFU loop, relays the reply) and a writer thread (serializes replies AND +//! broadcast notices onto the stream, one JSON object per line). + +use std::io::{BufRead, BufReader, Write}; +use std::os::unix::net::{UnixListener, UnixStream}; +use std::sync::mpsc::{self, Sender}; +use std::sync::{Arc, Mutex}; +use std::thread; + +use log::warn; + +use crate::proto::{Notice, Reply, Request}; +use crate::sfu::CtrlMsg; + +/// Every connected control client's outbound line channel (for notices). +pub type Subscribers = Arc>>>; + +/// Fan notices out to every connected control client. +pub fn broadcast_notices(rx: mpsc::Receiver, subscribers: Subscribers) { + for notice in rx { + let json = serde_json::to_string(¬ice).expect("notice to serialize"); + subscribers + .lock() + .unwrap() + .retain(|tx| tx.send(json.clone()).is_ok()); + } +} + +/// Accept control connections forever (runs on its own thread). +pub fn serve(listener: UnixListener, commands: Sender, subscribers: Subscribers) { + for stream in listener.incoming() { + match stream { + Ok(stream) => { + let commands = commands.clone(); + let subscribers = subscribers.clone(); + thread::spawn(move || handle_connection(stream, commands, subscribers)); + } + Err(e) => warn!("control accept failed: {e}"), + } + } +} + +fn handle_connection(stream: UnixStream, commands: Sender, subscribers: Subscribers) { + let write_half = match stream.try_clone() { + Ok(s) => s, + Err(e) => { + warn!("control connection clone failed: {e}"); + return; + } + }; + + // One writer thread serializes replies + notices; dies when senders drop + // or the peer closes (write error). + let (out_tx, out_rx) = mpsc::channel::(); + subscribers.lock().unwrap().push(out_tx.clone()); + + let writer = thread::spawn(move || { + let mut stream = write_half; + for line in out_rx { + if writeln!(stream, "{line}").and_then(|()| stream.flush()).is_err() { + break; + } + } + }); + + let reader = BufReader::new(stream); + for line in reader.lines() { + let Ok(line) = line else { break }; + if line.trim().is_empty() { + continue; + } + + let request: Request = match serde_json::from_str(&line) { + Ok(request) => request, + Err(e) => { + let reply = Reply::error(0, format!("bad request: {e}")); + let _ = out_tx.send(serde_json::to_string(&reply).unwrap()); + continue; + } + }; + + let (reply_tx, reply_rx) = mpsc::channel(); + if commands + .send(CtrlMsg { + request, + reply: reply_tx, + }) + .is_err() + { + break; // SFU loop gone (shutdown) + } + + match reply_rx.recv() { + Ok(reply) => { + let _ = out_tx.send(serde_json::to_string(&reply).unwrap()); + } + Err(_) => break, + } + } + + drop(out_tx); + let _ = writer.join(); +} diff --git a/rust/src/lib.rs b/rust/src/lib.rs index 4d34092..460437f 100644 --- a/rust/src/lib.rs +++ b/rust/src/lib.rs @@ -1,57 +1,13 @@ -//! blax_webrtc — the Rust media core for `blax-software/laravel-webrtc`. +//! blax-webrtc-sfu — the Rust data plane for `blax-software/laravel-webrtc`. //! -//! Owns the timing-critical WebRTC media path (ICE / DTLS / SRTP / RTP / Opus) -//! via `str0m` (a sans-IO Rust WebRTC implementation built for SFU + recording), -//! and exposes a thin surface to PHP via `ext-php-rs`. PHP drives signaling and -//! orchestration; the per-20ms media DSP runs here in Rust, so it never stalls -//! the PHP event loop or fights PHP's GC. -//! -//! STATUS: scaffold. The functions below are the intended FFI surface that -//! `BlaxSoftware\LaravelWebRtc\Media\Str0mMediaEngine` forwards to. They are not -//! implemented yet — this file documents the seam and keeps the crate buildable -//! once the dependencies are added. -//! -//! Build (once implemented): -//! cd rust && cargo php install --release -//! which compiles the cdylib and installs it as the `blax_webrtc` PHP extension. +//! PHP/Laravel owns the CONTROL plane (rooms, auth, signaling orchestration, +//! billing hooks) and drives this sidecar over a JSON-lines Unix socket. This +//! crate owns everything per-packet: ICE (connectivity), DTLS (key exchange), +//! SRTP (encrypted media), RTP forwarding between room peers (SFU — no +//! decoding), per-peer Ogg/Opus recording, and connection stats/usage events. -// use ext_php_rs::prelude::*; -// -// /// Accept a browser SDP offer for `peer_id`; return the answer SDP. -// #[php_function] -// pub fn blax_webrtc_offer(peer_id: &str, sdp_offer: &str) -> String { -// todo!("create a str0m Rtc, apply the remote offer, return the local answer") -// } -// -// /// Feed a trickle-ICE candidate for `peer_id`. -// #[php_function] -// pub fn blax_webrtc_add_ice_candidate(peer_id: &str, candidate: &str) { -// todo!("parse + add the candidate to the peer's str0m Rtc") -// } -// -// /// Start recording `peer_id`'s inbound audio to `path` (Opus/WAV). -// #[php_function] -// pub fn blax_webrtc_start_recording(peer_id: &str, path: &str) { -// todo!("tee the peer's decoded audio frames to a file writer") -// } -// -// #[php_function] -// pub fn blax_webrtc_stop_recording(peer_id: &str) { todo!() } -// -// /// Forward media between two peers (party-to-party / group calls). -// #[php_function] -// pub fn blax_webrtc_connect_peers(peer_a: &str, peer_b: &str) { todo!() } -// -// /// Bridge `peer_id` to an external realtime provider (e.g. OpenAI Realtime over -// /// a server-side WebSocket). The browser never sees the provider. -// #[php_function] -// pub fn blax_webrtc_bridge(peer_id: &str, options_json: &str) { todo!() } -// -// /// Close + free a peer. -// #[php_function] -// pub fn blax_webrtc_close(peer_id: &str) { todo!() } -// -// #[php_module] -// pub fn get_module(module: ModuleBuilder) -> ModuleBuilder { -// module -// } +pub mod control; +pub mod proto; +pub mod recording; +pub mod server; +pub mod sfu; diff --git a/rust/src/main.rs b/rust/src/main.rs new file mode 100644 index 0000000..466b29c --- /dev/null +++ b/rust/src/main.rs @@ -0,0 +1,55 @@ +use std::net::IpAddr; +use std::path::PathBuf; + +use anyhow::{bail, Context}; +use blax_webrtc_sfu::server::{self, Config}; + +const USAGE: &str = "blax-webrtc-sfu — Rust SFU data plane for blax-software/laravel-webrtc + +USAGE: + blax-webrtc-sfu [OPTIONS] + +OPTIONS: + --socket Control socket path [default: /tmp/blax-webrtc-sfu.sock] + --udp-ip Media UDP bind ip [default: 0.0.0.0] + --udp-port Media UDP port [default: 0 = ephemeral] + --public-ip IP advertised to browsers in ICE candidates + [default: the default-route interface address] + -h, --help Show this help +"; + +fn main() -> anyhow::Result<()> { + env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).init(); + + let mut config = Config { + socket: PathBuf::from("/tmp/blax-webrtc-sfu.sock"), + udp_ip: IpAddr::from([0, 0, 0, 0]), + udp_port: 0, + public_ip: None, + }; + + let mut args = std::env::args().skip(1); + while let Some(flag) = args.next() { + let mut value = |name: &str| { + args.next() + .with_context(|| format!("{name} requires a value")) + }; + match flag.as_str() { + "--socket" => config.socket = PathBuf::from(value("--socket")?), + "--udp-ip" => config.udp_ip = value("--udp-ip")?.parse().context("--udp-ip")?, + "--udp-port" => { + config.udp_port = value("--udp-port")?.parse().context("--udp-port")? + } + "--public-ip" => { + config.public_ip = Some(value("--public-ip")?.parse().context("--public-ip")?) + } + "-h" | "--help" => { + print!("{USAGE}"); + return Ok(()); + } + other => bail!("unknown option: {other}\n\n{USAGE}"), + } + } + + server::run(config) +} diff --git a/rust/src/proto.rs b/rust/src/proto.rs new file mode 100644 index 0000000..c4aa93f --- /dev/null +++ b/rust/src/proto.rs @@ -0,0 +1,179 @@ +//! JSON-lines control protocol between PHP (`RustMediaEngine`) and this sidecar. +//! +//! Requests carry an `id` the reply echoes back; events have no `id` and are +//! pushed to every connected control client. + +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Deserialize)] +pub struct Request { + pub id: u64, + #[serde(flatten)] + pub command: Command, +} + +#[derive(Debug, Deserialize)] +#[serde(tag = "cmd", rename_all = "snake_case")] +pub enum Command { + /// Liveness probe (also used by PHP to detect an already-running sidecar). + Ping, + /// Accept a browser SDP offer for `peer` in `room`; reply carries the answer. + /// Further renegotiation (new tracks) rides the WebRTC data channel directly. + AddPeer { + room: String, + peer: String, + offer: String, + #[serde(default)] + record: Option, + }, + /// Kick a peer (disconnects its transport; emits `peer_left`). + RemovePeer { room: String, peer: String }, + /// Server-side mute: stop forwarding the peer's audio to the room. + MutePeer { + room: String, + peer: String, + muted: bool, + }, + /// Start recording a peer's inbound audio to `path` (Ogg/Opus). + RecordStart { + room: String, + peer: String, + path: String, + }, + /// Stop and finalize a peer's recording. + RecordStop { room: String, peer: String }, + /// Snapshot of rooms/peers (connection state, mute, connected seconds). + Stats, + /// Graceful stop. + Shutdown, +} + +#[derive(Debug, Serialize)] +pub struct Reply { + pub id: u64, + pub ok: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub answer: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub rooms: Option>, +} + +impl Reply { + pub fn ok(id: u64) -> Self { + Reply { + id, + ok: true, + answer: None, + error: None, + rooms: None, + } + } + + pub fn answer(id: u64, sdp: String) -> Self { + Reply { + answer: Some(sdp), + ..Reply::ok(id) + } + } + + pub fn error(id: u64, error: impl Into) -> Self { + Reply { + ok: false, + error: Some(error.into()), + ..Reply::ok(id) + } + } + + pub fn stats(id: u64, rooms: Vec) -> Self { + Reply { + rooms: Some(rooms), + ..Reply::ok(id) + } + } +} + +#[derive(Debug, Serialize)] +pub struct RoomStats { + pub room: String, + pub peers: Vec, +} + +#[derive(Debug, Serialize)] +pub struct PeerStats { + pub peer: String, + pub connected: bool, + pub muted: bool, + pub seconds: f64, +} + +/// Server-pushed notifications (no `id`). +#[derive(Debug, Clone, Serialize)] +#[serde(tag = "event", rename_all = "snake_case")] +pub enum Notice { + PeerConnected { room: String, peer: String }, + PeerLeft { room: String, peer: String, seconds: f64 }, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parses_add_peer() { + let req: Request = serde_json::from_str( + r#"{"id":7,"cmd":"add_peer","room":"r1","peer":"p1","offer":"v=0...","record":"/tmp/rec/p1.ogg"}"#, + ) + .unwrap(); + assert_eq!(req.id, 7); + match req.command { + Command::AddPeer { + room, + peer, + offer, + record, + } => { + assert_eq!(room, "r1"); + assert_eq!(peer, "p1"); + assert_eq!(offer, "v=0..."); + assert_eq!(record.as_deref(), Some("/tmp/rec/p1.ogg")); + } + other => panic!("wrong command: {other:?}"), + } + } + + #[test] + fn record_defaults_to_none() { + let req: Request = + serde_json::from_str(r#"{"id":1,"cmd":"add_peer","room":"r","peer":"p","offer":"x"}"#) + .unwrap(); + match req.command { + Command::AddPeer { record, .. } => assert!(record.is_none()), + other => panic!("wrong command: {other:?}"), + } + } + + #[test] + fn reply_serializes_compactly() { + let json = serde_json::to_string(&Reply::answer(3, "v=0".into())).unwrap(); + assert_eq!(json, r#"{"id":3,"ok":true,"answer":"v=0"}"#); + + let json = serde_json::to_string(&Reply::error(4, "no such peer")).unwrap(); + assert_eq!(json, r#"{"id":4,"ok":false,"error":"no such peer"}"#); + } + + #[test] + fn notice_has_event_tag() { + let json = serde_json::to_string(&Notice::PeerLeft { + room: "r".into(), + peer: "p".into(), + seconds: 1.5, + }) + .unwrap(); + assert_eq!( + json, + r#"{"event":"peer_left","room":"r","peer":"p","seconds":1.5}"# + ); + } +} diff --git a/rust/src/recording.rs b/rust/src/recording.rs new file mode 100644 index 0000000..fc918a5 --- /dev/null +++ b/rust/src/recording.rs @@ -0,0 +1,150 @@ +//! Server-side per-peer call recording: Opus frames (as depayloaded by str0m) +//! muxed into an Ogg container, playable by browsers/ffmpeg without transcoding. + +use std::fs::File; +use std::io::Error; +use std::path::Path; + +use ogg::{PacketWriteEndInfo, PacketWriter}; + +const SAMPLE_RATE: f64 = 48_000.0; +/// Standard Opus pre-skip (3840 samples = 80ms at 48kHz). +const PRE_SKIP: u16 = 3840; + +pub struct OggOpusWriter { + writer: PacketWriter<'static, File>, + serial: u32, + /// RTP media time (seconds) of the first frame — granule positions are + /// computed relative to it, so pauses (DTX) keep the timeline correct. + base_time: Option, + granule: u64, + finished: bool, +} + +impl OggOpusWriter { + pub fn create(path: &Path, channels: u8) -> Result { + if let Some(dir) = path.parent() { + std::fs::create_dir_all(dir)?; + } + + let mut writer = PacketWriter::new(File::create(path)?); + // Stream serial only needs to be unique within the file; one stream per file. + let serial: u32 = 0x626c_6178; // "blax" + + writer.write_packet(opus_head(channels), serial, PacketWriteEndInfo::EndPage, 0)?; + writer.write_packet(opus_tags(), serial, PacketWriteEndInfo::EndPage, 0)?; + + Ok(OggOpusWriter { + writer, + serial, + base_time: None, + granule: 0, + finished: false, + }) + } + + /// Append one Opus frame. `time_seconds` is the frame's RTP media time + /// (`MediaData::time.as_seconds()`). + pub fn write_frame(&mut self, frame: &[u8], time_seconds: f64) -> Result<(), Error> { + if self.finished { + return Ok(()); + } + + let base = *self.base_time.get_or_insert(time_seconds); + let elapsed = (time_seconds - base).max(0.0); + // Granule = absolute decoded sample position at 48kHz. One 20ms frame = 960 + // samples; deriving from RTP time keeps sync even with DTX/packet loss. + let granule = (elapsed * SAMPLE_RATE).round() as u64 + 960; + self.granule = granule.max(self.granule); + + self.writer.write_packet( + frame.to_vec(), + self.serial, + PacketWriteEndInfo::NormalPacket, + self.granule, + ) + } + + /// Close the logical Ogg stream (writes the end-of-stream page). + pub fn finish(&mut self) -> Result<(), Error> { + if self.finished { + return Ok(()); + } + self.finished = true; + self.writer.write_packet( + Vec::new(), + self.serial, + PacketWriteEndInfo::EndStream, + self.granule, + ) + } +} + +impl Drop for OggOpusWriter { + fn drop(&mut self) { + let _ = self.finish(); + } +} + +/// +fn opus_head(channels: u8) -> Vec { + let mut head = Vec::with_capacity(19); + head.extend_from_slice(b"OpusHead"); + head.push(1); // version + head.push(channels); + head.extend_from_slice(&PRE_SKIP.to_le_bytes()); + head.extend_from_slice(&48_000u32.to_le_bytes()); // input sample rate + head.extend_from_slice(&0i16.to_le_bytes()); // output gain + head.push(0); // channel mapping family + head +} + +fn opus_tags() -> Vec { + let vendor = b"blax-webrtc-sfu"; + let mut tags = Vec::new(); + tags.extend_from_slice(b"OpusTags"); + tags.extend_from_slice(&(vendor.len() as u32).to_le_bytes()); + tags.extend_from_slice(vendor); + tags.extend_from_slice(&0u32.to_le_bytes()); // user comment count + tags +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn writes_a_playable_ogg_opus_skeleton() { + let dir = std::env::temp_dir().join("blax-webrtc-sfu-test"); + let path = dir.join("rec.ogg"); + let _ = std::fs::remove_file(&path); + + let mut w = OggOpusWriter::create(&path, 1).unwrap(); + // Two fake 20ms frames. + w.write_frame(&[0xFC, 0xFF, 0xFE], 10.0).unwrap(); + w.write_frame(&[0xFC, 0xFF, 0xFE], 10.02).unwrap(); + w.finish().unwrap(); + + let bytes = std::fs::read(&path).unwrap(); + assert_eq!(&bytes[..4], b"OggS", "ogg capture pattern"); + let head = bytes + .windows(8) + .position(|win| win == b"OpusHead") + .expect("OpusHead present"); + let tags = bytes + .windows(8) + .position(|win| win == b"OpusTags") + .expect("OpusTags present"); + assert!(head < tags); + } + + #[test] + fn granule_tracks_rtp_time() { + let dir = std::env::temp_dir().join("blax-webrtc-sfu-test"); + let mut w = OggOpusWriter::create(&dir.join("granule.ogg"), 1).unwrap(); + w.write_frame(&[0], 5.0).unwrap(); + assert_eq!(w.granule, 960); + w.write_frame(&[0], 6.0).unwrap(); // 1s gap (DTX) => 48000 samples later + assert_eq!(w.granule, 48_960); + } +} diff --git a/rust/src/server.rs b/rust/src/server.rs new file mode 100644 index 0000000..3fd4c64 --- /dev/null +++ b/rust/src/server.rs @@ -0,0 +1,85 @@ +//! Wires the pieces together: UDP media socket + Unix control socket + SFU loop. +//! Shared by `main.rs` and the integration tests. + +use std::net::{IpAddr, SocketAddr, UdpSocket}; +use std::os::unix::net::{UnixListener, UnixStream}; +use std::path::PathBuf; +use std::sync::mpsc; +use std::sync::{Arc, Mutex}; +use std::thread; + +use anyhow::{bail, Context}; +use log::info; + +use crate::control; +use crate::sfu::Sfu; + +pub struct Config { + /// Unix socket path for the JSON control protocol. + pub socket: PathBuf, + /// Bind address for the shared media UDP socket. + pub udp_ip: IpAddr, + /// UDP port (0 = ephemeral). + pub udp_port: u16, + /// IP advertised to browsers in the host ICE candidate. Defaults to the + /// default-route interface address (or the bind ip when it is specific). + pub public_ip: Option, +} + +/// Run the sidecar until a `shutdown` control command. Blocking. +pub fn run(config: Config) -> anyhow::Result<()> { + // Refuse to clobber a live sidecar; clean up a stale socket file. + if config.socket.exists() { + if UnixStream::connect(&config.socket).is_ok() { + bail!( + "another sidecar is already listening on {}", + config.socket.display() + ); + } + std::fs::remove_file(&config.socket).context("removing stale control socket")?; + } + if let Some(dir) = config.socket.parent() { + std::fs::create_dir_all(dir).context("creating control socket directory")?; + } + + let socket = UdpSocket::bind((config.udp_ip, config.udp_port)).context("binding UDP")?; + let local = socket.local_addr()?; + + let public_ip = config + .public_ip + .or_else(|| (!config.udp_ip.is_unspecified()).then_some(config.udp_ip)) + .or_else(default_route_ip) + .unwrap_or_else(|| IpAddr::from([127, 0, 0, 1])); + let advertise = SocketAddr::new(public_ip, local.port()); + + let (cmd_tx, cmd_rx) = mpsc::channel(); + let (notice_tx, notice_rx) = mpsc::channel(); + let subscribers: control::Subscribers = Arc::new(Mutex::new(Vec::new())); + + let listener = UnixListener::bind(&config.socket).context("binding control socket")?; + { + let subscribers = subscribers.clone(); + thread::spawn(move || control::serve(listener, cmd_tx, subscribers)); + } + thread::spawn(move || control::broadcast_notices(notice_rx, subscribers)); + + info!( + "blax-webrtc-sfu ready — control {} · media udp {} (advertising {})", + config.socket.display(), + local, + advertise + ); + + Sfu::new(socket, advertise, cmd_rx, notice_tx).run(); + + let _ = std::fs::remove_file(&config.socket); + Ok(()) +} + +/// The local IP of the default route (no packets are sent — UDP connect only +/// resolves the source address). +fn default_route_ip() -> Option { + let probe = UdpSocket::bind("0.0.0.0:0").ok()?; + probe.connect("8.8.8.8:80").ok()?; + Some(probe.local_addr().ok()?.ip()) +} diff --git a/rust/src/sfu.rs b/rust/src/sfu.rs new file mode 100644 index 0000000..9f2530f --- /dev/null +++ b/rust/src/sfu.rs @@ -0,0 +1,807 @@ +//! The SFU (Selective Forwarding Unit) media loop. +//! +//! One thread owns every peer's str0m `Rtc` plus the single shared UDP socket +//! (peers are demultiplexed via `rtc.accepts()`). Media is forwarded between +//! peers of the same room WITHOUT decoding. Commands arrive from the control +//! socket over an mpsc channel; notices flow back the other way. +//! +//! The negotiation model follows str0m's canonical SFU example (`chat.rs`): +//! the initial browser offer/answer goes through the control plane (PHP +//! signaling); every renegotiation after that (new tracks as peers join) is +//! exchanged over the WebRTC DATA CHANNEL directly — PHP never relays it. + +use std::collections::VecDeque; +use std::io::ErrorKind; +use std::net::{SocketAddr, UdpSocket}; +use std::ops::Deref; +use std::path::Path; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::mpsc::{Receiver, Sender, TryRecvError}; +use std::sync::{Arc, Weak}; +use std::time::{Duration, Instant}; + +use log::{debug, info, warn}; +use str0m::change::{SdpAnswer, SdpOffer, SdpPendingOffer}; +use str0m::channel::{ChannelData, ChannelId}; +use str0m::media::{Direction, KeyframeRequest, KeyframeRequestKind, MediaData, MediaKind, Mid, Rid}; +use str0m::net::{Protocol, Receive}; +use str0m::{Candidate, Event, IceConnectionState, Input, Output, Rtc}; + +use crate::proto::{Command, Notice, PeerStats, Reply, Request, RoomStats}; +use crate::recording::OggOpusWriter; + +/// A control request paired with where to send its reply. +pub struct CtrlMsg { + pub request: Request, + pub reply: Sender, +} + +pub struct Sfu { + socket: UdpSocket, + /// The candidate address advertised to browsers (public ip + UDP port). + advertise: SocketAddr, + commands: Receiver, + notices: Sender, + clients: Vec, +} + +impl Sfu { + pub fn new( + socket: UdpSocket, + advertise: SocketAddr, + commands: Receiver, + notices: Sender, + ) -> Self { + Sfu { + socket, + advertise, + commands, + notices, + clients: Vec::new(), + } + } + + /// Run until a `shutdown` command arrives. + pub fn run(mut self) { + let mut to_propagate: VecDeque = VecDeque::new(); + let mut buf = vec![0; 2000]; + + loop { + self.reap_dead_clients(); + + match self.drain_commands() { + ControlFlow::Continue => {} + ControlFlow::Shutdown => { + info!("shutdown command received, stopping SFU loop"); + return; + } + } + + // Poll clients until they all return a timeout. + let mut timeout = Instant::now() + Duration::from_millis(100); + for client in self.clients.iter_mut() { + let t = poll_until_timeout(client, &mut to_propagate, &self.socket); + timeout = timeout.min(t); + } + + // If we have an item to propagate, do that before sleeping. + if let Some(p) = to_propagate.pop_front() { + self.propagate(p); + continue; + } + + let duration = (timeout - Instant::now()).max(Duration::from_millis(1)); + self.socket + .set_read_timeout(Some(duration)) + .expect("setting socket read timeout"); + + if let Some(input) = read_socket_input(&self.socket, &mut buf) { + // rtc.accepts() demultiplexes the shared UDP socket to the right peer. + if let Some(client) = self.clients.iter_mut().find(|c| c.accepts(&input)) { + client.handle_input(input); + } else { + // Common: first STUN can arrive before add_peer completes. + debug!("no client accepts UDP input: {:?}", input); + } + } + + // Drive time forward in all clients. + let now = Instant::now(); + for client in &mut self.clients { + client.handle_input(Input::Timeout(now)); + } + + self.notify_connected(); + } + } + + fn drain_commands(&mut self) -> ControlFlow { + loop { + let msg = match self.commands.try_recv() { + Ok(msg) => msg, + Err(TryRecvError::Empty) => return ControlFlow::Continue, + Err(TryRecvError::Disconnected) => { + warn!("control channel disconnected, stopping"); + return ControlFlow::Shutdown; + } + }; + + let id = msg.request.id; + let reply = match msg.request.command { + Command::Ping => Reply::ok(id), + Command::Shutdown => { + let _ = msg.reply.send(Reply::ok(id)); + return ControlFlow::Shutdown; + } + Command::AddPeer { + room, + peer, + offer, + record, + } => self.add_peer(id, room, peer, &offer, record.as_deref()), + Command::RemovePeer { room, peer } => self.with_client(id, &room, &peer, |c| { + c.rtc.disconnect(); + Reply::ok(id) + }), + Command::MutePeer { room, peer, muted } => { + self.with_client(id, &room, &peer, |c| { + c.muted = muted; + Reply::ok(id) + }) + } + Command::RecordStart { room, peer, path } => { + self.with_client(id, &room, &peer, |c| match c.start_recording(&path) { + Ok(()) => Reply::ok(id), + Err(e) => Reply::error(id, format!("recording failed: {e}")), + }) + } + Command::RecordStop { room, peer } => self.with_client(id, &room, &peer, |c| { + c.stop_recording(); + Reply::ok(id) + }), + Command::Stats => Reply::stats(id, self.stats()), + }; + + let _ = msg.reply.send(reply); + } + } + + fn add_peer( + &mut self, + id: u64, + room: String, + peer: String, + offer_sdp: &str, + record: Option<&str>, + ) -> Reply { + let offer = match SdpOffer::from_sdp_string(offer_sdp) { + Ok(offer) => offer, + Err(e) => return Reply::error(id, format!("bad offer sdp: {e}")), + }; + + let mut rtc = Rtc::builder().build(); + + let candidate = match Candidate::host(self.advertise, "udp") { + Ok(candidate) => candidate, + Err(e) => return Reply::error(id, format!("bad host candidate: {e}")), + }; + rtc.add_local_candidate(candidate); + + let answer = match rtc.sdp_api().accept_offer(offer) { + Ok(answer) => answer, + Err(e) => return Reply::error(id, format!("offer rejected: {e}")), + }; + + // A rejoin with the same peer id replaces the old connection. + if let Some(old) = self + .clients + .iter_mut() + .find(|c| c.room == room && c.peer == peer) + { + info!("replacing existing peer {room}/{peer}"); + old.rtc.disconnect(); + } + + let mut client = Client::new(room.clone(), peer.clone(), rtc); + if let Some(path) = record { + if let Err(e) = client.start_recording(path) { + warn!("recording for {room}/{peer} failed to start: {e}"); + } + } + + // Open the tracks already present in the room towards the new peer. + for track in self + .clients + .iter() + .filter(|c| c.room == room) + .flat_map(|c| c.tracks_in.iter()) + { + client.handle_track_open(Arc::downgrade(&track.id)); + } + + info!("peer {room}/{peer} added ({} in room)", 1 + self + .clients + .iter() + .filter(|c| c.room == room) + .count()); + self.clients.push(client); + + Reply::answer(id, answer.to_sdp_string()) + } + + fn with_client( + &mut self, + id: u64, + room: &str, + peer: &str, + f: impl FnOnce(&mut Client) -> Reply, + ) -> Reply { + match self + .clients + .iter_mut() + .find(|c| c.room == room && c.peer == peer) + { + Some(client) => f(client), + None => Reply::error(id, format!("no such peer: {room}/{peer}")), + } + } + + fn stats(&self) -> Vec { + let mut rooms: Vec = Vec::new(); + for client in &self.clients { + let peer = PeerStats { + peer: client.peer.clone(), + connected: client.connected_at.is_some(), + muted: client.muted, + seconds: client + .connected_at + .map(|t| t.elapsed().as_secs_f64()) + .unwrap_or(0.0), + }; + match rooms.iter_mut().find(|r| r.room == client.room) { + Some(room) => room.peers.push(peer), + None => rooms.push(RoomStats { + room: client.room.clone(), + peers: vec![peer], + }), + } + } + rooms + } + + /// Emit `peer_connected` notices for clients that just reached Connected. + fn notify_connected(&mut self) { + for client in &mut self.clients { + if client.connected_at.is_some() && !client.notified_connected { + client.notified_connected = true; + let _ = self.notices.send(Notice::PeerConnected { + room: client.room.clone(), + peer: client.peer.clone(), + }); + } + } + } + + fn reap_dead_clients(&mut self) { + if self.clients.iter().all(|c| c.rtc.is_alive()) { + return; + } + + let clients = std::mem::take(&mut self.clients); + for mut client in clients { + if client.rtc.is_alive() { + self.clients.push(client); + continue; + } + + client.stop_recording(); + info!("peer {}/{} left", client.room, client.peer); + let _ = self.notices.send(Notice::PeerLeft { + room: client.room.clone(), + peer: client.peer.clone(), + seconds: client + .connected_at + .map(|t| t.elapsed().as_secs_f64()) + .unwrap_or(0.0), + }); + } + } + + /// Send one propagated item to the other clients in the origin's room. + fn propagate(&mut self, propagated: Propagated) { + let Some(client_id) = propagated.client_id() else { + return; + }; + + let Some(origin) = self.clients.iter().find(|c| c.id == client_id) else { + return; // origin left in the meantime + }; + let (room, origin_muted) = (origin.room.clone(), origin.muted); + + for client in &mut self.clients { + if client.id == client_id || client.room != room { + continue; + } + + match &propagated { + Propagated::TrackOpen(_, track_in) => client.handle_track_open(track_in.clone()), + Propagated::MediaData(_, data) => { + if !origin_muted { + client.handle_media_data_out(client_id, data) + } + } + Propagated::KeyframeRequest(_, req, origin_id, mid_in) => { + // Only the origin client handles the keyframe request. + if *origin_id == client.id { + client.handle_keyframe_request(*req, *mid_in) + } + } + Propagated::Noop | Propagated::Timeout(_) => {} + } + } + } +} + +enum ControlFlow { + Continue, + Shutdown, +} + +/// Poll all the output from the client until it returns a timeout. +fn poll_until_timeout( + client: &mut Client, + queue: &mut VecDeque, + socket: &UdpSocket, +) -> Instant { + loop { + if !client.rtc.is_alive() { + return Instant::now(); + } + + let propagated = client.poll_output(socket); + + if let Propagated::Timeout(t) = propagated { + return t; + } + + queue.push_back(propagated) + } +} + +fn read_socket_input<'a>(socket: &UdpSocket, buf: &'a mut Vec) -> Option> { + buf.resize(2000, 0); + + match socket.recv_from(buf) { + Ok((n, source)) => { + buf.truncate(n); + + let Ok(contents) = buf.as_slice().try_into() else { + return None; + }; + + Some(Input::Receive( + Instant::now(), + Receive { + proto: Protocol::Udp, + source, + destination: socket.local_addr().unwrap(), + contents, + }, + )) + } + + Err(e) => match e.kind() { + ErrorKind::WouldBlock | ErrorKind::TimedOut => None, + _ => panic!("UdpSocket read failed: {e:?}"), + }, + } +} + +struct Client { + id: ClientId, + room: String, + peer: String, + rtc: Rtc, + pending: Option, + cid: Option, + tracks_in: Vec, + tracks_out: Vec, + chosen_rid: Option, + muted: bool, + recorder: Option, + connected_at: Option, + notified_connected: bool, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct ClientId(u64); + +impl Deref for ClientId { + type Target = u64; + + fn deref(&self) -> &Self::Target { + &self.0 + } +} + +struct TrackIn { + origin: ClientId, + mid: Mid, + kind: MediaKind, +} + +struct TrackInEntry { + id: Arc, + last_keyframe_request: Option, +} + +struct TrackOut { + track_in: Weak, + state: TrackOutState, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum TrackOutState { + ToOpen, + Negotiating(Mid), + Open(Mid), +} + +impl TrackOut { + fn mid(&self) -> Option { + match self.state { + TrackOutState::ToOpen => None, + TrackOutState::Negotiating(m) | TrackOutState::Open(m) => Some(m), + } + } +} + +impl Client { + fn new(room: String, peer: String, rtc: Rtc) -> Client { + static ID_COUNTER: AtomicU64 = AtomicU64::new(0); + let next_id = ID_COUNTER.fetch_add(1, Ordering::SeqCst); + Client { + id: ClientId(next_id), + room, + peer, + rtc, + pending: None, + cid: None, + tracks_in: vec![], + tracks_out: vec![], + chosen_rid: None, + muted: false, + recorder: None, + connected_at: None, + notified_connected: false, + } + } + + fn start_recording(&mut self, path: &str) -> Result<(), std::io::Error> { + self.recorder = Some(OggOpusWriter::create(Path::new(path), 1)?); + Ok(()) + } + + fn stop_recording(&mut self) { + if let Some(mut recorder) = self.recorder.take() { + let _ = recorder.finish(); + } + } + + fn accepts(&self, input: &Input) -> bool { + self.rtc.accepts(input) + } + + fn handle_input(&mut self, input: Input) { + if !self.rtc.is_alive() { + return; + } + + if let Err(e) = self.rtc.handle_input(input) { + warn!("client {}/{} disconnected: {:?}", self.room, self.peer, e); + self.rtc.disconnect(); + } + } + + fn poll_output(&mut self, socket: &UdpSocket) -> Propagated { + if !self.rtc.is_alive() { + return Propagated::Noop; + } + + // Incoming tracks from room peers cause new entries in tracks_out that + // need SDP negotiation with this client (over its data channel). + if self.negotiate_if_needed() { + return Propagated::Noop; + } + + match self.rtc.poll_output() { + Ok(output) => self.handle_output(output, socket), + Err(e) => { + warn!( + "client {}/{} poll_output failed: {:?}", + self.room, self.peer, e + ); + self.rtc.disconnect(); + Propagated::Noop + } + } + } + + fn handle_output(&mut self, output: Output, socket: &UdpSocket) -> Propagated { + match output { + Output::Transmit(transmit) => { + if let Err(e) = socket.send_to(&transmit.contents, transmit.destination) { + warn!("UDP send_to failed: {e:?}"); + } + Propagated::Noop + } + Output::Timeout(t) => Propagated::Timeout(t), + Output::Event(e) => match e { + Event::Connected => { + self.connected_at = Some(Instant::now()); + Propagated::Noop + } + Event::IceConnectionStateChange(v) => { + if v == IceConnectionState::Disconnected { + self.rtc.disconnect(); + } + Propagated::Noop + } + Event::MediaAdded(e) => self.handle_media_added(e.mid, e.kind), + Event::MediaData(data) => self.handle_media_data_in(data), + Event::KeyframeRequest(req) => self.handle_incoming_keyframe_req(req), + Event::ChannelOpen(cid, _) => { + self.cid = Some(cid); + Propagated::Noop + } + Event::ChannelData(data) => self.handle_channel_data(data), + _ => Propagated::Noop, + }, + } + } + + fn handle_media_added(&mut self, mid: Mid, kind: MediaKind) -> Propagated { + let track_in = TrackInEntry { + id: Arc::new(TrackIn { + origin: self.id, + mid, + kind, + }), + last_keyframe_request: None, + }; + + let weak = Arc::downgrade(&track_in.id); + self.tracks_in.push(track_in); + + Propagated::TrackOpen(self.id, weak) + } + + fn handle_media_data_in(&mut self, data: MediaData) -> Propagated { + if !data.contiguous { + self.request_keyframe_throttled(data.mid, data.rid, KeyframeRequestKind::Fir); + } + + if !self.muted { + if let Some(recorder) = &mut self.recorder { + if data.params.spec().codec.is_audio() { + if let Err(e) = recorder.write_frame(&data.data, data.time.as_seconds()) { + warn!("recording write failed for {}/{}: {e}", self.room, self.peer); + self.recorder = None; + } + } + } + } + + Propagated::MediaData(self.id, data) + } + + fn request_keyframe_throttled( + &mut self, + mid: Mid, + rid: Option, + kind: KeyframeRequestKind, + ) { + let Some(mut writer) = self.rtc.writer(mid) else { + return; + }; + + let Some(track_entry) = self.tracks_in.iter_mut().find(|t| t.id.mid == mid) else { + return; + }; + + if track_entry + .last_keyframe_request + .map(|t| t.elapsed() < Duration::from_secs(1)) + .unwrap_or(false) + { + return; + } + + _ = writer.request_keyframe(rid, kind); + + track_entry.last_keyframe_request = Some(Instant::now()); + } + + fn handle_incoming_keyframe_req(&self, mut req: KeyframeRequest) -> Propagated { + let Some(track_out) = self.tracks_out.iter().find(|t| t.mid() == Some(req.mid)) else { + return Propagated::Noop; + }; + let Some(track_in) = track_out.track_in.upgrade() else { + return Propagated::Noop; + }; + + req.rid = self.chosen_rid; + + Propagated::KeyframeRequest(self.id, req, track_in.origin, track_in.mid) + } + + fn negotiate_if_needed(&mut self) -> bool { + if self.cid.is_none() || self.pending.is_some() { + // No data channel yet, or a negotiation is already in flight. + return false; + } + + let mut change = self.rtc.sdp_api(); + + for track in &mut self.tracks_out { + if let TrackOutState::ToOpen = track.state { + if let Some(track_in) = track.track_in.upgrade() { + let stream_id = track_in.origin.to_string(); + let mid = + change.add_media(track_in.kind, Direction::SendOnly, Some(stream_id), None); + track.state = TrackOutState::Negotiating(mid); + } + } + } + + if !change.has_changes() { + return false; + } + + let Some((offer, pending)) = change.apply() else { + return false; + }; + + let Some(mut channel) = self.cid.and_then(|id| self.rtc.channel(id)) else { + return false; + }; + + let json = serde_json::to_string(&offer).expect("offer to serialize"); + if channel.write(false, json.as_bytes()).is_err() { + return false; + } + + self.pending = Some(pending); + + true + } + + fn handle_channel_data(&mut self, d: ChannelData) -> Propagated { + if let Ok(offer) = serde_json::from_slice::<'_, SdpOffer>(&d.data) { + self.handle_offer(offer); + } else if let Ok(answer) = serde_json::from_slice::<'_, SdpAnswer>(&d.data) { + self.handle_answer(answer); + } + + Propagated::Noop + } + + fn handle_offer(&mut self, offer: SdpOffer) { + let Ok(answer) = self.rtc.sdp_api().accept_offer(offer) else { + warn!("client {}/{} offer rejected", self.room, self.peer); + return; + }; + + // Cancel any pending negotiation; it is redone after this offer. + for track in &mut self.tracks_out { + if let TrackOutState::Negotiating(_) = track.state { + track.state = TrackOutState::ToOpen; + } + } + + let Some(mut channel) = self.cid.and_then(|id| self.rtc.channel(id)) else { + return; + }; + + let json = serde_json::to_string(&answer).expect("answer to serialize"); + let _ = channel.write(false, json.as_bytes()); + } + + fn handle_answer(&mut self, answer: SdpAnswer) { + if let Some(pending) = self.pending.take() { + if self.rtc.sdp_api().accept_answer(pending, answer).is_err() { + warn!("client {}/{} answer rejected", self.room, self.peer); + return; + } + + for track in &mut self.tracks_out { + if let TrackOutState::Negotiating(m) = track.state { + track.state = TrackOutState::Open(m); + } + } + } + } + + fn handle_track_open(&mut self, track_in: Weak) { + self.tracks_out.push(TrackOut { + track_in, + state: TrackOutState::ToOpen, + }); + } + + fn handle_media_data_out(&mut self, origin: ClientId, data: &MediaData) { + // Map the incoming media to the outgoing track towards this client. + let Some(mid) = self + .tracks_out + .iter() + .find(|o| { + o.track_in + .upgrade() + .filter(|i| i.origin == origin && i.mid == data.mid) + .is_some() + }) + .and_then(|o| o.mid()) + else { + return; + }; + + if data.rid.is_some() && data.rid != Some("h".into()) { + // Simulcast layer selection (video); pass rid=None or the "h" layer. + return; + } + + if self.chosen_rid != data.rid { + self.chosen_rid = data.rid; + } + + let Some(writer) = self.rtc.writer(mid) else { + return; + }; + + let Some(pt) = writer.match_params(data.params) else { + return; + }; + + if let Err(e) = writer.write(pt, data.network_time, data.time, data.data.clone()) { + warn!("client {}/{} write failed: {:?}", self.room, self.peer, e); + self.rtc.disconnect(); + } + } + + fn handle_keyframe_request(&mut self, req: KeyframeRequest, mid_in: Mid) { + let has_incoming_track = self.tracks_in.iter().any(|i| i.id.mid == mid_in); + + if !has_incoming_track { + return; + } + + let Some(mut writer) = self.rtc.writer(mid_in) else { + return; + }; + + if let Err(e) = writer.request_keyframe(req.rid, req.kind) { + debug!("request_keyframe failed: {:?}", e); + } + } +} + +/// Events propagated between clients of a room. +#[allow(clippy::large_enum_variant)] +enum Propagated { + Noop, + Timeout(Instant), + TrackOpen(ClientId, Weak), + MediaData(ClientId, MediaData), + KeyframeRequest(ClientId, KeyframeRequest, ClientId, Mid), +} + +impl Propagated { + fn client_id(&self) -> Option { + match self { + Propagated::TrackOpen(c, _) + | Propagated::MediaData(c, _) + | Propagated::KeyframeRequest(c, _, _, _) => Some(*c), + _ => None, + } + } +} diff --git a/rust/tests/control.rs b/rust/tests/control.rs new file mode 100644 index 0000000..00ba267 --- /dev/null +++ b/rust/tests/control.rs @@ -0,0 +1,96 @@ +//! End-to-end control-plane test: boots the full sidecar (UDP + control socket +//! + SFU loop) in-process and speaks the JSON-lines protocol like PHP does. + +use std::io::{BufRead, BufReader, Write}; +use std::net::IpAddr; +use std::os::unix::net::UnixStream; +use std::path::PathBuf; +use std::thread; +use std::time::{Duration, Instant}; + +use blax_webrtc_sfu::server::{self, Config}; + +fn start_sidecar(socket: PathBuf) -> thread::JoinHandle<()> { + let config = Config { + socket, + udp_ip: IpAddr::from([127, 0, 0, 1]), + udp_port: 0, + public_ip: None, + }; + thread::spawn(move || server::run(config).expect("sidecar to run")) +} + +fn connect(socket: &PathBuf) -> UnixStream { + let deadline = Instant::now() + Duration::from_secs(5); + loop { + if let Ok(stream) = UnixStream::connect(socket) { + return stream; + } + assert!(Instant::now() < deadline, "control socket never came up"); + thread::sleep(Duration::from_millis(20)); + } +} + +fn roundtrip(reader: &mut impl BufRead, writer: &mut impl Write, req: &str) -> serde_json::Value { + writeln!(writer, "{req}").unwrap(); + writer.flush().unwrap(); + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + serde_json::from_str(&line).unwrap_or_else(|e| panic!("bad reply {line:?}: {e}")) +} + +#[test] +fn control_protocol_end_to_end() { + let socket = std::env::temp_dir().join(format!("blax-sfu-test-{}.sock", std::process::id())); + let _ = std::fs::remove_file(&socket); + let sidecar = start_sidecar(socket.clone()); + + let stream = connect(&socket); + let mut reader = BufReader::new(stream.try_clone().unwrap()); + let mut writer = stream; + + // ping + let reply = roundtrip(&mut reader, &mut writer, r#"{"id":1,"cmd":"ping"}"#); + assert_eq!(reply["id"], 1); + assert_eq!(reply["ok"], true); + + // empty stats + let reply = roundtrip(&mut reader, &mut writer, r#"{"id":2,"cmd":"stats"}"#); + assert_eq!(reply["ok"], true); + assert_eq!(reply["rooms"], serde_json::json!([])); + + // add_peer with a garbage offer errors cleanly + let reply = roundtrip( + &mut reader, + &mut writer, + r#"{"id":3,"cmd":"add_peer","room":"r1","peer":"p1","offer":"not sdp"}"#, + ); + assert_eq!(reply["ok"], false); + assert!( + reply["error"].as_str().unwrap().contains("offer"), + "unexpected error: {reply}" + ); + + // unknown peer errors cleanly + let reply = roundtrip( + &mut reader, + &mut writer, + r#"{"id":4,"cmd":"mute_peer","room":"r1","peer":"ghost","muted":true}"#, + ); + assert_eq!(reply["ok"], false); + assert!(reply["error"].as_str().unwrap().contains("no such peer")); + + // malformed json → error with id 0, connection stays usable + let reply = roundtrip(&mut reader, &mut writer, r#"{"id":5,"cmd":"nope"}"#); + assert_eq!(reply["ok"], false); + + let reply = roundtrip(&mut reader, &mut writer, r#"{"id":6,"cmd":"ping"}"#); + assert_eq!(reply["ok"], true); + + // shutdown stops the loop and removes the socket file + let reply = roundtrip(&mut reader, &mut writer, r#"{"id":7,"cmd":"shutdown"}"#); + assert_eq!(reply["ok"], true); + + sidecar.join().expect("sidecar thread to finish"); + assert!(!socket.exists(), "socket file cleaned up on shutdown"); +} diff --git a/src/Console/Commands/InstallSfuBinary.php b/src/Console/Commands/InstallSfuBinary.php new file mode 100644 index 0000000..4bd9691 --- /dev/null +++ b/src/Console/Commands/InstallSfuBinary.php @@ -0,0 +1,43 @@ +option('force') && ($found = $binaries->find())) { + $this->info("SFU sidecar already available: {$found}"); + + return self::SUCCESS; + } + + $this->line("Installing blax-webrtc-sfu {$binaries->version()} for {$binaries->platform()} …"); + + try { + $path = $binaries->install((bool) $this->option('force')); + } catch (SidecarException $e) { + $this->error($e->getMessage()); + + return self::FAILURE; + } + + $this->info("Installed: {$path}"); + + return self::SUCCESS; + } +} diff --git a/src/Console/Commands/RunSfuSidecar.php b/src/Console/Commands/RunSfuSidecar.php new file mode 100644 index 0000000..9e80a9b --- /dev/null +++ b/src/Console/Commands/RunSfuSidecar.php @@ -0,0 +1,52 @@ +ensureInstalled(); + } catch (SidecarException $e) { + $this->error($e->getMessage()); + + return self::FAILURE; + } + + $args = $supervisor->args(); + $this->info("exec {$binary} ".implode(' ', $args)); + + if (\function_exists('pcntl_exec')) { + pcntl_exec($binary, $args); // only returns on failure + + $this->error('pcntl_exec failed'); + + return self::FAILURE; + } + + passthru( + escapeshellarg($binary).' '.implode(' ', array_map('escapeshellarg', $args)), + $exit, + ); + + return $exit; + } +} diff --git a/src/Contracts/MediaEngine.php b/src/Contracts/MediaEngine.php index 4c9aec8..a435d74 100644 --- a/src/Contracts/MediaEngine.php +++ b/src/Contracts/MediaEngine.php @@ -9,9 +9,9 @@ namespace Blax\WebRtc\Contracts; * * PHP (this package) owns SIGNALING + orchestration; the MediaEngine owns the * real-time media path (ICE, DTLS, SRTP, RTP, Opus — the per-20ms DSP). The - * production engine (Str0mMediaEngine) delegates that to a Rust core (str0m, - * sans-IO) exposed to PHP via ext-php-rs, so the timing-critical work never runs - * in PHP itself. See the package `rust/` directory. + * production engine (RustMediaEngine) delegates that to the bundled Rust SFU + * sidecar (str0m, sans-IO) driven over a JSON control socket, so the + * timing-critical work never runs in PHP itself. See the package `rust/` dir. * * Because the engine terminates media server-side, it is also the single place * that can: diff --git a/src/Media/NullMediaEngine.php b/src/Media/NullMediaEngine.php index 45b55b7..385294a 100644 --- a/src/Media/NullMediaEngine.php +++ b/src/Media/NullMediaEngine.php @@ -23,7 +23,7 @@ final class NullMediaEngine implements MediaEngine { throw new \RuntimeException( 'NullMediaEngine cannot terminate media. Configure webrtc.media_engine '. - 'to Str0mMediaEngine (requires the blax_webrtc Rust extension) for real calls.' + 'to RustMediaEngine (the bundled Rust SFU sidecar) for backend-hosted calls.' ); } diff --git a/src/Media/Rust/BinaryManager.php b/src/Media/Rust/BinaryManager.php new file mode 100644 index 0000000..3ff8d07 --- /dev/null +++ b/src/Media/Rust/BinaryManager.php @@ -0,0 +1,159 @@ + $config the `webrtc.sfu` config block */ + public function __construct(private readonly array $config = []) {} + + /** E.g. "linux-x86_64", "linux-aarch64", "darwin-aarch64". */ + public function platform(): string + { + $os = match (PHP_OS_FAMILY) { + 'Darwin' => 'darwin', + default => strtolower(PHP_OS_FAMILY), + }; + $arch = match ($machine = php_uname('m')) { + 'arm64' => 'aarch64', + 'amd64' => 'x86_64', + default => $machine, + }; + + return "{$os}-{$arch}"; + } + + public function version(): string + { + return (string) ($this->config['version'] ?? self::VERSION); + } + + /** First usable binary (override → dev build → installed), or null. */ + public function find(): ?string + { + $packageRoot = dirname(__DIR__, 3); + $devBuilds = ($this->config['dev_builds'] ?? true) === true; + $candidates = array_filter([ + $this->config['binary'] ?? null, + $devBuilds ? $packageRoot.'/rust/target/release/blax-webrtc-sfu' : null, + $devBuilds ? $packageRoot.'/rust/target/debug/blax-webrtc-sfu' : null, + $this->installPath(), + ]); + + foreach ($candidates as $path) { + if (is_file($path) && is_executable($path)) { + return $path; + } + } + + return null; + } + + /** Where a downloaded binary lives (versioned + per platform). */ + public function installPath(): string + { + $dir = rtrim((string) ($this->config['install_path'] ?? sys_get_temp_dir().'/blax-webrtc'), '/'); + + return "{$dir}/blax-webrtc-sfu-{$this->version()}-{$this->platform()}"; + } + + /** Find or (if allowed) download — the plug-and-play entrypoint. */ + public function ensureInstalled(): string + { + if ($found = $this->find()) { + return $found; + } + + if (($this->config['auto_install'] ?? true) !== true) { + throw new SidecarException( + 'SFU sidecar binary not found and webrtc.sfu.auto_install is disabled — '. + 'run `php artisan webrtc:install`, build rust/ locally, or set webrtc.sfu.binary.' + ); + } + + return $this->install(); + } + + public function install(bool $force = false): string + { + $target = $this->installPath(); + if (! $force && is_file($target)) { + return $target; + } + + $url = $this->url('download_url'); + if ($url === '') { + throw new SidecarException('webrtc.sfu.download_url is not configured'); + } + + $binary = $this->fetch($url); + + // A configured checksum_url is a hard requirement, not best-effort — + // set it to null explicitly to opt out of verification. + if (($checksumUrl = $this->url('checksum_url')) !== '') { + $expected = strtolower((string) strtok(trim($this->fetch($checksumUrl)), " \t")); + $actual = hash('sha256', $binary); + if (! hash_equals($expected, $actual)) { + throw new SidecarException( + "sidecar checksum mismatch for {$url}: expected {$expected}, got {$actual}" + ); + } + } + + if (! is_dir($dir = dirname($target))) { + @mkdir($dir, 0775, true); + } + + $tmp = $target.'.download.'.getmypid(); + if (@file_put_contents($tmp, $binary) === false) { + throw new SidecarException("cannot write sidecar binary to {$tmp}"); + } + @chmod($tmp, 0755); + if (! @rename($tmp, $target)) { + @unlink($tmp); + throw new SidecarException("cannot move sidecar binary into place at {$target}"); + } + + return $target; + } + + private function url(string $key): string + { + return str_replace( + ['{version}', '{platform}'], + [$this->version(), $this->platform()], + (string) ($this->config[$key] ?? ''), + ); + } + + private function fetch(string $url): string + { + $context = stream_context_create([ + 'http' => ['timeout' => (float) ($this->config['download_timeout'] ?? 60), 'follow_location' => 1], + ]); + + $data = @file_get_contents($url, false, $context); + if ($data === false || $data === '') { + throw new SidecarException("sidecar download failed: {$url}"); + } + + return $data; + } +} diff --git a/src/Media/Rust/ControlClient.php b/src/Media/Rust/ControlClient.php new file mode 100644 index 0000000..1ed1190 --- /dev/null +++ b/src/Media/Rust/ControlClient.php @@ -0,0 +1,137 @@ +> */ + private array $events = []; + + public function __construct( + private readonly string $socketPath, + private readonly float $timeout = 5.0, + ) {} + + /** + * Build a client on an already-open stream (tests use a socket pair). + * + * @param resource $stream + */ + public static function fromStream($stream): self + { + $client = new self(''); + $client->stream = $stream; + + return $client; + } + + /** + * Send one command and wait for its reply. Throws on transport errors and + * on `ok:false` replies. + * + * @param array $params + * @return array the decoded reply (includes `ok` => true) + */ + public function request(string $cmd, array $params = []): array + { + $id = $this->nextId++; + $stream = $this->stream(); + + $json = json_encode(['id' => $id, 'cmd' => $cmd] + $params, JSON_UNESCAPED_SLASHES); + if (@fwrite($stream, $json."\n") === false) { + $this->close(); + throw new SidecarException("sidecar write failed ({$cmd})"); + } + + while (true) { + $line = @fgets($stream); + if ($line === false) { + $this->close(); + throw new SidecarException("sidecar closed or timed out awaiting the {$cmd} reply"); + } + + $reply = json_decode(trim($line), true); + if (! is_array($reply)) { + continue; + } + + if (isset($reply['event'])) { + $this->events[] = $reply; + + continue; + } + + if (($reply['id'] ?? null) !== $id) { + continue; + } + + if (($reply['ok'] ?? false) !== true) { + throw new SidecarException((string) ($reply['error'] ?? "sidecar error ({$cmd})")); + } + + return $reply; + } + } + + /** + * Notices (peer_connected / peer_left / …) received while waiting for + * replies. Clears the buffer. + * + * @return list> + */ + public function drainEvents(): array + { + [$events, $this->events] = [$this->events, []]; + + return $events; + } + + public function close(): void + { + if (is_resource($this->stream)) { + @fclose($this->stream); + } + $this->stream = null; + } + + /** @return resource */ + private function stream() + { + if (is_resource($this->stream)) { + return $this->stream; + } + + $stream = @stream_socket_client( + 'unix://'.$this->socketPath, + $errno, + $error, + $this->timeout, + ); + + if ($stream === false) { + throw new SidecarException( + "cannot connect to the SFU sidecar at {$this->socketPath}: ".($error ?: "errno {$errno}") + ); + } + + $seconds = (int) floor($this->timeout); + stream_set_timeout($stream, $seconds, (int) (($this->timeout - $seconds) * 1_000_000)); + + return $this->stream = $stream; + } +} diff --git a/src/Media/Rust/SidecarException.php b/src/Media/Rust/SidecarException.php new file mode 100644 index 0000000..522a4f1 --- /dev/null +++ b/src/Media/Rust/SidecarException.php @@ -0,0 +1,11 @@ + $config the `webrtc.sfu` config block */ + public function __construct( + private readonly BinaryManager $binaries, + private readonly array $config = [], + ) {} + + public function socketPath(): string + { + return (string) ($this->config['socket'] ?? sys_get_temp_dir().'/blax-webrtc-sfu.sock'); + } + + public function running(): bool + { + if (! file_exists($this->socketPath())) { + return false; + } + + try { + (new ControlClient($this->socketPath(), 2.0))->request('ping'); + + return true; + } catch (SidecarException) { + return false; + } + } + + public function ensureRunning(): void + { + if ($this->running()) { + return; + } + + if (($this->config['auto_spawn'] ?? true) !== true) { + throw new SidecarException( + 'the SFU sidecar is not running at '.$this->socketPath(). + ' and webrtc.sfu.auto_spawn is disabled — start it with `php artisan webrtc:sidecar`.' + ); + } + + $lockPath = $this->socketPath().'.lock'; + if (! is_dir($dir = dirname($lockPath))) { + @mkdir($dir, 0775, true); + } + $lock = @fopen($lockPath, 'c'); + if ($lock === false || ! flock($lock, LOCK_EX)) { + throw new SidecarException("cannot acquire the sidecar spawn lock at {$lockPath}"); + } + + try { + if ($this->running()) { + return; // another worker won the race + } + + $this->spawnDetached(); + $this->awaitSocket(); + } finally { + flock($lock, LOCK_UN); + fclose($lock); + } + } + + /** The CLI arguments matching this supervisor's config. @return list */ + public function args(): array + { + $args = ['--socket', $this->socketPath()]; + + if (($ip = (string) ($this->config['udp_ip'] ?? '')) !== '') { + array_push($args, '--udp-ip', $ip); + } + if (($port = (int) ($this->config['udp_port'] ?? 0)) > 0) { + array_push($args, '--udp-port', (string) $port); + } + if (($ip = (string) ($this->config['public_ip'] ?? '')) !== '') { + array_push($args, '--public-ip', $ip); + } + + return $args; + } + + public function logPath(): string + { + return (string) ($this->config['log'] ?? sys_get_temp_dir().'/blax-webrtc-sfu.log'); + } + + /** Graceful stop via the control socket (no-op when not running). */ + public function stop(): void + { + if (! $this->running()) { + return; + } + + try { + (new ControlClient($this->socketPath(), 2.0))->request('shutdown'); + } catch (SidecarException) { + // it went away — that is what we asked for + } + } + + private function spawnDetached(): void + { + $binary = $this->binaries->ensureInstalled(); + + $log = $this->logPath(); + if (! is_dir($dir = dirname($log))) { + @mkdir($dir, 0775, true); + } + + // `&` + i/o redirection detaches it from this worker; the shell exits + // and the sidecar reparents to init (survives the PHP worker). + $command = sprintf( + '%s %s >> %s 2>&1 < /dev/null &', + escapeshellarg($binary), + implode(' ', array_map('escapeshellarg', $this->args())), + escapeshellarg($log), + ); + + shell_exec($command); + } + + private function awaitSocket(): void + { + $deadline = microtime(true) + (float) ($this->config['spawn_timeout'] ?? 10.0); + + while (microtime(true) < $deadline) { + if ($this->running()) { + return; + } + usleep(100_000); + } + + throw new SidecarException( + 'the SFU sidecar did not come up within the spawn timeout — check '.$this->logPath() + ); + } +} diff --git a/src/Media/RustMediaEngine.php b/src/Media/RustMediaEngine.php new file mode 100644 index 0000000..e023945 --- /dev/null +++ b/src/Media/RustMediaEngine.php @@ -0,0 +1,161 @@ +booted = $client !== null; + } + + public function name(): string + { + return 'rust-sfu'; + } + + // ── Room-aware primary API ───────────────────────────────────────────── + + /** + * Add a peer to a hosted room: accepts the browser's SDP offer, returns + * the answer. Pass $recordPath to record the peer from the first packet + * (Ogg/Opus). Rejoining with the same ids replaces the old connection. + */ + public function addPeer(string $room, string $peer, string $sdpOffer, ?string $recordPath = null): string + { + $params = ['room' => $room, 'peer' => $peer, 'offer' => $sdpOffer]; + if ($recordPath !== null) { + $params['record'] = $recordPath; + } + + return (string) $this->client()->request('add_peer', $params)['answer']; + } + + public function removePeer(string $room, string $peer): void + { + $this->client()->request('remove_peer', ['room' => $room, 'peer' => $peer]); + } + + /** Server-side mute: the room stops hearing the peer (recording pauses too). */ + public function mutePeer(string $room, string $peer, bool $muted): void + { + $this->client()->request('mute_peer', ['room' => $room, 'peer' => $peer, 'muted' => $muted]); + } + + /** + * Rooms/peers snapshot: connection state, mute, connected seconds (usage). + * + * @return list}> + */ + public function stats(): array + { + return (array) ($this->client()->request('stats')['rooms'] ?? []); + } + + /** + * Buffered sidecar notices (peer_connected / peer_left with usage seconds). + * + * @return list> + */ + public function drainEvents(): array + { + return $this->client()->drainEvents(); + } + + // ── MediaEngine contract (peer id = "room/peer") ─────────────────────── + + public function offer(string $peerId, string $sdpOffer): string + { + [$room, $peer] = $this->split($peerId); + + return $this->addPeer($room, $peer, $sdpOffer); + } + + public function addIceCandidate(string $peerId, array $candidate): void + { + // Nothing to forward: the answer carries the sidecar's host candidate, + // and str0m (full ICE) learns the browser's candidates from the STUN + // binding requests arriving on the media socket. + } + + public function startRecording(string $peerId, string $path): void + { + [$room, $peer] = $this->split($peerId); + $this->client()->request('record_start', ['room' => $room, 'peer' => $peer, 'path' => $path]); + } + + public function stopRecording(string $peerId): void + { + [$room, $peer] = $this->split($peerId); + $this->client()->request('record_stop', ['room' => $room, 'peer' => $peer]); + } + + public function connectPeers(string $peerA, string $peerB): void + { + // Nothing to do: the SFU forwards between all peers of a room + // automatically — room membership IS the routing table. + } + + public function bridge(string $peerId, array $options): void + { + throw new SidecarException( + 'AI bridging inside the Rust data plane is not implemented yet — '. + 'bridge via Realtime\OpenAiRealtimeBridge for now (duplex in-SFU bridge is #1051).' + ); + } + + public function close(string $peerId): void + { + [$room, $peer] = $this->split($peerId); + $this->removePeer($room, $peer); + } + + // ─────────────────────────────────────────────────────────────────────── + + /** @return array{0:string,1:string} */ + private function split(string $peerId): array + { + $pos = strpos($peerId, '/'); + if ($pos === false) { + return ['default', $peerId]; + } + + return [substr($peerId, 0, $pos), substr($peerId, $pos + 1)]; + } + + private function client(): ControlClient + { + if (! $this->booted) { + $this->sidecar->ensureRunning(); + $this->booted = true; + } + + return $this->client ??= new ControlClient($this->sidecar->socketPath()); + } +} diff --git a/src/Media/Str0mMediaEngine.php b/src/Media/Str0mMediaEngine.php deleted file mode 100644 index 96fa84f..0000000 --- a/src/Media/Str0mMediaEngine.php +++ /dev/null @@ -1,82 +0,0 @@ -app->singleton(BinaryManager::class, function () { + return new BinaryManager((array) config('webrtc.sfu', [])); + }); + $this->app->singleton(SidecarSupervisor::class, function ($app) { + return new SidecarSupervisor($app->make(BinaryManager::class), (array) config('webrtc.sfu', [])); + }); + // Realtime AI bridge (provider hidden from the browser, model swappable server-side). $this->app->bind(OpenAiRealtimeBridge::class, function () { $endpoint = (string) config('webrtc.bridge.openai.endpoint', 'wss://api.openai.com/v1/realtime'); @@ -77,6 +90,12 @@ class WebRtcServiceProvider extends ServiceProvider __DIR__.'/../config/webrtc.php' => $this->app->configPath('webrtc.php'), ], 'webrtc-config'); + // The Rust SFU sidecar commands have no extra PHP dependencies. + $this->commands([ + InstallSfuBinary::class, + RunSfuSidecar::class, + ]); + // The bundled ReactPHP relay server needs blax-software/laravel-ws + // reactphp-kernel (dev-only deps). A host that only consumes the domain // layer (rooms/recording/realtime bridge) on its OWN transport won't have diff --git a/tests/BinaryManagerTest.php b/tests/BinaryManagerTest.php new file mode 100644 index 0000000..7397195 --- /dev/null +++ b/tests/BinaryManagerTest.php @@ -0,0 +1,117 @@ +dir = sys_get_temp_dir().'/blax-webrtc-binmgr-'.getmypid(); + @mkdir($this->dir, 0775, true); + } + + protected function tearDown(): void + { + foreach (glob($this->dir.'/*') ?: [] as $file) { + @unlink($file); + } + @rmdir($this->dir); + } + + public function test_platform_is_os_dash_arch(): void + { + $platform = (new BinaryManager)->platform(); + + $this->assertMatchesRegularExpression('/^(linux|darwin|windows|bsd|solaris)-[a-z0-9_]+$/', $platform); + $this->assertStringNotContainsString('amd64', $platform, 'arch aliases are normalized'); + $this->assertStringNotContainsString('arm64', $platform); + } + + public function test_an_explicit_binary_override_wins(): void + { + $binary = $this->dir.'/custom-sfu'; + file_put_contents($binary, '#!/bin/sh'); + chmod($binary, 0755); + + $manager = new BinaryManager(['binary' => $binary]); + + $this->assertSame($binary, $manager->find()); + } + + public function test_install_downloads_verifies_sha256_and_marks_executable(): void + { + $payload = 'fake-sfu-binary-payload'; + file_put_contents($this->dir.'/source', $payload); + // sha256sum format: " " + file_put_contents($this->dir.'/source.sha256', hash('sha256', $payload)." blax-webrtc-sfu\n"); + + $manager = new BinaryManager([ + 'dev_builds' => false, + 'install_path' => $this->dir, + 'download_url' => 'file://'.$this->dir.'/source', + 'checksum_url' => 'file://'.$this->dir.'/source.sha256', + ]); + + $installed = $manager->ensureInstalled(); + + $this->assertSame($manager->installPath(), $installed); + $this->assertSame($payload, file_get_contents($installed)); + $this->assertTrue(is_executable($installed)); + + // Second call finds it without downloading (source removed to prove it). + unlink($this->dir.'/source'); + $this->assertSame($installed, $manager->ensureInstalled()); + } + + public function test_a_checksum_mismatch_aborts_the_install(): void + { + file_put_contents($this->dir.'/source', 'tampered payload'); + file_put_contents($this->dir.'/source.sha256', hash('sha256', 'the real payload')."\n"); + + $manager = new BinaryManager([ + 'dev_builds' => false, + 'install_path' => $this->dir, + 'download_url' => 'file://'.$this->dir.'/source', + 'checksum_url' => 'file://'.$this->dir.'/source.sha256', + ]); + + $this->expectException(SidecarException::class); + $this->expectExceptionMessage('checksum mismatch'); + + $manager->install(); + + $this->assertFileDoesNotExist($manager->installPath()); + } + + public function test_auto_install_false_fails_with_actionable_guidance(): void + { + $manager = new BinaryManager([ + 'dev_builds' => false, + 'install_path' => $this->dir, + 'auto_install' => false, + ]); + + $this->expectException(SidecarException::class); + $this->expectExceptionMessage('webrtc:install'); + + $manager->ensureInstalled(); + } + + public function test_install_path_is_versioned_and_per_platform(): void + { + $manager = new BinaryManager(['install_path' => '/opt/bin', 'version' => 'v9.9.9']); + + $this->assertSame( + '/opt/bin/blax-webrtc-sfu-v9.9.9-'.$manager->platform(), + $manager->installPath(), + ); + } +} diff --git a/tests/ControlClientTest.php b/tests/ControlClientTest.php new file mode 100644 index 0000000..9108119 --- /dev/null +++ b/tests/ControlClientTest.php @@ -0,0 +1,97 @@ +sidecar = $theirs; + $this->client = ControlClient::fromStream($ours); + } + + private function sidecarSays(array ...$lines): void + { + foreach ($lines as $line) { + fwrite($this->sidecar, json_encode($line)."\n"); + } + } + + private function sidecarHeard(): array + { + return json_decode((string) fgets($this->sidecar), true); + } + + public function test_request_frames_json_lines_and_matches_the_reply_id(): void + { + $this->sidecarSays(['id' => 1, 'ok' => true, 'answer' => 'v=answer']); + + $reply = $this->client->request('add_peer', ['room' => 'r1', 'peer' => 'p1', 'offer' => 'v=0']); + + $this->assertSame('v=answer', $reply['answer']); + $this->assertSame( + ['id' => 1, 'cmd' => 'add_peer', 'room' => 'r1', 'peer' => 'p1', 'offer' => 'v=0'], + $this->sidecarHeard(), + ); + } + + public function test_pushed_events_are_buffered_not_mistaken_for_replies(): void + { + $this->sidecarSays( + ['event' => 'peer_left', 'room' => 'r1', 'peer' => 'p9', 'seconds' => 12.5], + ['id' => 1, 'ok' => true], + ); + + $this->client->request('ping'); + + $events = $this->client->drainEvents(); + $this->assertCount(1, $events); + $this->assertSame('peer_left', $events[0]['event']); + $this->assertSame([], $this->client->drainEvents(), 'drain clears the buffer'); + } + + public function test_ok_false_replies_throw_with_the_sidecar_error(): void + { + $this->sidecarSays(['id' => 1, 'ok' => false, 'error' => 'no such peer: r1/ghost']); + + $this->expectException(SidecarException::class); + $this->expectExceptionMessage('no such peer: r1/ghost'); + + $this->client->request('mute_peer', ['room' => 'r1', 'peer' => 'ghost', 'muted' => true]); + } + + public function test_a_closed_sidecar_throws_instead_of_hanging(): void + { + // Half-close: the sidecar stops sending (EOF on our reads) but still + // accepts our write — exercising the read-path failure. + stream_socket_shutdown($this->sidecar, STREAM_SHUT_WR); + + $this->expectException(SidecarException::class); + $this->expectExceptionMessage('closed or timed out'); + + $this->client->request('ping'); + } + + public function test_ids_increment_across_requests(): void + { + $this->sidecarSays(['id' => 1, 'ok' => true], ['id' => 2, 'ok' => true]); + + $this->client->request('ping'); + $this->client->request('ping'); + + $this->assertSame(1, $this->sidecarHeard()['id']); + $this->assertSame(2, $this->sidecarHeard()['id']); + } +} diff --git a/tests/RustMediaEngineTest.php b/tests/RustMediaEngineTest.php new file mode 100644 index 0000000..4c02c7b --- /dev/null +++ b/tests/RustMediaEngineTest.php @@ -0,0 +1,130 @@ +sidecar = $theirs; + + $supervisor = new SidecarSupervisor(new BinaryManager([]), ['auto_spawn' => false]); + $this->engine = new RustMediaEngine($supervisor, ControlClient::fromStream($ours)); + } + + private function sidecarSays(array $line): void + { + fwrite($this->sidecar, json_encode($line)."\n"); + } + + private function sidecarHeard(): array + { + return json_decode((string) fgets($this->sidecar), true); + } + + public function test_add_peer_speaks_the_control_protocol(): void + { + $this->sidecarSays(['id' => 1, 'ok' => true, 'answer' => 'v=answer']); + + $answer = $this->engine->addPeer('lobby', 'alice', 'v=offer', '/rec/alice.ogg'); + + $this->assertSame('v=answer', $answer); + $this->assertSame( + [ + 'id' => 1, + 'cmd' => 'add_peer', + 'room' => 'lobby', + 'peer' => 'alice', + 'offer' => 'v=offer', + 'record' => '/rec/alice.ogg', + ], + $this->sidecarHeard(), + ); + } + + public function test_media_engine_contract_splits_composite_peer_ids(): void + { + $this->sidecarSays(['id' => 1, 'ok' => true, 'answer' => 'v=a']); + $this->engine->offer('lobby/alice', 'v=offer'); + $heard = $this->sidecarHeard(); + $this->assertSame(['lobby', 'alice'], [$heard['room'], $heard['peer']]); + $this->assertArrayNotHasKey('record', $heard); + + $this->sidecarSays(['id' => 2, 'ok' => true]); + $this->engine->startRecording('lobby/alice', '/rec/alice.ogg'); + $heard = $this->sidecarHeard(); + $this->assertSame('record_start', $heard['cmd']); + $this->assertSame('/rec/alice.ogg', $heard['path']); + + $this->sidecarSays(['id' => 3, 'ok' => true]); + $this->engine->close('lobby/alice'); + $this->assertSame('remove_peer', $this->sidecarHeard()['cmd']); + } + + public function test_bare_peer_ids_fall_back_to_the_default_room(): void + { + $this->sidecarSays(['id' => 1, 'ok' => true, 'answer' => 'v=a']); + $this->engine->offer('alice', 'v=offer'); + $heard = $this->sidecarHeard(); + $this->assertSame(['default', 'alice'], [$heard['room'], $heard['peer']]); + } + + public function test_mute_stats_and_events(): void + { + $this->sidecarSays(['id' => 1, 'ok' => true]); + $this->engine->mutePeer('lobby', 'alice', true); + $this->assertTrue($this->sidecarHeard()['muted']); + + $this->sidecarSays([ + 'id' => 2, + 'ok' => true, + 'rooms' => [['room' => 'lobby', 'peers' => [['peer' => 'alice', 'connected' => true, 'muted' => true, 'seconds' => 4.2]]]], + ]); + $stats = $this->engine->stats(); + $this->assertSame('lobby', $stats[0]['room']); + + fwrite($this->sidecar, json_encode(['event' => 'peer_connected', 'room' => 'lobby', 'peer' => 'bob'])."\n"); + $this->sidecarSays(['id' => 3, 'ok' => true]); + $this->engine->removePeer('lobby', 'bob'); + $this->assertSame('peer_connected', $this->engine->drainEvents()[0]['event']); + } + + public function test_ice_candidates_and_connect_peers_are_deliberate_noops(): void + { + // Neither may touch the socket: str0m learns remote candidates from + // STUN, and room membership already routes media. + $this->engine->addIceCandidate('lobby/alice', ['candidate' => 'x']); + $this->engine->connectPeers('lobby/alice', 'lobby/bob'); + + stream_set_blocking($this->sidecar, false); + $this->assertFalse(fgets($this->sidecar), 'no bytes were written to the sidecar'); + } + + public function test_bridge_is_not_implemented_in_the_data_plane_yet(): void + { + $this->expectException(SidecarException::class); + $this->expectExceptionMessage('OpenAiRealtimeBridge'); + + $this->engine->bridge('lobby/alice', ['model' => 'gpt-realtime']); + } + + public function test_engine_name(): void + { + $this->assertSame('rust-sfu', $this->engine->name()); + } +} diff --git a/tests/SidecarSupervisorTest.php b/tests/SidecarSupervisorTest.php new file mode 100644 index 0000000..ae401dd --- /dev/null +++ b/tests/SidecarSupervisorTest.php @@ -0,0 +1,107 @@ + sys_get_temp_dir().'/blax-sfu-never-'.getmypid().'.sock', + 'auto_spawn' => false, + ]); + + $this->assertFalse($supervisor->running()); + + $this->expectException(SidecarException::class); + $this->expectExceptionMessage('webrtc:sidecar'); + + $supervisor->ensureRunning(); + } + + public function test_args_mirror_the_config(): void + { + $supervisor = new SidecarSupervisor(new BinaryManager([]), [ + 'socket' => '/run/sfu.sock', + 'udp_ip' => '10.0.0.5', + 'udp_port' => 41000, + 'public_ip' => '203.0.113.9', + ]); + + $this->assertSame( + ['--socket', '/run/sfu.sock', '--udp-ip', '10.0.0.5', '--udp-port', '41000', '--public-ip', '203.0.113.9'], + $supervisor->args(), + ); + } + + /** + * REAL cross-language round-trip: plug-and-play spawn of the actual Rust + * sidecar (local cargo build), the full control protocol through + * RustMediaEngine, and a graceful stop. Skipped until rust/ is built. + */ + public function test_spawns_and_drives_the_real_sidecar(): void + { + $binary = \dirname(__DIR__).'/rust/target/debug/blax-webrtc-sfu'; + if (! is_file($binary)) { + $this->markTestSkipped('rust sidecar not built — run `cargo build` in rust/'); + } + + $socket = sys_get_temp_dir().'/blax-sfu-e2e-'.getmypid().'.sock'; + $config = [ + 'socket' => $socket, + 'binary' => $binary, + 'auto_install' => false, + 'udp_ip' => '127.0.0.1', + 'log' => sys_get_temp_dir().'/blax-sfu-e2e-'.getmypid().'.log', + 'spawn_timeout' => 15.0, + ]; + $supervisor = new SidecarSupervisor(new BinaryManager($config), $config); + + try { + $this->assertFalse($supervisor->running()); + + $supervisor->ensureRunning(); // finds the binary + spawns detached + $this->assertTrue($supervisor->running()); + + $engine = new RustMediaEngine($supervisor); + + // Empty stats on a fresh sidecar. + $this->assertSame([], $engine->stats()); + + // A garbage offer must come back as a clean protocol error. + try { + $engine->addPeer('lobby', 'alice', 'this is not sdp'); + $this->fail('expected the sidecar to reject a garbage offer'); + } catch (SidecarException $e) { + $this->assertStringContainsString('offer', $e->getMessage()); + } + + // Unknown peers error cleanly too. + try { + $engine->mutePeer('lobby', 'ghost', true); + $this->fail('expected a no-such-peer error'); + } catch (SidecarException $e) { + $this->assertStringContainsString('no such peer', $e->getMessage()); + } + + // ensureRunning is idempotent while it lives. + $supervisor->ensureRunning(); + $this->assertTrue($supervisor->running()); + } finally { + $supervisor->stop(); + @unlink($socket); + @unlink($socket.'.lock'); + @unlink($config['log']); + } + + $this->assertFalse($supervisor->running(), 'graceful shutdown via the control socket'); + } +}