From 8110d852e07b193ca75d8def0ecc484e403a9bcc Mon Sep 17 00:00:00 2001 From: Joe Blau Date: Sat, 4 Jul 2026 19:20:19 -0400 Subject: [PATCH] fix: audio-mix and stop/start races in both publishers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two pre-existing robustness gaps shared by RTMPPublisher and SessionPublisher, surfaced by the issue #20 resilience review. Fixed in both via the shared layer. 1. App-audio mix went permanently silent if the mic route was dead from go-live. HaishinKit renders the multitrack mix only when the MAIN track (mic, track 0) appends; the mic-stall failover only promoted app audio to the main track once the mic had appended at least once (`lastMicAppendAt > 0`), so a mic dead from the start never promoted and the whole mix — app audio included — stayed silent forever. MicStallEvaluator now measures mic silence from `startedAt` (go-live) when the mic has never appeared, so a dead-from-start route still promotes app audio once the grace window passes. The previously-working stall path is byte-for-byte unchanged (micReference == lastMicAppendAt when > 0). 2. stop() during start()'s pre-`isRunning` setup leaked long-lived tasks. Both publishers ran their setup awaits (factory/mixer setup + startRunning) before setting isRunning and spawning the path-supervisor / watchdog / frame-repeat tasks; a stop() interleaving there early-returned via `guard isRunning`, then start() resumed and spawned tasks it could never cancel (leaving a 2s watchdog timer waking forever and the mixer running). Added a `guard !userInitiatedStop` bail-out after mixer.startRunning in both, plus a second re-check in RTMP after the two status awaits (a narrower window Session doesn't have) so the supervisors/watchdog are never spawned post-teardown. RTMP's normal go-live path is unchanged (both guards fire only on a concurrent stop). Reviewed adversarially. StreamCore: 99 tests pass (+3 mic-stall cases); full app builds against the iOS 27 SDK. Co-Authored-By: Claude Opus 4.8 (1M context) --- StreamBroadcast/RTMPPublisher.swift | 18 +++++++- StreamBroadcast/SessionPublisher.swift | 14 ++++++- StreamCore/MicStallEvaluator.swift | 18 +++++--- StreamCoreTests/SharedSupervisionTests.swift | 44 +++++++++++++++++--- 4 files changed, 80 insertions(+), 14 deletions(-) diff --git a/StreamBroadcast/RTMPPublisher.swift b/StreamBroadcast/RTMPPublisher.swift index 69b5831..cef57ad 100644 --- a/StreamBroadcast/RTMPPublisher.swift +++ b/StreamBroadcast/RTMPPublisher.swift @@ -151,6 +151,9 @@ actor RTMPPublisher: Publisher { private var lastMicAppendAt: UInt64 = 0 private var lastAppAppendAt: UInt64 = 0 private var micTrackStalled = false + /// When the broadcast went live, so a mic route that is dead from the START (no + /// buffer ever) still trips the failover — its silence is measured from here. + private var startedAt: UInt64 = 0 /// Builds + starts the pipeline, connects, and begins publishing. The video /// size is NOT set here — it is locked from the first frame via `setOutputSize`. @@ -174,6 +177,11 @@ actor RTMPPublisher: Publisher { await mixer.startRunning() + // stop() may have interleaved during the setup awaits above (actor reentrancy) + // and early-returned via its `guard isRunning` branch before we set isRunning. + // Bail out cleanly rather than spawn long-lived tasks it could never cancel. + guard !userInitiatedStop else { await mixer.stopRunning(); return } + if settings.backupEnabled { streamLog.warning("Local backup disabled: a second real-time video encoder is not enabled") } @@ -182,6 +190,7 @@ actor RTMPPublisher: Publisher { // and publish failures therefore retry just like a mid-stream drop instead // of ending the user-owned capture session. isRunning = true + startedAt = DispatchTime.now().uptimeNanoseconds startAudioConsumers() startFrameRepeat() // 4th long-lived task, spawned BEFORE the first connect so path gating @@ -209,6 +218,10 @@ actor RTMPPublisher: Publisher { // attempts cannot be replayed as stale recovery events. let connectionStatuses = await connection.status let streamStatuses = await stream.status + // stop() can interleave during the two status awaits above (actor reentrancy) + // after passing its own `guard isRunning`; re-check so the supervisors and the + // watchdog timer aren't spawned — and then leaked, uncancellable — post-teardown. + guard isRunning, !userInitiatedStop else { return } connectionSupervisorTask = Task { [weak self] in await self?.superviseConnection(connectionStatuses) @@ -646,10 +659,11 @@ actor RTMPPublisher: Publisher { // quiet, promote track 1 to the mix clock; the first mic buffer back // flips it home (see appendMic). No timeline rebase on either side — // video/app kept flowing, so mic samples re-enter already aligned. - if settings.includeAppAudio, !micTrackStalled, lastMicAppendAt > 0, + if settings.includeAppAudio, !micTrackStalled, startedAt > 0, MicStallEvaluator.shouldPromoteApp(now: now, lastMicAppendAt: lastMicAppendAt, - lastAppAppendAt: lastAppAppendAt) { + lastAppAppendAt: lastAppAppendAt, + startedAt: startedAt) { micTrackStalled = true await applyAudioMixerSettings() streamLog.warning("Mic buffers stalled >4s; app audio is now the mix clock") diff --git a/StreamBroadcast/SessionPublisher.swift b/StreamBroadcast/SessionPublisher.swift index 1411e13..353db3a 100644 --- a/StreamBroadcast/SessionPublisher.swift +++ b/StreamBroadcast/SessionPublisher.swift @@ -100,6 +100,9 @@ actor SessionPublisher: Publisher { private var micTrackStalled = false private var lastMicAppendAt: UInt64 = 0 private var lastAppAppendAt: UInt64 = 0 + /// When the broadcast went live, so a mic route dead from the START (no buffer + /// ever) still trips the failover — its silence is measured from here. + private var startedAt: UInt64 = 0 /// Proactive network-path supervision (Wi-Fi <-> 5G handoffs, dead zones). private var currentPath: NetworkPathSnapshot? private var lastPathChangeAt: UInt64 = 0 @@ -178,7 +181,13 @@ actor SessionPublisher: Publisher { await mixer.setVideoMixerSettings(vm) await mixer.startRunning() + // stop() may have interleaved during the setup awaits above (actor reentrancy) + // and early-returned via its `guard isRunning` branch before we set isRunning. + // Bail out cleanly rather than spawn long-lived tasks it could never cancel. + guard !userInitiatedStop else { await mixer.stopRunning(); return } + isRunning = true + startedAt = DispatchTime.now().uptimeNanoseconds startAudioConsumers() startFrameRepeat() // Path supervision spawned BEFORE the first connect so path gating covers it. @@ -551,10 +560,11 @@ actor SessionPublisher: Publisher { guard now &- lastMediaAt < 10_000_000_000 else { watchdog.stalledTicks = 0; return } // Mic-stall failover: app audio flowing but the mic gone quiet -> promote // track 1 to the mix clock; the first mic buffer back flips it home. - if settings.includeAppAudio, !micTrackStalled, lastMicAppendAt > 0, + if settings.includeAppAudio, !micTrackStalled, startedAt > 0, MicStallEvaluator.shouldPromoteApp(now: now, lastMicAppendAt: lastMicAppendAt, - lastAppAppendAt: lastAppAppendAt) { + lastAppAppendAt: lastAppAppendAt, + startedAt: startedAt) { micTrackStalled = true await applyAudioMixerSettings() sessionLog.warning("Mic buffers stalled >4s; app audio is now the mix clock") diff --git a/StreamCore/MicStallEvaluator.swift b/StreamCore/MicStallEvaluator.swift index ae018ea..4263826 100644 --- a/StreamCore/MicStallEvaluator.swift +++ b/StreamCore/MicStallEvaluator.swift @@ -17,12 +17,20 @@ public enum MicStallEvaluator { public static let micStallThreshold: UInt64 = 4_000_000_000 // 4s /// Whether app audio should take over the mix clock. The caller pre-checks the - /// preconditions (`includeAppAudio`, not already stalled, `lastMicAppendAt > 0`) - /// and owns the mixer re-apply; this is only the timing test. Uses wrapping - /// subtraction to match the publisher's `uptimeNanoseconds` arithmetic. + /// preconditions (`includeAppAudio`, not already stalled, capture live) and owns + /// the mixer re-apply; this is only the timing test. Uses wrapping subtraction to + /// match the publisher's `uptimeNanoseconds` arithmetic. + /// + /// `startedAt` is when the broadcast went live. When the mic has NEVER produced a + /// buffer (`lastMicAppendAt == 0`) its silence is measured from go-live, so a mic + /// route that is dead from the start still promotes app audio — otherwise the main + /// (mic) track never appends and HaishinKit renders the WHOLE mix silent, dropping + /// the flowing app audio too. public static func shouldPromoteApp(now: UInt64, lastMicAppendAt: UInt64, - lastAppAppendAt: UInt64) -> Bool { - now &- lastAppAppendAt < appFreshWindow && now &- lastMicAppendAt > micStallThreshold + lastAppAppendAt: UInt64, + startedAt: UInt64) -> Bool { + let micReference = lastMicAppendAt > 0 ? lastMicAppendAt : startedAt + return now &- lastAppAppendAt < appFreshWindow && now &- micReference > micStallThreshold } } diff --git a/StreamCoreTests/SharedSupervisionTests.swift b/StreamCoreTests/SharedSupervisionTests.swift index 23766f6..a771610 100644 --- a/StreamCoreTests/SharedSupervisionTests.swift +++ b/StreamCoreTests/SharedSupervisionTests.swift @@ -173,25 +173,32 @@ import StreamCore } @Suite struct MicStallEvaluatorTests { + // startedAt is irrelevant once the mic has appended at least once (micReference + // = lastMicAppendAt); a fixed early value keeps these focused on the stall path. + private let started: UInt64 = 1_000_000_000 + @Test("App audio fresh + mic silent past 4s promotes app to the mix clock") func promotesWhenMicStalls() { #expect(MicStallEvaluator.shouldPromoteApp(now: 10_000_000_000, lastMicAppendAt: 5_000_000_000, // 5s ago - lastAppAppendAt: 9_000_000_000)) // 1s ago + lastAppAppendAt: 9_000_000_000, // 1s ago + startedAt: started)) } @Test("A mic quiet for under 4s does not promote") func micNotYetStalled() { #expect(!MicStallEvaluator.shouldPromoteApp(now: 10_000_000_000, lastMicAppendAt: 7_000_000_000, // 3s ago - lastAppAppendAt: 9_000_000_000)) + lastAppAppendAt: 9_000_000_000, + startedAt: started)) } @Test("Stale app audio does not promote (nothing to promote to)") func appNotFresh() { #expect(!MicStallEvaluator.shouldPromoteApp(now: 10_000_000_000, lastMicAppendAt: 5_000_000_000, - lastAppAppendAt: 7_000_000_000)) // 3s ago > 2s + lastAppAppendAt: 7_000_000_000, // 3s ago > 2s + startedAt: started)) } @Test("Thresholds are exclusive at the exact boundary") @@ -199,10 +206,37 @@ import StreamCore // micAge exactly 4s -> not stalled (strict >); appAge exactly 2s -> not fresh (strict <). #expect(!MicStallEvaluator.shouldPromoteApp(now: 10_000_000_000, lastMicAppendAt: 6_000_000_000, // exactly 4s - lastAppAppendAt: 9_000_000_000)) + lastAppAppendAt: 9_000_000_000, + startedAt: started)) #expect(!MicStallEvaluator.shouldPromoteApp(now: 10_000_000_000, lastMicAppendAt: 5_000_000_000, - lastAppAppendAt: 8_000_000_000)) // exactly 2s + lastAppAppendAt: 8_000_000_000, // exactly 2s + startedAt: started)) + } + + @Test("A mic dead from go-live still promotes app audio once past the grace window") + func promotesWhenMicNeverAppeared() { + // lastMicAppendAt == 0: silence is measured from startedAt (go-live). + #expect(MicStallEvaluator.shouldPromoteApp(now: 10_000_000_000, + lastMicAppendAt: 0, // never appeared + lastAppAppendAt: 9_000_000_000, // app flowing + startedAt: 1_000_000_000)) // 9s since go-live + } + + @Test("A mic dead from go-live does not promote within the grace window") + func withinGraceAfterGoLive() { + #expect(!MicStallEvaluator.shouldPromoteApp(now: 5_000_000_000, + lastMicAppendAt: 0, + lastAppAppendAt: 4_500_000_000, // app fresh + startedAt: 2_000_000_000)) // only 3s since go-live + } + + @Test("A mic dead from go-live with no app audio does not promote") + func neverAppearedButNoAppAudio() { + #expect(!MicStallEvaluator.shouldPromoteApp(now: 10_000_000_000, + lastMicAppendAt: 0, + lastAppAppendAt: 0, // app never flowed + startedAt: 1_000_000_000)) } }