fix: audio-mix and stop/start races in both publishers

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")

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")

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
     }
 }

StreamCoreTests/SharedSupervisionTests.swift

@@ -173,36 +173,70 @@ 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")
     func boundaries() {
         // 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))
     }
 }