diff --git a/core/include/stplugin/session.h b/core/include/stplugin/session.h index c2b6e5f..95e48ef 100644 --- a/core/include/stplugin/session.h +++ b/core/include/stplugin/session.h @@ -11,6 +11,7 @@ You may obtain a copy of the License at #pragma once +#include #include #include @@ -54,6 +55,22 @@ struct SessionConfig { 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. /// /// Threading contract, which the OBS adapter depends on: @@ -67,6 +84,11 @@ struct SessionConfig { /// call. /// - The state handler is invoked from whichever thread observed the /// 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 /// synchronised against a running session. class LiveKitSession { @@ -80,6 +102,7 @@ public: void setVideoHandler(VideoFrameHandler handler); void setAudioHandler(AudioFrameHandler handler); void setStateHandler(SessionStateHandler handler); + void setDiagnosticHandler(DiagnosticHandler handler); /// 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 diff --git a/core/include/stplugin/session_types.h b/core/include/stplugin/session_types.h index 1f2c5b3..e7c544d 100644 --- a/core/include/stplugin/session_types.h +++ b/core/include/stplugin/session_types.h @@ -19,6 +19,7 @@ You may obtain a copy of the License at // self-consistent -- all live here, and LiveKitSession is the (much thinner) // piece that wires real SDK callbacks into them. +#include #include #include #include @@ -137,6 +138,82 @@ private: 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 // --------------------------------------------------------------------------- @@ -177,4 +254,13 @@ using VideoFrameHandler = std::function; using AudioFrameHandler = std::function; using SessionStateHandler = std::function; +/// 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; + } // namespace stplugin diff --git a/core/src/session.cpp b/core/src/session.cpp index cce09d5..6d5fa43 100644 --- a/core/src/session.cpp +++ b/core/src/session.cpp @@ -17,6 +17,7 @@ You may obtain a copy of the License at #include #include #include +#include #include #include #include @@ -30,6 +31,7 @@ You may obtain a copy of the License at #include #include #include +#include #include #include @@ -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 track; + std::shared_ptr 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 video_frames{0}; std::atomic 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 queue; @@ -183,6 +200,26 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate { std::shared_ptr 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 current_video_track; + std::shared_ptr 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 track = nullptr) + void post(CommandType type, std::shared_ptr track = nullptr, + std::shared_ptr publication = nullptr) { { std::lock_guard 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 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 &publication, bool unmuted) + { + if (!publication || publication->kind() != livekit::TrackKind::KIND_VIDEO) + return; + std::string current_sid; + { + std::lock_guard 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 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 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 &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 guard(watchdog_mutex); + watchdog_running = true; + } + watchdog_thread = std::thread([this] { watchdogLoop(); }); + } + + void stopWatchdog() + { + { + std::lock_guard 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 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 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 publication; + std::shared_ptr track; + { + std::lock_guard 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 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 &track, + const std::shared_ptr &publication) { if (!track) return; @@ -413,6 +606,18 @@ struct LiveKitSession::Impl : public livekit::RoomDelegate { if (!stream) return; + { + std::lock_guard 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 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(video_stream); video_stream.reset(); + { + std::lock_guard 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 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 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 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) { diff --git a/core/src/session_types.cpp b/core/src/session_types.cpp index 998984f..b9d3a5c 100644 --- a/core/src/session_types.cpp +++ b/core/src/session_types.cpp @@ -11,6 +11,8 @@ You may obtain a copy of the License at #include "stplugin/session_types.h" +#include + namespace stplugin { const char *describePixelFormat(PixelFormat format) @@ -175,4 +177,62 @@ void SessionStateMachine::onAudioDetached() 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 diff --git a/core/tests/test_session.cpp b/core/tests/test_session.cpp index 536ff74..c11a0e0 100644 --- a/core/tests/test_session.cpp +++ b/core/tests/test_session.cpp @@ -14,8 +14,10 @@ You may obtain a copy of the License at // COVERED headlessly -- the session's own decision-making: the state // machine's transitions (including the publisher-swap and reconnect paths // that motivated this plugin), track selection, frame geometry validation, -// and the real connect() failure paths against the real SDK (bad URL, -// unreachable host, garbage token). +// the stall-recovery watchdog's timing/backoff decisions (StallWatchdog, +// 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: // a successful connect, actual subscription, and actual decoded frames @@ -209,6 +211,134 @@ void testFailureAndRecovery() 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) // --------------------------------------------------------------------------- @@ -335,6 +465,10 @@ int main() testReconnect(); testFailureAndRecovery(); + testStallWatchdogFiresAfterThreshold(); + testStallWatchdogDoesNotFireWhenNotExpectingFrames(); + testStallWatchdogBacksOffRatherThanLooping(); + LiveKitSession::globalInitialize(); testConnectRejectsIncompleteConfig(); testConnectToUnreachableServerFailsCleanly(); diff --git a/obs-adapter/src/plugin-main.cpp b/obs-adapter/src/plugin-main.cpp index f049d42..d05d4c9 100644 --- a/obs-adapter/src/plugin-main.cpp +++ b/obs-adapter/src/plugin-main.cpp @@ -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->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->setStatus(detail.empty() ? describeSessionState(state) : detail); self->status_is_error.store(state == SessionState::Failed);