From 3b66e46afc3094bbb997db8de25ed179f0125ebb Mon Sep 17 00:00:00 2001 From: soulhackzlol Date: Sat, 20 Jun 2026 18:31:20 +0200 Subject: [PATCH 1/3] Fix VOD-audio session races (v0.1.8) The egress supervisor wakes every ~2s and, for a Twitch destination with VOD audio enabled, asks Twitch's API to allocate an IVS session and then points egress at the returned URL. The old code spawned that API call on every tick until one came back. When Twitch took longer than 2s to answer, the calls piled up: each allocated a different IVS session and each changed the override URL, which restarted egress. A single stream then showed up as several short sessions in Twitch Inspector, and the first one came up Source-Only. Two guards, both per-destination so multistreaming is unaffected: - Single-flight latch (try_claim_vod_fetch): only one fetch runs at a time. The claim re-checks the override under its mutex, so a fetch that just landed can't let a duplicate slip through. - Session epoch: bumped on every publisher disconnect, under the same mutex that holds the override. A fetch records the epoch when it starts and only writes its result if the epoch still matches. A request that returns after OBS disconnected, with an IVS token now bound to a dead session, is discarded instead of poisoning the next stream. Non-Twitch destinations never enter this branch. The synchronous /obs/multitrack-config proxy is untouched: it runs inside OBS's blocking request before the session starts, so it was never exposed to either race. Also syncs Cargo.lock to the package version (was left at 0.1.6). --- CHANGELOG.md | 39 +++++++- Cargo.lock | 2 +- Cargo.toml | 2 +- src/controller.rs | 237 +++++++++++++++++++++++++++++++++++++++++++++- src/main.rs | 74 +++++++++++---- 5 files changed, 329 insertions(+), 25 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f075d7c..19547db 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,42 @@ All notable changes will land here. Format loosely follows Nothing yet. +## [0.1.8] - VOD-audio session race fix + +**VOD audio could go live Source-Only with duplicate sessions.** On a +stream with VOD audio enabled, Twitch Inspector sometimes showed several +short sessions for a single go-live, the first one Transmuxed +(Source-Only) instead of Transcoded, and viewers hit a playback error. + +Root cause: the egress supervisor wakes every ~2s and, for a Twitch +destination with VOD audio on, asks Twitch's API to allocate an IVS +session, then points egress at the URL it returns. The old code fired a +fresh GetClientConfiguration call on every tick until one came back. When +Twitch answered slower than the 2s tick, the calls stacked. Each one +allocated its own IVS session and rewrote the destination's override URL, +and each rewrite restarted egress, so a single OBS stream reached Twitch +as a string of brief sessions. Whichever session won the race decided +whether the stream came up Transcoded or Source-Only. + +Two guards, both per-destination so multistreaming is unaffected: + +- A single-flight latch keeps exactly one session fetch in flight at a + time. It re-checks the override under its lock, so a fetch that just + finished can't let a duplicate slip in right behind it. + +- A session epoch, bumped on every publisher disconnect, records which + publisher session a fetch belongs to. A request that returns after OBS + disconnected, whose IVS token is now bound to a dead session, is + discarded instead of being written into the next stream. The apply check + and the disconnect both touch the override under the same lock, so they + cannot interleave. + +The `/obs/multitrack-config` proxy path (Enhanced Broadcasting via the +registered OBS service, and the experimental VOD+EB Launch button) was +never affected: it runs inside OBS's blocking config request, before the +stream starts, so it always sets the override for the session about to +begin. Non-Twitch destinations do not use this path at all. + ## [0.1.7] - Tour + VOD audio UX fixes **App tour overlay step updated.** The tour bubble for the Overlay tab @@ -902,7 +938,8 @@ See `0.1.0` below for the full feature list. port pre-flight, and process / RSS sampler have Windows-specific paths. - No automated release pipeline yet. Build from source per the README. -[Unreleased]: https://github.com/Soulhackzlol/InstantClone/compare/v0.1.7...HEAD +[Unreleased]: https://github.com/Soulhackzlol/InstantClone/compare/v0.1.8...HEAD +[0.1.8]: https://github.com/Soulhackzlol/InstantClone/compare/v0.1.7...v0.1.8 [0.1.7]: https://github.com/Soulhackzlol/InstantClone/compare/v0.1.6...v0.1.7 [0.1.6]: https://github.com/Soulhackzlol/InstantClone/compare/v0.1.0...v0.1.6 [0.1.0]: https://github.com/Soulhackzlol/InstantClone/releases/tag/v0.1.0 diff --git a/Cargo.lock b/Cargo.lock index 482bffb..bd7dff6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -218,7 +218,7 @@ dependencies = [ [[package]] name = "instantclone" -version = "0.1.6" +version = "0.1.8" dependencies = [ "bytes", "flate2", diff --git a/Cargo.toml b/Cargo.toml index 7278b0a..914a41c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "instantclone" -version = "0.1.7" +version = "0.1.8" edition = "2021" authors = ["s1moscs"] license = "GPL-3.0-only" diff --git a/src/controller.rs b/src/controller.rs index a4e0d9b..219750c 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -171,6 +171,31 @@ pub struct DestinationState { /// publisher disconnect so the next normal stream goes back to /// the configured URL. pub eb_override_url: std::sync::Mutex>, + /// In-flight latch for the VOD-audio IVS session fetch. Invariant: + /// `true` for exactly as long as one `fetch_twitch_vod_session` task is + /// running, `false` otherwise. The supervisor fires every ~2 s; without + /// this latch every tick while `eb_override_url` is None would launch a + /// fresh Twitch API request, each allocating a *different* IVS session + /// and forcing an extra egress restart (the multi-session / Source-Only + /// symptom). Claimed via `try_claim_vod_fetch`; the fetch task clears it + /// when it finishes, success or failure. Deliberately decoupled from + /// `eb_override_url`'s lifecycle - the override may be cleared from + /// several places (publisher disconnect, the multitrack-config proxy's + /// stale-override cleanup), and none of them need to know this latch + /// exists. The override being Some is what blocks a re-fetch; this latch + /// only prevents concurrent ones. + pub vod_fetch_pending: AtomicBool, + /// Generation counter for the publisher session this destination's + /// override belongs to. Bumped (under `eb_override_url`'s mutex) every + /// time the publisher disconnects. A VOD-session fetch can take up to + /// 15 s; if OBS disconnects while one is in flight, the IVS session it + /// returns is bound to the now-dead publisher session and pointing the + /// next stream at it would land on an endpoint with no live session + /// (the Source-Only failure). The fetch captures this epoch when it is + /// spawned and, at apply time, writes the override only if the epoch + /// still matches - both reads happen under the override mutex so the + /// check can't race the disconnect's clear+bump. + pub session_epoch: AtomicU64, } impl DestinationState { @@ -203,9 +228,70 @@ impl DestinationState { // (the destination_state lazy-init in particular). pass_through_multitrack_video: AtomicBool::new(false), pass_through_multitrack_audio: AtomicBool::new(false), + vod_fetch_pending: AtomicBool::new(false), + session_epoch: AtomicU64::new(0), } } + /// Try to claim the right to fetch this destination's VOD-audio IVS + /// session. Returns `true` for at most one caller per in-flight fetch. + /// + /// Two checks, in order: + /// 1. Atomically latch `vod_fetch_pending`. If it was already set, a + /// fetch is in flight - bail. + /// 2. Re-read `eb_override_url` *through its mutex* (the authoritative + /// source of truth for "do we have a session"). The supervisor reads + /// the override at the top of its loop, a moment before this claim; + /// a fetch that started on an earlier tick may have completed in that + /// gap. If a session now exists we release the latch and bail so we + /// never allocate a redundant one. + /// + /// The caller MUST clear `vod_fetch_pending` when the fetch finishes + /// (both on success and failure) to preserve the latch's invariant. + pub fn try_claim_vod_fetch(&self) -> bool { + if self.vod_fetch_pending.swap(true, Ordering::Relaxed) { + return false; + } + if self.eb_override_url.lock().unwrap().is_some() { + self.vod_fetch_pending.store(false, Ordering::Relaxed); + return false; + } + true + } + + /// Read the current session epoch. Capture this when spawning a VOD + /// session fetch and pass it back to `apply_vod_session_if_current`. + pub fn session_epoch(&self) -> u64 { + self.session_epoch.load(Ordering::Relaxed) + } + + /// Apply a freshly-fetched VOD-audio IVS session URL, but only if the + /// publisher session that requested it is still the current one. + /// `captured_epoch` is `session_epoch()` read when the fetch was spawned. + /// If a disconnect has since bumped the epoch, the fetched session is + /// bound to a dead publisher session - writing it would point the next + /// stream at a stale IVS endpoint - so we discard it and return false. + /// The epoch is read under the override mutex, so it cannot race + /// `invalidate_session_override`'s clear+bump. + pub fn apply_vod_session_if_current(&self, url: String, captured_epoch: u64) -> bool { + let mut guard = self.eb_override_url.lock().unwrap(); + if self.session_epoch.load(Ordering::Relaxed) != captured_epoch { + return false; + } + *guard = Some(url); + true + } + + /// Invalidate this destination's VOD/EB override on publisher + /// disconnect: clear the URL and bump the session epoch in one locked + /// section, so a fetch still in flight discards its (now-stale) result + /// rather than writing it into the next session. + pub fn invalidate_session_override(&self) { + let mut guard = self.eb_override_url.lock().unwrap(); + *guard = None; + self.session_epoch.fetch_add(1, Ordering::Relaxed); + } + fn note_outbound_bytes(&self, n: usize) { let now = process_now_ms(); let total = self @@ -1005,7 +1091,18 @@ impl Controller { // /obs/multitrack-config proxy sets a fresh override on // every new EB session anyway. for (_id, state) in self.all_destination_states() { - *state.eb_override_url.lock().unwrap() = None; + // Clear the override AND bump the session epoch atomically, + // so a VOD-session fetch still in flight discards its stale + // result instead of writing a dead-session IVS URL into the + // next stream (the late-completion race). + state.invalidate_session_override(); + // Note: we deliberately do NOT touch `vod_fetch_pending` + // here. It's owned solely by the fetch lifecycle (claim sets + // it, the task clears it on completion within ~15 s). If a + // fetch is in flight across a disconnect/reconnect, leaving + // the latch set is what stops a second concurrent fetch from + // being spawned for the same destination - clearing it here + // would reintroduce the multi-session bug on a fast restart. } self.log("ingest: publisher disconnected"); self.fire_webhook("⚠️", "OBS publisher disconnected."); @@ -2418,6 +2515,144 @@ mod tests { } } + /// `try_claim_vod_fetch` is single-flight: exactly one caller wins + /// while a fetch is in flight. The supervisor fires every ~2 s; before + /// this guard every tick spawned a fresh Twitch API call, each + /// allocating a distinct IVS session and forcing an extra egress + /// restart - the multi-session, wrong-broadcast-type symptom seen in + /// Twitch Inspector. The latch is released by the fetch task on + /// completion (modelled here by the explicit store), after which the + /// next tick may claim again (e.g. to retry a failed fetch). + #[test] + fn try_claim_vod_fetch_admits_one_claimant() { + let s = DestinationState::new("main".into()); + assert!(s.try_claim_vod_fetch(), "first caller must win the claim"); + for tick in 0..5 { + assert!( + !s.try_claim_vod_fetch(), + "tick {tick} must see a fetch already in flight" + ); + } + // Fetch task finished (success or failure) - latch released. + s.vod_fetch_pending.store(false, Ordering::Relaxed); + assert!( + s.try_claim_vod_fetch(), + "after the fetch ends the next tick must be able to claim" + ); + } + + /// The claim re-checks `eb_override_url` under its mutex before + /// committing: a fetch that completed on an earlier tick (setting the + /// override) must abort a redundant claim AND leave the latch clear, + /// so the destination isn't left falsely "fetching" forever. + #[test] + fn try_claim_vod_fetch_skips_when_session_already_allocated() { + let s = DestinationState::new("main".into()); + *s.eb_override_url.lock().unwrap() = Some("rtmps://ivs/session".into()); + assert!( + !s.try_claim_vod_fetch(), + "must not claim when a session already exists" + ); + assert!( + !s.vod_fetch_pending.load(Ordering::Relaxed), + "a skipped claim must release the latch, not leave it stuck" + ); + } + + /// Regression guard for the lockout hole: the latch is decoupled from + /// the override's lifecycle. The multitrack-config proxy (web.rs) and + /// publisher disconnect both clear `eb_override_url` without touching + /// the latch. After a successful fetch (override set, latch clear), + /// clearing the override - as those paths do - must let the next tick + /// re-claim and fetch a fresh session, not lock the destination into + /// the legacy Source-Only ingest forever. + #[test] + fn try_claim_vod_fetch_reclaims_after_override_cleared() { + let s = DestinationState::new("main".into()); + // Post-success state: session allocated, latch released by the task. + *s.eb_override_url.lock().unwrap() = Some("rtmps://ivs/session".into()); + s.vod_fetch_pending.store(false, Ordering::Relaxed); + assert!( + !s.try_claim_vod_fetch(), + "a live session must block a re-fetch" + ); + // Override cleared elsewhere (proxy cleanup / disconnect). + *s.eb_override_url.lock().unwrap() = None; + assert!( + s.try_claim_vod_fetch(), + "cleared override must allow a fresh fetch - no permanent lockout" + ); + } + + /// A fetch that returns while still in its own session applies normally. + #[test] + fn apply_vod_session_if_current_writes_when_session_unchanged() { + let s = DestinationState::new("main".into()); + let epoch = s.session_epoch(); + assert!( + s.apply_vod_session_if_current("rtmps://ivs/live".into(), epoch), + "same-session fetch must apply" + ); + assert_eq!( + *s.eb_override_url.lock().unwrap(), + Some("rtmps://ivs/live".into()) + ); + } + + /// Regression guard for the late-completion race: a VOD-session fetch + /// spawned in one publisher session must NOT write its IVS URL if OBS + /// disconnected (and bumped the epoch) while the request was in flight. + /// Writing it would point the next stream at a dead session - the + /// Source-Only failure this whole path exists to avoid. + #[test] + fn apply_vod_session_if_current_discards_after_disconnect() { + let s = DestinationState::new("main".into()); + let epoch = s.session_epoch(); // captured when the fetch is spawned + s.invalidate_session_override(); // OBS disconnects mid-fetch + assert!( + !s.apply_vod_session_if_current("rtmps://ivs/stale".into(), epoch), + "a fetch outliving its session must be discarded" + ); + assert!( + s.eb_override_url.lock().unwrap().is_none(), + "the stale URL must not leak into the next session" + ); + } + + /// `invalidate_session_override` clears the URL and advances the epoch + /// together, so the disconnect both forgets the old session and trips + /// any in-flight fetch's apply-time guard. + #[test] + fn invalidate_session_override_clears_url_and_bumps_epoch() { + let s = DestinationState::new("main".into()); + *s.eb_override_url.lock().unwrap() = Some("rtmps://ivs/old".into()); + let before = s.session_epoch(); + s.invalidate_session_override(); + assert!( + s.eb_override_url.lock().unwrap().is_none(), + "url must clear" + ); + assert_eq!(s.session_epoch(), before + 1, "epoch must advance"); + } + + /// Across a full disconnect/reconnect, a fetch from the NEW session + /// still applies: the epoch the supervisor captures after reconnect + /// matches the current one, so only the pre-disconnect fetch is stale. + #[test] + fn apply_vod_session_if_current_applies_for_fresh_session_after_reconnect() { + let s = DestinationState::new("main".into()); + s.invalidate_session_override(); // disconnect bumps epoch + let fresh_epoch = s.session_epoch(); // supervisor re-captures post-reconnect + assert!( + s.apply_vod_session_if_current("rtmps://ivs/new".into(), fresh_epoch), + "a fetch from the new session must apply" + ); + assert_eq!( + *s.eb_override_url.lock().unwrap(), + Some("rtmps://ivs/new".into()) + ); + } + /// `note_multitrack_video` is sticky-with-decay: once set, the /// chip stays lit for a short window after the last multi-track /// tag arrived (so a momentary pause between IDRs doesn't drop diff --git a/src/main.rs b/src/main.rs index 6042fd1..d7b4b07 100644 --- a/src/main.rs +++ b/src/main.rs @@ -412,29 +412,61 @@ async fn supervise_egress(mut rx: watch::Receiver, ctrl: Arc { - *state.eb_override_url.lock().unwrap() = Some(url); - ctrl_for_log.log(format!( - "[{}] VOD-audio session allocated by Twitch{}", - dest_id, - if want_eb { " (with EB ladder)" } else { "" } - )); + // Single-flight: only one session fetch runs at a time. + // Without this the supervisor would spawn a new API call + // every ~2 s while no override is set, each allocating a + // different IVS session and each triggering its own egress + // restart. try_claim_vod_fetch also re-checks the override + // under its mutex, so a fetch that just completed on the + // previous tick can't race a redundant second one in. + if state.try_claim_vod_fetch() { + let key = dest.stream_key.clone(); + let want_eb = dest.vod_audio_inject_eb; + let ctrl_for_log = ctrl.clone(); + let dest_id = dest.id.clone(); + // Capture the session this fetch belongs to. If OBS + // disconnects before it returns, the epoch moves on and + // apply_vod_session_if_current drops the stale result. + let epoch = state.session_epoch(); + tokio::spawn(async move { + match web::fetch_twitch_vod_session(key, want_eb).await { + Some(url) => { + // apply_vod_session_if_current publishes the + // URL only if we're still in the session that + // asked for it. A tick that then observes the + // latch free re-checks the override (now Some) + // and skips, so it can't spawn a duplicate. + if state.apply_vod_session_if_current(url, epoch) { + ctrl_for_log.log(format!( + "[{}] VOD-audio session allocated by Twitch{}", + dest_id, + if want_eb { " (with EB ladder)" } else { "" } + )); + } else { + ctrl_for_log.log(format!( + "[{}] VOD-audio: discarded stale session - publisher \ + reconnected while the request was in flight", + dest_id + )); + } + } + None => { + ctrl_for_log.log(format!( + "[{}] VOD-audio: Twitch API call failed - retrying on next tick", + dest_id + )); + } } - None => { - ctrl_for_log.log(format!( - "[{}] VOD-audio: Twitch API call failed - retrying on next tick", - dest_id - )); - } - } - }); + // Latch invariant: clear it the moment the fetch ends, + // whatever the outcome. On a successful apply the + // override now blocks re-fetch; otherwise the next + // tick is free to retry. + state + .vod_fetch_pending + .store(false, std::sync::atomic::Ordering::Relaxed); + }); + } } let effective_url = override_url.as_deref().unwrap_or(url.as_str()).to_string(); let url = &effective_url; From cc83893ea639b1a9cad5473ebb018e09f25bad81 Mon Sep 17 00:00:00 2001 From: soulhackzlol Date: Sun, 21 Jun 2026 11:35:02 +0200 Subject: [PATCH 2/3] Move VOD fetch completion onto DestinationState + concurrency tests complete_vod_fetch applies the session result (epoch-guarded) and releases the in-flight latch in one place, so the whole latch lifecycle (claim in try_claim_vod_fetch, release here) lives on the type that owns it instead of being split into the supervisor closure. The closure is now just fetch, complete, log. Tests: release-on-every-path for complete_vod_fetch (success, stale discard, failure), single-flight under real thread contention, and an apply-vs-disconnect stress test that asserts the override is never left stale regardless of which thread wins. --- src/controller.rs | 159 ++++++++++++++++++++++++++++++++++++++++++++++ src/main.rs | 51 ++++++--------- 2 files changed, 179 insertions(+), 31 deletions(-) diff --git a/src/controller.rs b/src/controller.rs index 219750c..dd75a63 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -198,6 +198,19 @@ pub struct DestinationState { pub session_epoch: AtomicU64, } +/// Outcome of finishing a VOD-session fetch, returned by +/// `complete_vod_fetch` so the supervisor can log it without re-deriving +/// what happened. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum VodFetchOutcome { + /// Session applied - egress will restart onto the IVS endpoint. + Applied, + /// Result dropped: the publisher session it was for ended mid-fetch. + DiscardedStale, + /// Twitch's API returned nothing; the next supervisor tick retries. + Failed, +} + impl DestinationState { pub fn new(id: String) -> Self { Self { @@ -292,6 +305,32 @@ impl DestinationState { self.session_epoch.fetch_add(1, Ordering::Relaxed); } + /// Finish a VOD-session fetch: apply the result if our publisher + /// session is still current, then release the in-flight latch no + /// matter what. Keeping the release here - not in the supervisor + /// closure - puts the whole latch lifecycle (claim in + /// `try_claim_vod_fetch`, release here) on one type, so a fetch can + /// never leave the latch stuck. Returns the outcome for the caller + /// to log. + pub fn complete_vod_fetch( + &self, + result: Option, + captured_epoch: u64, + ) -> VodFetchOutcome { + let outcome = match result { + Some(url) => { + if self.apply_vod_session_if_current(url, captured_epoch) { + VodFetchOutcome::Applied + } else { + VodFetchOutcome::DiscardedStale + } + } + None => VodFetchOutcome::Failed, + }; + self.vod_fetch_pending.store(false, Ordering::Relaxed); + outcome + } + fn note_outbound_bytes(&self, n: usize) { let now = process_now_ms(); let total = self @@ -2584,6 +2623,33 @@ mod tests { ); } + /// The single-flight guarantee under real thread contention: when a + /// swarm of threads races to claim the same destination (the situation + /// the atomic swap exists for), exactly one wins. A sequential test + /// can't prove this - it's the concurrent claim that the supervisor's + /// every-2s wake-up plus an in-flight fetch can produce. Deterministic + /// (no sleeps): the atomic swap has exactly one false -> true edge, so + /// the winner count is always 1 regardless of scheduling. + #[test] + fn try_claim_vod_fetch_admits_exactly_one_under_contention() { + let s = DestinationState::new("main".into()); + let winners = std::sync::atomic::AtomicUsize::new(0); + std::thread::scope(|scope| { + for _ in 0..32 { + scope.spawn(|| { + if s.try_claim_vod_fetch() { + winners.fetch_add(1, Ordering::Relaxed); + } + }); + } + }); + assert_eq!( + winners.load(Ordering::Relaxed), + 1, + "exactly one racing claimant may hold the single-flight latch" + ); + } + /// A fetch that returns while still in its own session applies normally. #[test] fn apply_vod_session_if_current_writes_when_session_unchanged() { @@ -2653,6 +2719,99 @@ mod tests { ); } + /// `complete_vod_fetch` on success applies the URL, reports Applied, + /// and releases the latch so the override (now Some) is what blocks + /// any re-fetch. + #[test] + fn complete_vod_fetch_applies_and_releases_on_success() { + let s = DestinationState::new("main".into()); + assert!(s.try_claim_vod_fetch(), "latch held, as in production"); + let epoch = s.session_epoch(); + assert_eq!( + s.complete_vod_fetch(Some("rtmps://ivs/live".into()), epoch), + VodFetchOutcome::Applied + ); + assert_eq!( + *s.eb_override_url.lock().unwrap(), + Some("rtmps://ivs/live".into()) + ); + assert!( + !s.vod_fetch_pending.load(Ordering::Relaxed), + "latch must be released after a successful completion" + ); + } + + /// A result that outlived its session reports DiscardedStale, writes + /// nothing, and STILL releases the latch (the previously untested + /// release-on-every-path invariant). + #[test] + fn complete_vod_fetch_discards_stale_and_releases() { + let s = DestinationState::new("main".into()); + assert!(s.try_claim_vod_fetch()); + let epoch = s.session_epoch(); + s.invalidate_session_override(); // disconnect mid-fetch + assert_eq!( + s.complete_vod_fetch(Some("rtmps://ivs/stale".into()), epoch), + VodFetchOutcome::DiscardedStale + ); + assert!(s.eb_override_url.lock().unwrap().is_none()); + assert!( + !s.vod_fetch_pending.load(Ordering::Relaxed), + "latch must be released even when the result is discarded" + ); + } + + /// A failed fetch reports Failed, leaves no override, and releases the + /// latch so the next supervisor tick can retry. + #[test] + fn complete_vod_fetch_releases_latch_on_failure() { + let s = DestinationState::new("main".into()); + assert!(s.try_claim_vod_fetch()); + let epoch = s.session_epoch(); + assert_eq!(s.complete_vod_fetch(None, epoch), VodFetchOutcome::Failed); + assert!(s.eb_override_url.lock().unwrap().is_none()); + assert!( + !s.vod_fetch_pending.load(Ordering::Relaxed), + "latch must be released on failure" + ); + assert!( + s.try_claim_vod_fetch(), + "a released latch lets the next tick retry" + ); + } + + /// Stress the apply-vs-disconnect race: a late fetch's apply and the + /// disconnect that should invalidate it run on two threads. Both take + /// the override mutex for their whole body, so the two orderings are + /// the only possibilities and both MUST end with no override - either + /// the disconnect clears the just-written URL, or it bumps the epoch + /// first so apply discards. A stale Some surviving here would be the + /// Source-Only bug. A barrier collides the critical sections; the + /// invariant holds every iteration regardless of who wins. + #[test] + fn apply_and_invalidate_never_leave_a_stale_override() { + use std::sync::Barrier; + for _ in 0..200 { + let s = DestinationState::new("main".into()); + let epoch = s.session_epoch(); + let barrier = Barrier::new(2); + std::thread::scope(|scope| { + scope.spawn(|| { + barrier.wait(); + s.apply_vod_session_if_current("rtmps://ivs/late".into(), epoch); + }); + scope.spawn(|| { + barrier.wait(); + s.invalidate_session_override(); + }); + }); + assert!( + s.eb_override_url.lock().unwrap().is_none(), + "racing apply against the disconnect must never leave a stale override" + ); + } + } + /// `note_multitrack_video` is sticky-with-decay: once set, the /// chip stays lit for a short window after the last multi-track /// tag arrived (so a momentary pause between IDRs doesn't drop diff --git a/src/main.rs b/src/main.rs index d7b4b07..e59cbdc 100644 --- a/src/main.rs +++ b/src/main.rs @@ -430,41 +430,30 @@ async fn supervise_egress(mut rx: watch::Receiver, ctrl: Arc { - // apply_vod_session_if_current publishes the - // URL only if we're still in the session that - // asked for it. A tick that then observes the - // latch free re-checks the override (now Some) - // and skips, so it can't spawn a duplicate. - if state.apply_vod_session_if_current(url, epoch) { - ctrl_for_log.log(format!( - "[{}] VOD-audio session allocated by Twitch{}", - dest_id, - if want_eb { " (with EB ladder)" } else { "" } - )); - } else { - ctrl_for_log.log(format!( - "[{}] VOD-audio: discarded stale session - publisher \ - reconnected while the request was in flight", - dest_id - )); - } - } - None => { + let result = web::fetch_twitch_vod_session(key, want_eb).await; + // complete_vod_fetch applies the URL only if we're + // still in the session that asked for it, then releases + // the latch whatever the outcome. A tick that then + // observes the latch free re-checks the override (now + // Some on success) and skips, so no duplicate spawns. + match state.complete_vod_fetch(result, epoch) { + controller::VodFetchOutcome::Applied => ctrl_for_log.log(format!( + "[{}] VOD-audio session allocated by Twitch{}", + dest_id, + if want_eb { " (with EB ladder)" } else { "" } + )), + controller::VodFetchOutcome::DiscardedStale => { ctrl_for_log.log(format!( - "[{}] VOD-audio: Twitch API call failed - retrying on next tick", + "[{}] VOD-audio: discarded stale session - publisher \ + reconnected while the request was in flight", dest_id - )); + )) } + controller::VodFetchOutcome::Failed => ctrl_for_log.log(format!( + "[{}] VOD-audio: Twitch API call failed - retrying on next tick", + dest_id + )), } - // Latch invariant: clear it the moment the fetch ends, - // whatever the outcome. On a successful apply the - // override now blocks re-fetch; otherwise the next - // tick is free to retry. - state - .vod_fetch_pending - .store(false, std::sync::atomic::Ordering::Relaxed); }); } } From 650508d7372a9d7d4417582ca70195250be8f8c5 Mon Sep 17 00:00:00 2001 From: soulhackzlol Date: Sun, 21 Jun 2026 11:35:02 +0200 Subject: [PATCH 3/3] docs: transcode reality, long-session validation, test count - Transcoded ladder is only guaranteed for Twitch Partners; Affiliates get it opportunistically and far less so above ~6 Mbps. Note that this is Twitch's allocation behaviour, not the proxy (OBS's native Twitch preset auto-caps bitrate, the Custom-server path InstantClone uses does not). - Longest validated run is ~5h, ~10h cumulative across sessions. - Reframe sync ring-append disk I/O as a deliberate trade-off (page cache is the async buffer) rather than a planned fix. - Bump test count 185 -> 197. --- README.md | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 6275c25..6fe7c3b 100644 --- a/README.md +++ b/README.md @@ -284,7 +284,7 @@ No npm. No submodules. No platform SDKs. The dashboard HTML is minified + gzippe ## Status -**Daily-driver ready on Windows.** I use it on my own streams. CI runs fmt + clippy (with `-D warnings`) + 185 tests on every push, and a tagged commit auto-builds + publishes a release artifact with a `SHA256SUMS.txt` checksum file alongside (no code-signing certificate yet, so the OS may warn on first launch). +**Daily-driver ready on Windows.** I use it on my own streams. CI runs fmt + clippy (with `-D warnings`) + 197 tests on every push, and a tagged commit auto-builds + publishes a release artifact with a `SHA256SUMS.txt` checksum file alongside (no code-signing certificate yet, so the OS may warn on first launch). **What's solid** @@ -303,8 +303,9 @@ No npm. No submodules. No platform SDKs. The dashboard HTML is minified + gzippe **What's rough, honestly** - **Windows only.** macOS / Linux aren't tested or packaged. Several modules (tray, port pre-flight, RSS sampler) have Windows-specific code paths that need parallel implementations. -- **Multi-hour streams unproven.** Longest validated real-world stream is ~30 minutes. The supervisor + keepalive + ack logic is designed to handle indefinite sessions but nobody has stress-tested an 8-hour run. -- **Sync disk I/O on the async hot path** for ring append. Page cache absorbs it at typical stream rates, but a flush stall could freeze other tasks. `spawn_blocking` is on the v0.2 list. +- **Multi-hour streams lightly validated.** Longest single run is ~5 hours, with roughly ~10 hours cumulative across test sessions. The supervisor + keepalive + ack logic is designed to handle indefinite sessions; an 8+ hour daily-driver run still wants more mileage. +- **Transcoded ladder isn't guaranteed without EB.** Only Twitch Partners get a transcode slot every time. Affiliates get one opportunistically (capacity-dependent, and far less likely above ~6 Mbps); everyone else stays Source-Only. Above ~8 Mbps under Source-Only, some viewers' hardware decoders fail with Error #1000. This is Twitch's allocation behaviour, not the proxy: OBS's native Twitch preset auto-caps bitrate to stay transcode-eligible, but the Custom-server path (which InstantClone has to be) skips that cap. For a guaranteed ladder use Enhanced Broadcasting (or the VOD+EB Launch button); otherwise keep bitrate near ~6000 Kbps. +- **Sync disk I/O on the ring-append hot path, by choice.** The buffered write lands in the OS page cache in microseconds and the kernel flushes in the background, so the page cache is already acting as the async buffer, and the index and the bytes advance under one lock so a reader never sees a tag whose bytes aren't on disk yet. The tail risk is a writeback stall under memory or slow-disk pressure, which on the current-thread runtime would briefly freeze egress too, not just ingest. Moving the write off-thread adds per-tag overhead and reopens a write-vs-index consistency window, so it isn't worth it for typical hardware. If low-end disks ever become a priority the real lever is the runtime (separate the disk-blocking ingest from egress), not async writes. A someday-maybe I'll revisit, not a blocker. - **A handful of `unwrap()` on lock guards.** Fine because `panic = "abort"` means a poison condition can't propagate, but still on the cleanup list. - **Hand-rolled HTTP server.** Smaller binary than `hyper`, but I now own the entire HTTP CVE surface. Worth re-evaluating if the surface grows.