Author SHA1 Message Date
shadowdaoandClaude Sonnet 5 3f2933ae3f fix(session): recover stalled video subscriptions with a keyframe-forcing watchdog
Build / macOS (macos-latest) (push) Successful in 1m7s
Build / Linux (ubuntu-24.04) (push) Successful in 1m14s
Build / macOS (macos-latest) (pull_request) Successful in 53s
Build / Linux (ubuntu-24.04) (pull_request) Successful in 1m19s
Build / Windows (windows-latest) (push) Failing after 3m25s
Build / Windows (windows-latest) (pull_request) Failing after 3m0s
Camera sources in the OBS plugin were dropping out at random and never
recovering, while the same players stayed healthy in browser talkback.
Measured on the live server: every OBS plugin subscriber racked up
862-1099 nackMisses (retransmit requests for packets the SFU had already
aged out of its send buffer -- unrecoverable loss) and sat at plis == 2
for a multi-hour session, versus 0/adaptive-PLI for a browser subscriber
in the same room. The pinned client-sdk-cpp (1.10.1) exposes no
PLI/keyframe-request API at all, so a decoder that lost a frame that way
had no way to resync for the rest of the show.

Add a stall-recovery watchdog: StallWatchdog (session_types.h/.cpp) is a
pure, fake-clock-testable class that decides when a video subscription
has gone too long (2000ms, kStallRecoveryTimeout) without a decoded
frame reaching OBS. LiveKitSession::Impl polls it from a dedicated
thread and, through the existing command queue (never touching the SDK
off the worker thread), toggles the publication's setEnabled(false)/
setEnabled(true) -- the one lever this SDK exposes that makes the SFU
restart delivery, and a restart always begins with a keyframe.

Repeated attempts against the same stall back off exponentially
(2s/4s/8s/16s/30s-capped, mirroring the shape of the adapter's own
reconnect backoff) so a genuinely gone publisher is retried on a bounded
cadence instead of hammered every 2 seconds. A muted, disabled or
unsubscribed track, an audio-only source, or a disconnected session
never arms the watchdog: new onTrackMuted/onTrackUnmuted handlers
suspend and resume its clock, always re-baselining from "now" rather
than a stale timestamp, so un-muting after a long legitimate camera-off
period cannot read as a multi-minute stall. Each attempt, and eventual
recovery, is logged through a new DiagnosticHandler that the OBS adapter
maps onto obs_log at the same severities the file already uses for other
notable events.

Also investigated (not changed): the setVideoQuality(HIGH) pin added for
an earlier simulcast-resize bug. The SDK's own docs and the FFI binary's
wire-protocol strings show setVideoQuality only bounds spatial/simulcast
layer selection (UpdateTrackSettings.quality), never temporal layers or
frame rate, which the SFU's own congestion control governs independently
-- so this pin is unlikely to be the cause of the packet-loss symptom,
and is probably an inert no-op now that publishing is pinned server-side
to a single spatial layer (L1T3). Left in place since that is not fully
unambiguous from the SDK alone. Full writeup in
.stall-recovery-report.md (untracked, not part of this commit).

Adds 3 new headless StallWatchdog tests (fires-after-threshold, does-
not-fire-when-muted/disabled, backs-off-rather-than-loops) to
test_session.cpp; caught a real bug during development where
onFrameDelivered() reset the backoff duration but not the
next-attempt-allowed timestamp, throttling a just-recovered stream
against its own stale backoff. Built and all 6 CTest suites pass
(124/124 checks in test_session); also verified clean under
ThreadSanitizer with warning counts unchanged from the pre-change
baseline.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-20 19:04:14 -07:00
jknapp 48a74e8c67 Merge pull request 'docs: the macOS bundle was fine all along, and record the GUI verification' (#6) from docs/macos-packaging-and-gui-status into main 2026-09-10 12:31:33 +00:00
shadowdaoandClaude Opus 5 ab455a9cfe docs: the macOS bundle was fine all along, and record the GUI verification
Two corrections and one promotion, all from evidence rather than inference.

1. The "macOS packaging gap (known, unfixed)" section was WRONG, and it
   contradicted the release notes for the same build. It claimed the artifact
   is "a bare streamer-tools-camera.so" with a relative libobs install name
   that "will not load in OBS.app as it stands". Downloading and inspecting
   the shipped streamer-tools-camera-v0.1.0-macos.zip shows otherwise:

   - a proper streamer-tools-camera.plugin bundle -- Contents/MacOS/<name> is
     Mach-O MH_BUNDLE (what OBS loads), with Info.plist (BNDL, correct
     CFBundleExecutable), Contents/Resources/locale/en-US.ini, and both
     LiveKit dylibs in Contents/Frameworks/
   - install names are right: the module loads
     @rpath/libobs.framework/Versions/A/libobs and carries
     LC_RPATH @executable_path/../Frameworks, which inside OBS.app resolves to
     OBS.app/Contents/Frameworks; @rpath/liblivekit.dylib resolves through
     LC_RPATH @loader_path/../Frameworks to the bundle's own copy, and
     liblivekit.dylib finds liblivekit_ffi.dylib through LC_RPATH @loader_path.
     Nothing points into a build tree.
   - all three binaries carry LC_CODE_SIGNATURE, which is not optional:
     arm64 macOS refuses to load unsigned code.

   The real macOS limitation is different and now stated: the bundle is
   arm64-only (no x86_64 slice), macOS 13+. Nobody has still ever opened it in
   OBS.app -- well formed and signed is a prior, not a load.

2. Linux and Windows are confirmed working in the OBS GUI: video and audio
   both arrive and hold up across a session, Linux by the project owner and
   Windows by two directors independently, and the plugin carried a live show
   on 2026-09-07. Since listing cameras requires an API call, that also
   retires "the WinHTTP backend has never run against a real streamer-tools
   server".

   Scoped, not inflated: MEASURED A/V sync and latency against the egress path
   are still unverified (no drift reported is not a measurement), as is
   mid-show publisher restart. The "Not verified anywhere" list is now
   deduplicated and says exactly that.

3. The CI table's Windows row still said "Failing, fix pushed and awaiting a
   completed run". It is green, after the WinHTTP deadline fix.

Release notes template updated to match, and v0.1.0's published notes have
been regenerated through it so the public page stops repeating the bare-.so
claim.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AzGnvQ6wfD7bw7PZN35ft9
2026-09-10 05:31:09 -07:00
jknapp e763d61ed3 Merge pull request 'fix(http): bound the WinHTTP response-header wait, and stop docs triggering builds' (#5) from fix/winhttp-response-header-timeout into main
Build / macOS (macos-latest) (push) Successful in 49s
Build / Linux (ubuntu-24.04) (push) Successful in 1m49s
Build / Windows (windows-latest) (push) Successful in 4m5s
2026-09-10 01:54:16 +00:00
8 changed files with 648 additions and 72 deletions
+14 -10
View File
@@ -25,7 +25,7 @@ set -euo pipefail
MACOS_BUNDLE_FOUND="${MACOS_BUNDLE_FOUND:-false}" MACOS_BUNDLE_FOUND="${MACOS_BUNDLE_FOUND:-false}"
if [ "${MACOS_BUNDLE_FOUND}" = "true" ]; then if [ "${MACOS_BUNDLE_FOUND}" = "true" ]; then
MACOS_NOTE="This archive contains a \`.plugin\` bundle." MACOS_NOTE="This archive contains a \`.plugin\` bundle (verified on v0.1.0: MH_BUNDLE + Info.plist, libobs via \`@rpath\` + \`@executable_path/../Frameworks\`, LiveKit dylibs bundled, all three binaries code-signed). **arm64 only -- no Intel slice**, macOS 13+. Never yet loaded in OBS.app by a human."
else else
MACOS_NOTE="This archive is packaged as a bare \`streamer-tools-camera.so\` (the layout \`build/package/\` currently produces on macOS), **not** an OBS.app-loadable \`.plugin\` bundle. It will not load in the OBS GUI as-is -- see the \"macOS packaging gap\" section of \`README.md\`." MACOS_NOTE="This archive is packaged as a bare \`streamer-tools-camera.so\` (the layout \`build/package/\` currently produces on macOS), **not** an OBS.app-loadable \`.plugin\` bundle. It will not load in the OBS GUI as-is -- see the \"macOS packaging gap\" section of \`README.md\`."
fi fi
@@ -36,18 +36,22 @@ cat > "${NOTES_FILE}" <<EOF
Built from commit \`${SHA}\`. Built from commit \`${SHA}\`.
**Read the per-platform notes below before relying on this.** The module has **Confirmed working on Linux and Windows, including a live show.** The plugin
been loaded in the OBS GUI exactly once -- Windows, OBS 32.2.2 on Windows 11, carried a real broadcast on 2026-09-07. Video and audio both arrive and hold
2026-09-09 -- and only "it loads and registers its source type" is confirmed up across a session: Linux verified by the project owner, Windows by two
there. Whether video renders correctly, A/V sync, end-to-end latency and directors independently.
mid-show publisher restart are all still unverified on every platform. See
"What is verified, and how" in \`README.md\` for exactly what is backed by Still unverified: **macOS in the OBS GUI** (nobody has opened it -- see the
automated tests versus a human watching OBS. table), **measured** A/V sync and end-to-end latency against the existing
egress path (no drift reported, but nothing measured), and whether a publisher
restarting mid-show recovers cleanly on screen. See "What is verified, and
how" in \`README.md\` for what is backed by automated tests versus a human
watching OBS.
| Platform | Archive | Notes | | Platform | Archive | Notes |
|---|---|---| |---|---|---|
| Linux (x64) | \`streamer-tools-camera-${TAG}-linux-x64.zip\` | Functionally complete and verified end to end against a real LiveKit server and a real libobs (see README); OBS GUI itself still unverified | | Linux (x64) | \`streamer-tools-camera-${TAG}-linux-x64.zip\` | Functionally complete, verified end to end against a real LiveKit server and a real libobs (see README), and confirmed working in the OBS GUI |
| Windows (x64) | \`streamer-tools-camera-${TAG}-windows-x64.zip\` | Built and tested by this workflow's Windows job. Confirmed to load in the OBS GUI (32.2.2 / Windows 11, 2026-09-09); nothing past module load is verified. The WinHTTP backend has never been exercised against a real streamer-tools server, only a loopback test server -- see README's Windows CI section | | Windows (x64) | \`streamer-tools-camera-${TAG}-windows-x64.zip\` | Built and tested by this workflow's Windows job, and confirmed working in the OBS GUI by two directors independently (first load: OBS 32.2.2 / Windows 11) -- which also exercises the WinHTTP backend against a real streamer-tools server |
| macOS | \`streamer-tools-camera-${TAG}-macos.zip\` | Built and tested by this workflow's macOS job. ${MACOS_NOTE} | | macOS | \`streamer-tools-camera-${TAG}-macos.zip\` | Built and tested by this workflow's macOS job. ${MACOS_NOTE} |
## Installing ## Installing
+61 -49
View File
@@ -24,15 +24,21 @@ The plugin is **functionally complete on Linux and verified end to end there**
real streamer-tools API shape, and pushes decoded frames into real streamer-tools API shape, and pushes decoded frames into
`obs_source_output_video`/`_audio`). `obs_source_output_video`/`_audio`).
**First confirmed OBS GUI load: Windows, 2026-09-09** — the v0.1.0 release **Confirmed working in the OBS GUI on Linux and Windows, including a live
artifact loaded into OBS 32.2.2 on Windows 11 (build 26200) on a director's show.** The plugin carried a real broadcast on 2026-09-07 and was reported to
machine, from work well. Video and audio both arrive and hold up across a session: Linux
`C:\ProgramData\obs-studio\plugins\streamer-tools-camera\bin\64bit\`. verified by the project owner, Windows by two directors independently
That retires "the module will not even load in a real OBS" for Windows. It (2026-09-09/10; the first Windows load was OBS 32.2.2 on Windows 11 build
does **not** yet cover whether video renders correctly, colours, A/V sync or 26200, from
latency — see "Not verified anywhere" below for what is still open. Linux and `C:\ProgramData\obs-studio\plugins\streamer-tools-camera\bin\64bit\`).
macOS have still never been opened in the GUI; macOS builds the real module in Because listing cameras requires an API call, that also retires "the WinHTTP
CI but its artifact is not yet loadable (see the macOS packaging gap under CI). backend has never run against a real streamer-tools server".
What that does **not** cover: measured A/V sync and end-to-end latency against
the existing egress path (no drift reported over a session, but nothing was
measured), mid-show publisher restart, and **macOS in the GUI — still never
opened by anyone**, though its artifact is now known to be correctly packaged
(see macOS packaging below). See "Not verified anywhere" for the current list.
⚠️ **The install directory is not the same on every platform, and getting it ⚠️ **The install directory is not the same on every platform, and getting it
wrong fails silently.** On Windows it is wrong fails silently.** On Windows it is
@@ -169,15 +175,15 @@ plugins. This bit a director on 2026-09-09: a correctly-shaped install under
`AppData\Roaming` produced a log with zero mention of the module. `AppData\Roaming` produced a log with zero mention of the module.
The module resolves the LiveKit libraries from `$ORIGIN` (verified: `ldd` on the staged copy resolves both The module resolves the LiveKit libraries from `$ORIGIN` (verified: `ldd` on the staged copy resolves both
to `bin/64bit/`), not from the build tree. macOS is not this shape; see the to `bin/64bit/`), not from the build tree. macOS is not this shape — it ships
macOS packaging gap under CI. a `.plugin` bundle; see macOS packaging under CI.
## Testing this by hand ## Testing this by hand
**The module has been loaded in the OBS GUI on Windows once (2026-09-09, OBS **Linux and Windows are confirmed working in the GUI — video and audio over a
32.2.2 / Windows 11 26200) — nothing beyond "it loads and registers its source" real session, Linux by the project owner and Windows by two directors
is confirmed there, and Linux and macOS have never been opened in the GUI at independently (2026-09-09/10). macOS has never been opened in the GUI by
all.** To do it on Linux: anyone.** To repeat the Linux run:
``` ```
mkdir -p ~/.config/obs-studio/plugins/streamer-tools-camera mkdir -p ~/.config/obs-studio/plugins/streamer-tools-camera
@@ -233,21 +239,17 @@ livekit-server 1.13.6 in dev mode):
| Two sources in one OBS process | same harness with a second source added: both connect with distinct nonce identities, both receive frames, both tear down cleanly | | Two sources in one OBS process | same harness with a second source added: both connect with distinct nonce identities, both receive frames, both tear down cleanly |
**Not verified anywhere:** **Not verified anywhere:**
- Anything past module load in the OBS GUI. Windows 2026-09-09 confirms the - **macOS in the OBS GUI.** Nobody has opened it. Its artifact is now known to
module loads and its source type appears; whether video actually renders be a correctly-formed, correctly-linked, code-signed `.plugin` bundle
(right way up, right colours), what the A/V sync and latency look like, and (verified by inspecting the shipped v0.1.0 zip — see macOS packaging under
whether a publisher restarting mid-show recovers on screen are all still CI), and it is arm64-only, so Intel Macs are out regardless. "The bundle is
unanswered. Linux and macOS have not been opened in the GUI at all. well formed" is not "OBS loaded it".
- macOS beyond "CI builds and links the real module and the core tests pass". - **Measured** A/V sync and end-to-end latency against the existing egress
Its artifact is a bare `.so` with a relative libobs install name and will path. A live show and several sessions on Linux and Windows produced no
not load in OBS.app — see the macOS packaging gap under CI. reported drift, which is not the same as a measurement — and the timestamp
- Windows beyond "the core library and the WinHTTP backend compile and their caveat above is the reason to want real numbers.
tests pass", from runs predating the current fixes. The WinHTTP backend has - Whether a publisher restarting mid-show recovers cleanly on screen.
never run against a real streamer-tools server, only against the loopback - Token expiry across a session longer than an hour (see below).
test server in `test_api_client`.
- A/V sync and end-to-end latency against the existing egress path.
- Behaviour against the real production streamer-tools server (only against a
stand-in serving the same shapes).
- Token expiry after an hour. Expiry is handled *reactively*: a fatal - Token expiry after an hour. Expiry is handled *reactively*: a fatal
disconnect makes the worker mint a fresh token and reconnect. The design disconnect makes the worker mint a fresh token and reconnect. The design
doc's "proactively refreshed before expiry" is **not** implemented — doc's "proactively refreshed before expiry" is **not** implemented —
@@ -261,8 +263,8 @@ runners available to this repo under the `CyberCoveLLC` org.
| Job | `runs-on` | Runner | State | | Job | `runs-on` | Runner | State |
|---|---|---|---| |---|---|---|---|
| `linux` | `ubuntu-24.04` | `localhost.localdomain` | **Green.** Builds the real adapter against Ubuntu's libobs-dev 30.0.2, runs all six test suites, uploads `build/package` as an artifact | | `linux` | `ubuntu-24.04` | `localhost.localdomain` | **Green.** Builds the real adapter against Ubuntu's libobs-dev 30.0.2, runs all six test suites, uploads `build/package` as an artifact |
| `macos` | `macos-latest` | `home-mac` (Global) | **Green.** Builds libobs 30.0.2 from source, then the real adapter; 6/6 tests; artifact uploaded. But see the macOS packaging gap below | | `macos` | `macos-latest` | `home-mac` (Global) | **Green.** Builds libobs 30.0.2 from source, then the real adapter; 6/6 tests; artifact uploaded as a `.plugin` bundle. Never loaded in OBS.app, and arm64-only — see macOS packaging below |
| `windows` | `windows-latest` | `winvm-builder` (org-scoped) | **Failing, fix pushed and awaiting a completed run.** Every completed run so far has failed; the latest got as far as building libobs and stopped on an OBS-side `OBS::w32-pthreads` target that its own modern CMake path never defines. A bootstrap patch for that gap has been pushed but not yet confirmed by a green run; see below | | `windows` | `windows-latest` | `winvm-builder` (org-scoped) | **Green.** Builds libobs 30.0.2 from source, then the real adapter; 6/6 tests; artifact staged. Was red twice more after the bootstrap was fixed, both times on `test_api_client`'s timeout probe — see "WinHTTP timeouts are not deadlines" below |
The Linux job is pinned to `ubuntu-24.04` rather than `ubuntu-latest`: this The Linux job is pinned to `ubuntu-24.04` rather than `ubuntu-latest`: this
instance's two Linux runners answer `ubuntu-latest` with different releases, instance's two Linux runners answer `ubuntu-latest` with different releases,
@@ -356,26 +358,36 @@ obs-studio, builds libobs from source, builds and links the real adapter,
passes 6/6 tests, and uploads its artifact. `otool -L` on the result shows it passes 6/6 tests, and uploads its artifact. `otool -L` on the result shows it
linked against libobs and `@rpath/liblivekit.dylib`. linked against libobs and `@rpath/liblivekit.dylib`.
### macOS packaging gap (known, unfixed) ### macOS packaging (was described here as broken; it is not)
**The macOS artifact will not load in OBS.app as it stands.** Two reasons, **This section used to claim the macOS artifact was a bare
neither of which CI can catch, because CI only proves it compiles and links: `streamer-tools-camera.so` with a relative libobs install name that "will not
load in OBS.app as it stands". That is wrong, and it contradicted the release
notes for the same build.** Corrected 2026-09-10 by inspecting the shipped
`streamer-tools-camera-v0.1.0-macos.zip` itself:
1. It is a bare `streamer-tools-camera.so`. OBS on macOS loads plugins as - It is a proper bundle: `streamer-tools-camera.plugin/Contents/MacOS/streamer-tools-camera`
`<name>.plugin` bundles (`Contents/MacOS/<name>`, `Contents/Resources/`, (Mach-O **`MH_BUNDLE`**, which is what OBS loads), plus `Info.plist`
an `Info.plist`), which is what obs-plugintemplate's (`CFBundlePackageType BNDL`, `CFBundleExecutable streamer-tools-camera`),
`cmake/macos/helpers.cmake` builds and which this project deliberately did `Contents/Resources/locale/en-US.ini`, and both LiveKit dylibs under
not vendor. `Contents/Frameworks/`.
2. `otool -L` shows the libobs dependency recorded as the relative path - The install names are right, which was the specific doubt. The module loads
`libobs/libobs.framework/Versions/A/libobs`, inherited from the `@rpath/libobs.framework/Versions/A/libobs` and carries
from-source libobs's own install name. A real plugin needs `LC_RPATH @executable_path/../Frameworks` — inside OBS.app that resolves to
`@rpath/libobs.framework/Versions/A/libobs` plus an `LC_RPATH` pointing at `OBS.app/Contents/Frameworks`, where libobs lives. `@rpath/liblivekit.dylib`
`OBS.app/Contents/Frameworks`. resolves through `LC_RPATH @loader_path/../Frameworks` to the bundle's own
copy, and `liblivekit.dylib` finds `liblivekit_ffi.dylib` through its own
`LC_RPATH @loader_path`. Nothing points into a build tree.
- All three binaries carry an `LC_CODE_SIGNATURE` (superblob `0xfade0cc0`),
which is not optional: arm64 macOS refuses to load unsigned code at all.
Fixing this means either vendoring the template's macOS bundle helpers or **The real macOS limitation is different: the bundle is arm64-only.** There is
adding an `install_name_tool` pass and a bundle layout — bounded work, but no x86_64 slice, so Intel Macs cannot load it, and `LSMinimumSystemVersion` is
work that has to be done and checked on an actual Mac. It is deliberately not `13.0`. Shipping a universal binary would mean building both slices and
attempted here rather than guessed at. `lipo`-ing them, on a Mac.
Everything above is static inspection of the artifact. **Nobody has yet opened
it in OBS.app** — well-formed and signed is a strong prior, not a load.
### Where the Windows bootstrap got to ### Where the Windows bootstrap got to
+23
View File
@@ -11,6 +11,7 @@ You may obtain a copy of the License at
#pragma once #pragma once
#include <chrono>
#include <memory> #include <memory>
#include <string> #include <string>
@@ -54,6 +55,22 @@ struct SessionConfig {
int connect_timeout_ms = 15000; int connect_timeout_ms = 15000;
}; };
/// How long the stall-recovery watchdog waits for a decoded video frame
/// before treating the subscription as stalled and forcing a fresh
/// keyframe (see StallWatchdog's comment in session_types.h for why this
/// exists -- unrecoverable packet loss with no PLI/keyframe-request API in
/// the pinned SDK). Long enough that ordinary jitter never trips it (a
/// healthy 30fps subscription delivers a frame at least every ~33ms);
/// short enough a director barely has time to notice before it recovers.
constexpr std::chrono::milliseconds kStallRecoveryTimeout{2000};
/// Ceiling for the backoff between repeated recovery attempts against the
/// SAME stall. Starts at kStallRecoveryTimeout and doubles each attempt, so
/// a genuinely gone publisher (crashed encoder, dead upstream network) is
/// retried every 2s, 4s, 8s, ... 30s rather than hammered every 2 seconds
/// for the rest of the show.
constexpr std::chrono::milliseconds kStallRecoveryMaxBackoff{30000};
/// Wraps livekit::Room for exactly one subscribed slot. /// Wraps livekit::Room for exactly one subscribed slot.
/// ///
/// Threading contract, which the OBS adapter depends on: /// Threading contract, which the OBS adapter depends on:
@@ -67,6 +84,11 @@ struct SessionConfig {
/// call. /// call.
/// - The state handler is invoked from whichever thread observed the /// - The state handler is invoked from whichever thread observed the
/// change. It must not block and must not call back into this object. /// change. It must not block and must not call back into this object.
/// - The diagnostic handler (currently just the stall-recovery watchdog,
/// see kStallRecoveryTimeout below) may be invoked from the internal
/// command-queue worker thread or a video reader thread. Same rules as
/// the state handler: must not block, must not call back into this
/// object.
/// - All handlers must be installed before connect(); they are not /// - All handlers must be installed before connect(); they are not
/// synchronised against a running session. /// synchronised against a running session.
class LiveKitSession { class LiveKitSession {
@@ -80,6 +102,7 @@ public:
void setVideoHandler(VideoFrameHandler handler); void setVideoHandler(VideoFrameHandler handler);
void setAudioHandler(AudioFrameHandler handler); void setAudioHandler(AudioFrameHandler handler);
void setStateHandler(SessionStateHandler handler); void setStateHandler(SessionStateHandler handler);
void setDiagnosticHandler(DiagnosticHandler handler);
/// Connect and start subscribing. Returns true once the room is up; the /// Connect and start subscribing. Returns true once the room is up; the
/// selected slot's tracks may still arrive later (or not at all, if the /// selected slot's tracks may still arrive later (or not at all, if the
+86
View File
@@ -19,6 +19,7 @@ You may obtain a copy of the License at
// self-consistent -- all live here, and LiveKitSession is the (much thinner) // self-consistent -- all live here, and LiveKitSession is the (much thinner)
// piece that wires real SDK callbacks into them. // piece that wires real SDK callbacks into them.
#include <chrono>
#include <cstddef> #include <cstddef>
#include <cstdint> #include <cstdint>
#include <functional> #include <functional>
@@ -137,6 +138,82 @@ private:
bool has_audio_ = false; bool has_audio_ = false;
}; };
// ---------------------------------------------------------------------------
// Stall-recovery watchdog (pure timing/decision logic)
// ---------------------------------------------------------------------------
/// Decides WHEN to force a fresh keyframe on an already-subscribed video
/// track. It does not touch LiveKit or OBS at all -- LiveKitSession::Impl
/// (session.cpp) is what actually carries the decision out
/// (RemoteTrackPublication::setEnabled(false) then setEnabled(true)), and
/// only from its SDK command-queue thread, the same rule every other SDK
/// interaction in that file already follows.
///
/// Why this exists: measured on the live server, comparing subscribers in
/// the same LiveKit room over the same 30-minute window, every OBS plugin
/// connection racked up ~862-1099 nackMisses and ~2200-3100 nackRepeated --
/// nackMisses means the subscriber asked the SFU to retransmit a packet
/// that had already aged out of its send buffer, i.e. unrecoverable loss --
/// while a browser subscriber in the same room saw 0 and 0. A decoder that
/// loses a frame that way cannot resync without a fresh keyframe. Every OBS
/// plugin connection also sat at `plis` == 2 for a multi-hour session (a
/// browser adapts and asks for keyframes normally), and grepping the pinned
/// client-sdk-cpp (1.10.1) headers turns up no PLI/keyframe-request API at
/// all. So today, once that happens, the source just stays broken for the
/// rest of the show -- "drops at random and never recovers". Toggling the
/// subscription off and back on is the one lever this SDK exposes that
/// forces the SFU to stop and restart delivery of the track, and a restart
/// always begins with a keyframe. This class is the "have we gone too long
/// without a frame, and is it still worth trying again" clock behind that
/// lever; see LiveKitSession::Impl::recoverVideo() for where it is pulled.
///
/// Not thread-safe on its own, deliberately -- same contract as
/// SessionStateMachine above: LiveKitSession::Impl owns the lock (in
/// practice the same state_mutex that guards `machine`).
class StallWatchdog {
public:
StallWatchdog(std::chrono::milliseconds timeout, std::chrono::milliseconds max_backoff);
/// Call whenever whether a frame could legitimately arrive right now
/// changes: true once a video track is subscribed (and unmuted), false
/// on detach/unsubscribe/mute/disconnect. Flipping to false always
/// clears all timing state; flipping back to true always starts a
/// brand-new grace period rather than measuring from a stale timestamp.
/// That is what stops an unmute -- or an ordinary publisher swap --
/// from firing the INSTANT it resumes, off a "last frame" that might
/// actually be minutes old: a muted, disabled or unsubscribed track, an
/// audio-only source, or a disconnected session must never trip this.
void setExpectingFrames(bool expecting, std::chrono::steady_clock::time_point now);
/// Call every time a decoded video frame is actually delivered.
void onFrameDelivered(std::chrono::steady_clock::time_point now);
/// Call periodically (finer-grained than the configured timeout).
/// Returns true exactly when a recovery attempt should be made right
/// now; each true also arms the backoff before the next one is even
/// considered, so a caller polling in a tight loop still cannot fire
/// back-to-back attempts against a publisher that never comes back --
/// see the class comment: hammering every 2 seconds forever against a
/// genuinely gone publisher is worse than a frozen source.
bool poll(std::chrono::steady_clock::time_point now);
/// Attempts made since the current stall started (since the last frame,
/// or since expecting-frames most recently became true). Reset by
/// onFrameDelivered and by setExpectingFrames.
int attemptsThisStall() const { return attempts_; }
private:
std::chrono::milliseconds timeout_;
std::chrono::milliseconds max_backoff_;
bool expecting_ = false;
bool have_baseline_ = false;
std::chrono::steady_clock::time_point baseline_{};
std::chrono::milliseconds backoff_{};
std::chrono::steady_clock::time_point next_attempt_allowed_{};
int attempts_ = 0;
};
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
// Frames handed to the OBS adapter // Frames handed to the OBS adapter
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
@@ -177,4 +254,13 @@ using VideoFrameHandler = std::function<void(const VideoFrameData &)>;
using AudioFrameHandler = std::function<void(const AudioFrameData &)>; using AudioFrameHandler = std::function<void(const AudioFrameData &)>;
using SessionStateHandler = std::function<void(SessionState state, const std::string &detail)>; using SessionStateHandler = std::function<void(SessionState state, const std::string &detail)>;
/// Severity for LiveKitSession's own diagnostic log lines (currently just
/// the stall-recovery watchdog). Kept separate from OBS's LOG_* levels and
/// from the SDK's own livekit::LogLevel so core/ stays free of any OBS
/// dependency -- the adapter maps this onto obs_log the same way it already
/// maps livekit::LogLevel (see plugin-main.cpp's livekit log bridge).
enum class DiagnosticLevel { Info, Warning };
using DiagnosticHandler = std::function<void(DiagnosticLevel level, const std::string &message)>;
} // namespace stplugin } // namespace stplugin
+261 -11
View File
@@ -17,6 +17,7 @@ You may obtain a copy of the License at
#include <deque> #include <deque>
#include <exception> #include <exception>
#include <mutex> #include <mutex>
#include <string>
#include <thread> #include <thread>
#include <utility> #include <utility>
#include <vector> #include <vector>
@@ -30,6 +31,7 @@ You may obtain a copy of the License at
#include <livekit/room_delegate.h> #include <livekit/room_delegate.h>
#include <livekit/room_event_types.h> #include <livekit/room_event_types.h>
#include <livekit/track.h> #include <livekit/track.h>
#include <livekit/track_publication.h>
#include <livekit/video_frame.h> #include <livekit/video_frame.h>
#include <livekit/video_stream.h> #include <livekit/video_stream.h>
@@ -138,6 +140,13 @@ int &globalRefCount()
return n; return n;
} }
// --- Stall-recovery watchdog tuning -----------------------------------------
/// How often the watchdog thread checks StallWatchdog's clock. Deliberately
/// finer than kStallRecoveryTimeout (session.h) so detection latency tracks
/// the threshold itself, not the threshold plus a whole polling period.
constexpr std::chrono::milliseconds kWatchdogPollInterval{250};
} // namespace } // namespace
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
@@ -145,11 +154,12 @@ int &globalRefCount()
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
struct LiveKitSession::Impl : public livekit::RoomDelegate { struct LiveKitSession::Impl : public livekit::RoomDelegate {
enum class CommandType { AttachVideo, DetachVideo, AttachAudio, DetachAudio, Stop }; enum class CommandType { AttachVideo, DetachVideo, AttachAudio, DetachAudio, RecoverVideo, Stop };
struct Command { struct Command {
CommandType type; CommandType type;
std::shared_ptr<livekit::Track> track; std::shared_ptr<livekit::Track> track;
std::shared_ptr<livekit::RemoteTrackPublication> publication;
}; };
livekit::Room room; livekit::Room room;
@@ -157,10 +167,14 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
mutable std::mutex state_mutex; mutable std::mutex state_mutex;
SessionStateMachine machine; SessionStateMachine machine;
// Guarded by state_mutex too, same contract as `machine` -- see
// StallWatchdog's own comment for what it decides and why it exists.
StallWatchdog stall_watchdog{kStallRecoveryTimeout, kStallRecoveryMaxBackoff};
VideoFrameHandler on_video; VideoFrameHandler on_video;
AudioFrameHandler on_audio; AudioFrameHandler on_audio;
SessionStateHandler on_state; SessionStateHandler on_state;
DiagnosticHandler on_diagnostic;
std::atomic<std::uint64_t> video_frames{0}; std::atomic<std::uint64_t> video_frames{0};
std::atomic<std::uint64_t> audio_frames{0}; std::atomic<std::uint64_t> audio_frames{0};
@@ -170,7 +184,10 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
// livekit::AudioStream happens on `worker`, never on a room event thread: // livekit::AudioStream happens on `worker`, never on a room event thread:
// the SDK's room callbacks run on its own event thread and blocking or // the SDK's room callbacks run on its own event thread and blocking or
// re-entering there stalls every other event (and Room::disconnect() from // re-entering there stalls every other event (and Room::disconnect() from
// inside one is documented to deadlock outright). // inside one is documented to deadlock outright). The stall-recovery
// watchdog thread follows the same rule: it never touches
// RemoteTrackPublication itself, only posts CommandType::RecoverVideo
// and lets the worker thread do it (see recoverVideo() below).
std::mutex queue_mutex; std::mutex queue_mutex;
std::condition_variable queue_cv; std::condition_variable queue_cv;
std::deque<Command> queue; std::deque<Command> queue;
@@ -183,6 +200,26 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
std::shared_ptr<livekit::AudioStream> audio_stream; std::shared_ptr<livekit::AudioStream> audio_stream;
std::thread audio_thread; std::thread audio_thread;
// The publication/track backing the CURRENT video subscription, kept
// around purely so recoverVideo() and the mute-change handlers have
// something to act on without reaching into `video_stream` (which is
// worker-thread-exclusive, per the comment above). Set together in
// attachVideo(), cleared together in detachVideo(), both on the worker
// thread; read from the room event thread (handleMuteChange) and the
// worker thread (recoverVideo()) under this mutex.
std::mutex video_track_mutex;
std::shared_ptr<livekit::Track> current_video_track;
std::shared_ptr<livekit::RemoteTrackPublication> current_video_publication;
// The watchdog's own timer thread. It owns no SDK state and calls no SDK
// method directly -- see the queue comment above. `watchdog_mutex` only
// ever guards the shutdown flag/condvar pair, never `stall_watchdog`
// (that is guarded by `state_mutex`, alongside `machine`).
std::mutex watchdog_mutex;
std::condition_variable watchdog_cv;
bool watchdog_running = false;
std::thread watchdog_thread;
bool connected = false; bool connected = false;
~Impl() override = default; ~Impl() override = default;
@@ -207,17 +244,29 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
handler(state, detail); handler(state, detail);
} }
void post(CommandType type, std::shared_ptr<livekit::Track> track = nullptr) void post(CommandType type, std::shared_ptr<livekit::Track> track = nullptr,
std::shared_ptr<livekit::RemoteTrackPublication> publication = nullptr)
{ {
{ {
std::lock_guard<std::mutex> guard(queue_mutex); std::lock_guard<std::mutex> guard(queue_mutex);
if (!worker_running) if (!worker_running)
return; return;
queue.push_back(Command{type, std::move(track)}); queue.push_back(Command{type, std::move(track), std::move(publication)});
} }
queue_cv.notify_one(); queue_cv.notify_one();
} }
void logDiagnostic(DiagnosticLevel level, const std::string &message)
{
DiagnosticHandler handler;
{
std::lock_guard<std::mutex> guard(state_mutex);
handler = on_diagnostic;
}
if (handler)
handler(level, message);
}
// Handles the wanted video track once matched, shared by onTrackSubscribed // Handles the wanted video track once matched, shared by onTrackSubscribed
// (a fresh subscription) and attachExistingTracks (one already up when // (a fresh subscription) and attachExistingTracks (one already up when
// this session started watching). Two responsibilities that only make // this session started watching). Two responsibilities that only make
@@ -261,7 +310,42 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
// it already had, which is the pre-existing behaviour. // it already had, which is the pre-existing behaviour.
} }
} }
post(CommandType::AttachVideo, track); // `publication` rides along so attachVideo() can remember it: it is
// the handle the stall-recovery watchdog later toggles
// (setEnabled(false)/(true)) to force a fresh keyframe. See
// recoverVideo() and StallWatchdog's comment in session_types.h.
post(CommandType::AttachVideo, track, publication);
}
// Fired for ANY track (any participant, any kind) muting or unmuting.
// Filtered down to "is this the video publication we are currently
// watching" by SID, which also naturally excludes every audio mute and
// every other participant's tracks without a separate identity/kind
// check.
//
// Why this exists: the stall-recovery watchdog (see StallWatchdog's
// comment) treats "no decoded frame for kStallRecoveryTimeout" as a
// stall worth toggling the subscription over. A publisher who
// legitimately turned their camera off produces exactly that symptom on
// purpose, and toggling their subscription every couple of seconds for
// as long as they stay off would be an endless, pointless loop against
// healthy behaviour. Muting suspends the watchdog's clock entirely;
// unmuting starts a brand-new grace period rather than reading "muted
// for twenty minutes" as "stalled for twenty minutes".
void handleMuteChange(const std::shared_ptr<livekit::TrackPublication> &publication, bool unmuted)
{
if (!publication || publication->kind() != livekit::TrackKind::KIND_VIDEO)
return;
std::string current_sid;
{
std::lock_guard<std::mutex> guard(video_track_mutex);
if (current_video_publication)
current_sid = current_video_publication->sid();
}
if (current_sid.empty() || publication->sid() != current_sid)
return;
std::lock_guard<std::mutex> guard(state_mutex);
stall_watchdog.setExpectingFrames(unmuted, std::chrono::steady_clock::now());
} }
// --- RoomDelegate ------------------------------------------------------ // --- RoomDelegate ------------------------------------------------------
@@ -304,6 +388,16 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
post(CommandType::DetachAudio); post(CommandType::DetachAudio);
} }
void onTrackMuted(livekit::Room &, const livekit::TrackMutedEvent &event) override
{
handleMuteChange(event.publication, false);
}
void onTrackUnmuted(livekit::Room &, const livekit::TrackUnmutedEvent &event) override
{
handleMuteChange(event.publication, true);
}
void onReconnecting(livekit::Room &, const livekit::ReconnectingEvent &) override void onReconnecting(livekit::Room &, const livekit::ReconnectingEvent &) override
{ {
mutateState([](SessionStateMachine &m) { m.onReconnecting(); }); mutateState([](SessionStateMachine &m) { m.onReconnecting(); });
@@ -359,7 +453,7 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
void workerLoop() void workerLoop()
{ {
for (;;) { for (;;) {
Command command{CommandType::Stop, nullptr}; Command command{CommandType::Stop, nullptr, nullptr};
{ {
std::unique_lock<std::mutex> lock(queue_mutex); std::unique_lock<std::mutex> lock(queue_mutex);
queue_cv.wait(lock, [this] { return !queue.empty(); }); queue_cv.wait(lock, [this] { return !queue.empty(); });
@@ -369,7 +463,7 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
switch (command.type) { switch (command.type) {
case CommandType::AttachVideo: case CommandType::AttachVideo:
attachVideo(command.track); attachVideo(command.track, command.publication);
break; break;
case CommandType::DetachVideo: case CommandType::DetachVideo:
detachVideo(); detachVideo();
@@ -380,6 +474,9 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
case CommandType::DetachAudio: case CommandType::DetachAudio:
detachAudio(); detachAudio();
break; break;
case CommandType::RecoverVideo:
recoverVideo();
break;
case CommandType::Stop: case CommandType::Stop:
detachVideo(); detachVideo();
detachAudio(); detachAudio();
@@ -388,7 +485,103 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
} }
} }
void attachVideo(const std::shared_ptr<livekit::Track> &track) // --- stall-recovery watchdog --------------------------------------------
//
// See StallWatchdog's comment (session_types.h) for the measured
// evidence and why toggling the publication is the only lever
// available. This thread does exactly one thing: tick StallWatchdog's
// clock and, when it says to, post CommandType::RecoverVideo so the
// worker thread does the actual SDK call. It never touches
// livekit::VideoStream, livekit::Room or RemoteTrackPublication itself.
void startWatchdog()
{
{
std::lock_guard<std::mutex> guard(watchdog_mutex);
watchdog_running = true;
}
watchdog_thread = std::thread([this] { watchdogLoop(); });
}
void stopWatchdog()
{
{
std::lock_guard<std::mutex> guard(watchdog_mutex);
if (!watchdog_running)
return;
watchdog_running = false;
}
watchdog_cv.notify_all();
if (watchdog_thread.joinable())
watchdog_thread.join();
}
void watchdogLoop()
{
std::unique_lock<std::mutex> lock(watchdog_mutex);
while (watchdog_running) {
// kWatchdogPollInterval is finer than kStallRecoveryTimeout so
// detection latency tracks the threshold itself rather than the
// threshold plus a whole polling period; woken early on
// shutdown by stopWatchdog()'s notify_all().
watchdog_cv.wait_for(lock, kWatchdogPollInterval);
if (!watchdog_running)
break;
lock.unlock();
bool should_fire;
{
std::lock_guard<std::mutex> guard(state_mutex);
should_fire = stall_watchdog.poll(std::chrono::steady_clock::now());
}
if (should_fire)
post(CommandType::RecoverVideo);
lock.lock();
}
}
// Actually pulls the lever: toggles the video publication off and back
// on, which makes the SFU stop and restart delivery of that track --
// and a restart always begins with a keyframe. Only ever called from
// the worker thread (via CommandType::RecoverVideo), same as every
// other RemoteTrackPublication/VideoStream call in this file.
void recoverVideo()
{
std::shared_ptr<livekit::RemoteTrackPublication> publication;
std::shared_ptr<livekit::Track> track;
{
std::lock_guard<std::mutex> guard(video_track_mutex);
publication = current_video_publication;
track = current_video_track;
}
// Detached, replaced, or muted between the watchdog deciding to
// fire and the worker getting to this command -- nothing to do, and
// silently: this is the expected shape of the race, not a failure
// worth logging.
if (!publication || !track || track->muted())
return;
int attempt = 0;
{
std::lock_guard<std::mutex> guard(state_mutex);
attempt = stall_watchdog.attemptsThisStall();
}
logDiagnostic(DiagnosticLevel::Warning,
"no decoded video frame for >= " + std::to_string(kStallRecoveryTimeout.count()) +
"ms; toggling the subscription to force a fresh keyframe (attempt " +
std::to_string(attempt) + ")");
try {
publication->setEnabled(false);
publication->setEnabled(true);
} catch (const std::exception &e) {
logDiagnostic(DiagnosticLevel::Warning, std::string("stall-recovery toggle failed: ") + e.what());
}
}
void attachVideo(const std::shared_ptr<livekit::Track> &track,
const std::shared_ptr<livekit::RemoteTrackPublication> &publication)
{ {
if (!track) if (!track)
return; return;
@@ -413,6 +606,18 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
if (!stream) if (!stream)
return; return;
{
std::lock_guard<std::mutex> guard(video_track_mutex);
current_video_track = track;
current_video_publication = publication;
}
{
// A fresh subscription (or a publisher swap) starts a brand-new
// grace period -- see StallWatchdog::setExpectingFrames.
std::lock_guard<std::mutex> guard(state_mutex);
stall_watchdog.setExpectingFrames(true, std::chrono::steady_clock::now());
}
video_stream = stream; video_stream = stream;
video_thread = std::thread([this, stream] { videoReaderLoop(stream); }); video_thread = std::thread([this, stream] { videoReaderLoop(stream); });
mutateState([](SessionStateMachine &m) { m.onVideoAttached(); }); mutateState([](SessionStateMachine &m) { m.onVideoAttached(); });
@@ -426,6 +631,20 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
video_thread.join(); video_thread.join();
const bool had = static_cast<bool>(video_stream); const bool had = static_cast<bool>(video_stream);
video_stream.reset(); video_stream.reset();
{
std::lock_guard<std::mutex> guard(video_track_mutex);
current_video_track.reset();
current_video_publication.reset();
}
{
// Nothing subscribed means nothing expected -- see
// StallWatchdog::setExpectingFrames. Unconditional, not gated on
// `had`: this also covers the detachVideo() at the top of
// attachVideo() above, which is exactly the publisher-swap
// moment the grace period needs to restart from.
std::lock_guard<std::mutex> guard(state_mutex);
stall_watchdog.setExpectingFrames(false, std::chrono::steady_clock::now());
}
if (had) if (had)
mutateState([](SessionStateMachine &m) { m.onVideoDetached(); }); mutateState([](SessionStateMachine &m) { m.onVideoDetached(); });
} }
@@ -548,6 +767,23 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
out.plane_count = count; out.plane_count = count;
video_frames.fetch_add(1); video_frames.fetch_add(1);
// Tell the stall-recovery watchdog a frame actually made it all the
// way to "about to hand to OBS" -- not merely that the SDK's queue
// produced an event, which the drop paths above also see. See
// StallWatchdog's comment for why this exists.
int attempts_before_this_frame = 0;
{
std::lock_guard<std::mutex> guard(state_mutex);
attempts_before_this_frame = stall_watchdog.attemptsThisStall();
stall_watchdog.onFrameDelivered(std::chrono::steady_clock::now());
}
if (attempts_before_this_frame > 0) {
logDiagnostic(DiagnosticLevel::Info, "video resumed after " +
std::to_string(attempts_before_this_frame) +
" stall-recovery attempt(s)");
}
handler(out); handler(out);
} }
@@ -633,6 +869,12 @@ void LiveKitSession::setStateHandler(SessionStateHandler handler)
impl_->on_state = std::move(handler); impl_->on_state = std::move(handler);
} }
void LiveKitSession::setDiagnosticHandler(DiagnosticHandler handler)
{
std::lock_guard<std::mutex> guard(impl_->state_mutex);
impl_->on_diagnostic = std::move(handler);
}
bool LiveKitSession::connect(const SessionConfig &config) bool LiveKitSession::connect(const SessionConfig &config)
{ {
if (impl_->connected) if (impl_->connected)
@@ -652,6 +894,10 @@ bool LiveKitSession::connect(const SessionConfig &config)
} }
impl_->startWorker(); impl_->startWorker();
// Runs for the lifetime of the worker: connected-but-nothing-subscribed
// is a no-op for StallWatchdog (see setExpectingFrames), so there is no
// reason to start/stop it separately from the worker it posts to.
impl_->startWatchdog();
livekit::RoomOptions options; livekit::RoomOptions options;
// auto_subscribe is what makes track_subscribed events (and therefore any // auto_subscribe is what makes track_subscribed events (and therefore any
@@ -680,6 +926,7 @@ bool LiveKitSession::connect(const SessionConfig &config)
} catch (const std::exception &e) { } catch (const std::exception &e) {
ok = false; ok = false;
impl_->mutateState([&](SessionStateMachine &m) { m.onConnectFailed(e.what()); }); impl_->mutateState([&](SessionStateMachine &m) { m.onConnectFailed(e.what()); });
impl_->stopWatchdog();
impl_->stopWorker(); impl_->stopWorker();
impl_->room.setDelegate(nullptr); impl_->room.setDelegate(nullptr);
return false; return false;
@@ -689,6 +936,7 @@ bool LiveKitSession::connect(const SessionConfig &config)
impl_->mutateState([](SessionStateMachine &m) { impl_->mutateState([](SessionStateMachine &m) {
m.onConnectFailed("could not connect to LiveKit (check the server URL, or the token may have expired)"); m.onConnectFailed("could not connect to LiveKit (check the server URL, or the token may have expired)");
}); });
impl_->stopWatchdog();
impl_->stopWorker(); impl_->stopWorker();
impl_->room.setDelegate(nullptr); impl_->room.setDelegate(nullptr);
return false; return false;
@@ -705,9 +953,11 @@ void LiveKitSession::disconnect()
if (!impl_) if (!impl_)
return; return;
// Order matters: stop the readers first so nothing is mid-read on a // Order matters: stop the watchdog and the readers first so nothing is
// stream the room is about to tear down, then disconnect the room, then // mid-read (or about to post a recovery command) on a stream the room
// drop the delegate so no event can arrive at a half-destroyed object. // is about to tear down, then disconnect the room, then drop the
// delegate so no event can arrive at a half-destroyed object.
impl_->stopWatchdog();
impl_->stopWorker(); impl_->stopWorker();
if (impl_->connected) { if (impl_->connected) {
+60
View File
@@ -11,6 +11,8 @@ You may obtain a copy of the License at
#include "stplugin/session_types.h" #include "stplugin/session_types.h"
#include <algorithm>
namespace stplugin { namespace stplugin {
const char *describePixelFormat(PixelFormat format) const char *describePixelFormat(PixelFormat format)
@@ -175,4 +177,62 @@ void SessionStateMachine::onAudioDetached()
has_audio_ = false; has_audio_ = false;
} }
// ---------------------------------------------------------------------------
// StallWatchdog
// ---------------------------------------------------------------------------
StallWatchdog::StallWatchdog(std::chrono::milliseconds timeout, std::chrono::milliseconds max_backoff)
: timeout_(timeout), max_backoff_(max_backoff), backoff_(timeout)
{
}
void StallWatchdog::setExpectingFrames(bool expecting, std::chrono::steady_clock::time_point now)
{
expecting_ = expecting;
// Always re-baseline from `now`, whichever direction this flips.
// Losing the baseline (rather than, say, keeping the old one around for
// when expecting_ next becomes true) is what stops a track that was
// muted for the last twenty minutes from reading as "twenty minutes
// stalled" the instant it unmutes.
have_baseline_ = expecting;
baseline_ = now;
backoff_ = timeout_;
next_attempt_allowed_ = now;
attempts_ = 0;
}
void StallWatchdog::onFrameDelivered(std::chrono::steady_clock::time_point now)
{
have_baseline_ = true;
baseline_ = now;
backoff_ = timeout_;
// A recovered stream must be able to fire again the moment a FRESH
// stall clears the (now-reset) timeout, not sit throttled by whatever
// backoff a previous, unrelated stall had climbed to -- next_attempt_
// allowed_ belongs to that old stall and is meaningless once frames are
// flowing again.
next_attempt_allowed_ = now;
attempts_ = 0;
}
bool StallWatchdog::poll(std::chrono::steady_clock::time_point now)
{
if (!expecting_ || !have_baseline_)
return false;
if (now - baseline_ < timeout_)
return false;
if (now < next_attempt_allowed_)
return false;
++attempts_;
// Next attempt against this SAME stall is not allowed until the backoff
// elapses, and the backoff itself doubles (capped) each time -- 2s, 4s,
// 8s, ... up to max_backoff_ -- so a publisher that is genuinely gone
// gets progressively less frequent toggles instead of one every 2
// seconds for the rest of the show.
next_attempt_allowed_ = now + backoff_;
backoff_ = std::min(backoff_ * 2, max_backoff_);
return true;
}
} // namespace stplugin } // namespace stplugin
+136 -2
View File
@@ -14,8 +14,10 @@ You may obtain a copy of the License at
// COVERED headlessly -- the session's own decision-making: the state // COVERED headlessly -- the session's own decision-making: the state
// machine's transitions (including the publisher-swap and reconnect paths // machine's transitions (including the publisher-swap and reconnect paths
// that motivated this plugin), track selection, frame geometry validation, // that motivated this plugin), track selection, frame geometry validation,
// and the real connect() failure paths against the real SDK (bad URL, // the stall-recovery watchdog's timing/backoff decisions (StallWatchdog,
// unreachable host, garbage token). // driven with a fake clock -- see its own section below), and the real
// connect() failure paths against the real SDK (bad URL, unreachable
// host, garbage token).
// //
// NOT COVERED here -- anything that needs a LiveKit server to answer: // NOT COVERED here -- anything that needs a LiveKit server to answer:
// a successful connect, actual subscription, and actual decoded frames // a successful connect, actual subscription, and actual decoded frames
@@ -209,6 +211,134 @@ void testFailureAndRecovery()
ST_ASSERT(idle.state() == SessionState::Idle); ST_ASSERT(idle.state() == SessionState::Idle);
} }
// ---------------------------------------------------------------------------
// StallWatchdog -- the stall-recovery watchdog's pure timing/decision logic.
//
// This is deliberately driven with an explicit, fake clock (arbitrary
// steady_clock::time_points built by hand, never std::chrono::...::now())
// rather than real sleeps: every case below needs to be exact about
// "1999ms in" vs "2001ms in" and about backoff boundaries, and a test that
// actually slept for 30+ seconds to exercise the backoff ceiling would be
// exactly the kind of slow, flaky test this project's whole headless-test
// philosophy exists to avoid. See StallWatchdog's own comment
// (session_types.h) for the measured server evidence this exists to fix.
// ---------------------------------------------------------------------------
void testStallWatchdogFiresAfterThreshold()
{
const auto t0 = std::chrono::steady_clock::now();
StallWatchdog w(std::chrono::milliseconds(2000), std::chrono::milliseconds(30000));
// Nothing subscribed yet: polling is a no-op, no matter how much time
// has "passed" -- an audio-only source or a disconnected session must
// never fire.
ST_ASSERT(!w.poll(t0 + std::chrono::milliseconds(10000)));
// A track becomes subscribed. Still well under the threshold: quiet.
w.setExpectingFrames(true, t0);
ST_ASSERT(!w.poll(t0 + std::chrono::milliseconds(500)));
ST_ASSERT(!w.poll(t0 + std::chrono::milliseconds(1999)));
// No frame ever arrived, and the threshold has now elapsed: fires
// exactly once when asked right at/after the boundary.
ST_ASSERT(w.poll(t0 + std::chrono::milliseconds(2000)));
ST_ASSERT_EQ(w.attemptsThisStall(), 1);
// A frame arriving resets the clock -- the far more common case in a
// healthy stream, where onFrameDelivered() is called every ~33ms and
// poll() (every kWatchdogPollInterval) never sees 2000ms of silence.
StallWatchdog healthy(std::chrono::milliseconds(2000), std::chrono::milliseconds(30000));
healthy.setExpectingFrames(true, t0);
for (int ms = 0; ms <= 5000; ms += 33)
healthy.onFrameDelivered(t0 + std::chrono::milliseconds(ms));
ST_ASSERT(!healthy.poll(t0 + std::chrono::milliseconds(5010)));
ST_ASSERT_EQ(healthy.attemptsThisStall(), 0);
}
void testStallWatchdogDoesNotFireWhenNotExpectingFrames()
{
const auto t0 = std::chrono::steady_clock::now();
StallWatchdog w(std::chrono::milliseconds(2000), std::chrono::milliseconds(30000));
// A muted (or disabled/unsubscribed) track is expected silence, not a
// stall -- setExpectingFrames(false, ...) is exactly what
// LiveKitSession::Impl::handleMuteChange (and detachVideo()) call in
// that case. It must not fire no matter how long it stays that way.
w.setExpectingFrames(false, t0);
ST_ASSERT(!w.poll(t0 + std::chrono::milliseconds(2000)));
ST_ASSERT(!w.poll(t0 + std::chrono::milliseconds(60000)));
ST_ASSERT(!w.poll(t0 + std::chrono::milliseconds(600000)));
// Un-muting (setExpectingFrames(true, ...)) starts a BRAND NEW grace
// period from that moment -- it must not read "was silent for ten
// minutes" as "stalled for ten minutes" and fire immediately.
const auto unmuted_at = t0 + std::chrono::milliseconds(600000);
w.setExpectingFrames(true, unmuted_at);
ST_ASSERT(!w.poll(unmuted_at + std::chrono::milliseconds(1999)));
ST_ASSERT(w.poll(unmuted_at + std::chrono::milliseconds(2000)));
}
void testStallWatchdogBacksOffRatherThanLooping()
{
const auto t0 = std::chrono::steady_clock::now();
StallWatchdog w(std::chrono::milliseconds(2000), std::chrono::milliseconds(30000));
w.setExpectingFrames(true, t0);
// First attempt at the threshold.
auto now = t0 + std::chrono::milliseconds(2000);
ST_ASSERT(w.poll(now));
ST_ASSERT_EQ(w.attemptsThisStall(), 1);
// A genuinely gone publisher: no frame ever comes back. Immediately
// asking again (the naive "retry every poll interval forever" a
// watchdog without backoff would do) must NOT fire -- that is precisely
// the "hammered every 2 seconds forever" this backoff exists to avoid.
ST_ASSERT(!w.poll(now + std::chrono::milliseconds(250)));
ST_ASSERT(!w.poll(now + std::chrono::milliseconds(1999)));
// The backoff after attempt 1 is the base timeout (2000ms): the second
// attempt is allowed at +2000ms from the first, not before.
now += std::chrono::milliseconds(2000);
ST_ASSERT(w.poll(now));
ST_ASSERT_EQ(w.attemptsThisStall(), 2);
// Backoff doubles: the third attempt needs a 4000ms gap, not 2000ms.
ST_ASSERT(!w.poll(now + std::chrono::milliseconds(3999)));
now += std::chrono::milliseconds(4000);
ST_ASSERT(w.poll(now));
ST_ASSERT_EQ(w.attemptsThisStall(), 3);
// ... and again to 8000ms, and again to 16000ms.
ST_ASSERT(!w.poll(now + std::chrono::milliseconds(7999)));
now += std::chrono::milliseconds(8000);
ST_ASSERT(w.poll(now));
ST_ASSERT_EQ(w.attemptsThisStall(), 4);
ST_ASSERT(!w.poll(now + std::chrono::milliseconds(15999)));
now += std::chrono::milliseconds(16000);
ST_ASSERT(w.poll(now));
ST_ASSERT_EQ(w.attemptsThisStall(), 5);
// The backoff is capped: doubling 16000ms would be 32000ms, but it
// never exceeds max_backoff (30000ms) no matter how many attempts have
// failed, so a publisher that comes back after an hour is still
// retried at a bounded cadence, not abandoned.
for (int i = 0; i < 6; ++i) {
ST_ASSERT(!w.poll(now + std::chrono::milliseconds(29999)));
now += std::chrono::milliseconds(30000);
ST_ASSERT(w.poll(now));
}
// A frame finally arrives: the stall is over, and the NEXT one (a fresh
// stall, not a continuation) starts back at the base cadence rather
// than staying parked at the 30s ceiling forever.
w.onFrameDelivered(now);
ST_ASSERT_EQ(w.attemptsThisStall(), 0);
ST_ASSERT(!w.poll(now + std::chrono::milliseconds(1999)));
ST_ASSERT(w.poll(now + std::chrono::milliseconds(2000)));
ST_ASSERT_EQ(w.attemptsThisStall(), 1);
}
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
// Real SDK, failure paths only (no LiveKit server available headlessly) // Real SDK, failure paths only (no LiveKit server available headlessly)
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
@@ -335,6 +465,10 @@ int main()
testReconnect(); testReconnect();
testFailureAndRecovery(); testFailureAndRecovery();
testStallWatchdogFiresAfterThreshold();
testStallWatchdogDoesNotFireWhenNotExpectingFrames();
testStallWatchdogBacksOffRatherThanLooping();
LiveKitSession::globalInitialize(); LiveKitSession::globalInitialize();
testConnectRejectsIncompleteConfig(); testConnectRejectsIncompleteConfig();
testConnectToUnreachableServerFailsCleanly(); testConnectToUnreachableServerFailsCleanly();
+7
View File
@@ -375,6 +375,13 @@ void *sourceCreate(obs_data_t *settings, obs_source_t *source)
self->session->setVideoHandler([self](const VideoFrameData &frame) { outputVideoFrame(self, frame); }); self->session->setVideoHandler([self](const VideoFrameData &frame) { outputVideoFrame(self, frame); });
self->session->setAudioHandler([self](const AudioFrameData &frame) { outputAudioFrame(self, frame); }); self->session->setAudioHandler([self](const AudioFrameData &frame) { outputAudioFrame(self, frame); });
// The stall-recovery watchdog (core/src/session.cpp) is the only thing
// that currently uses this: it logs each toggle-the-subscription
// recovery attempt, and its eventual success, so a stalled-and-fixed
// camera is diagnosable from an OBS log afterward instead of invisible.
self->session->setDiagnosticHandler([](DiagnosticLevel level, const std::string &message) {
obs_log(level == DiagnosticLevel::Warning ? LOG_WARNING : LOG_INFO, "%s", message.c_str());
});
self->session->setStateHandler([self](SessionState state, const std::string &detail) { self->session->setStateHandler([self](SessionState state, const std::string &detail) {
self->setStatus(detail.empty() ? describeSessionState(state) : detail); self->setStatus(detail.empty() ? describeSessionState(state) : detail);
self->status_is_error.store(state == SessionState::Failed); self->status_is_error.store(state == SessionState::Failed);