CyberCoveLLC/obs-streamer-tools-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

fix(session): recover stalled video subscriptions with a keyframe-forcing watchdog
Build / macOS (macos-latest) (push) Successful in 1m7s
Details
Build / Linux (ubuntu-24.04) (push) Successful in 1m14s
Details
Build / macOS (macos-latest) (pull_request) Successful in 53s
Details
Build / Linux (ubuntu-24.04) (pull_request) Successful in 1m19s
Details
Build / Windows (windows-latest) (push) Failing after 3m25s
Details
Build / Windows (windows-latest) (pull_request) Failing after 3m0s
Details

Browse Source 
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>
This commit is contained in:
shadowdao
2026-09-20 19:04:14 -07:00
co-authored by Claude Sonnet 5
parent 48a74e8c67
commit 3f2933ae3f
6 changed files with 573 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/src/session.cpp
+261 -11
View File
@@ -17,6 +17,7 @@ You may obtain a copy of the License at
#include <deque>
#include <exception>
#include <mutex>
#include <string>
#include <thread>
#include <utility>
#include <vector>
@@ -30,6 +31,7 @@ You may obtain a copy of the License at
#include <livekit/room_delegate.h>
#include <livekit/room_event_types.h>
#include <livekit/track.h>
#include <livekit/track_publication.h>
#include <livekit/video_frame.h>
#include <livekit/video_stream.h>
@@ -138,6 +140,13 @@ int &globalRefCount()
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
// ---------------------------------------------------------------------------
@@ -145,11 +154,12 @@ int &globalRefCount()
// ---------------------------------------------------------------------------
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 {
CommandType type;
std::shared_ptr<livekit::Track> track;
std::shared_ptr<livekit::RemoteTrackPublication> publication;
};
livekit::Room room;
@@ -157,10 +167,14 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
mutable std::mutex state_mutex;
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;
AudioFrameHandler on_audio;
SessionStateHandler on_state;
DiagnosticHandler on_diagnostic;
std::atomic<std::uint64_t> video_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:
// 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
// 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::condition_variable queue_cv;
std::deque<Command> queue;
@@ -183,6 +200,26 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
std::shared_ptr<livekit::AudioStream> audio_stream;
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;
~Impl() override = default;
@@ -207,17 +244,29 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
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);
if (!worker_running)
return;
queue.push_back(Command{type, std::move(track)});
queue.push_back(Command{type, std::move(track), std::move(publication)});
}
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
// (a fresh subscription) and attachExistingTracks (one already up when
// 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.
}
}
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 ------------------------------------------------------
@@ -304,6 +388,16 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
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
{
mutateState([](SessionStateMachine &m) { m.onReconnecting(); });
@@ -359,7 +453,7 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
void workerLoop()
{
for (;;) {
Command command{CommandType::Stop, nullptr};
Command command{CommandType::Stop, nullptr, nullptr};
{
std::unique_lock<std::mutex> lock(queue_mutex);
queue_cv.wait(lock, [this] { return !queue.empty(); });
@@ -369,7 +463,7 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
switch (command.type) {
case CommandType::AttachVideo:
attachVideo(command.track);
attachVideo(command.track, command.publication);
break;
case CommandType::DetachVideo:
detachVideo();
@@ -380,6 +474,9 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
case CommandType::DetachAudio:
detachAudio();
break;
case CommandType::RecoverVideo:
recoverVideo();
break;
case CommandType::Stop:
detachVideo();
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)
return;
@@ -413,6 +606,18 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
if (!stream)
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_thread = std::thread([this, stream] { videoReaderLoop(stream); });
mutateState([](SessionStateMachine &m) { m.onVideoAttached(); });
@@ -426,6 +631,20 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
video_thread.join();
const bool had = static_cast<bool>(video_stream);
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)
mutateState([](SessionStateMachine &m) { m.onVideoDetached(); });
}
@@ -548,6 +767,23 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate {
out.plane_count = count;
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);
}
@@ -633,6 +869,12 @@ void LiveKitSession::setStateHandler(SessionStateHandler 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)
{
if (impl_->connected)
@@ -652,6 +894,10 @@ bool LiveKitSession::connect(const SessionConfig &config)
}
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;
// 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) {
ok = false;
impl_->mutateState([&](SessionStateMachine &m) { m.onConnectFailed(e.what()); });
impl_->stopWatchdog();
impl_->stopWorker();
impl_->room.setDelegate(nullptr);
return false;
@@ -689,6 +936,7 @@ bool LiveKitSession::connect(const SessionConfig &config)
impl_->mutateState([](SessionStateMachine &m) {
m.onConnectFailed("could not connect to LiveKit (check the server URL, or the token may have expired)");
});
impl_->stopWatchdog();
impl_->stopWorker();
impl_->room.setDelegate(nullptr);
return false;
@@ -705,9 +953,11 @@ void LiveKitSession::disconnect()
if (!impl_)
return;
// Order matters: stop the readers first so nothing is mid-read on a
// stream the room is about to tear down, then disconnect the room, then
// drop the delegate so no event can arrive at a half-destroyed object.
// Order matters: stop the watchdog and the readers first so nothing is
// mid-read (or about to post a recovery command) on a stream the room
// 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();
if (impl_->connected) {