From f2e8f3c4be15567fafbd050112dbbd2b23104e77 Mon Sep 17 00:00:00 2001 From: Andreas Busslinger Date: Sat, 8 Aug 2026 22:46:27 +0200 Subject: [PATCH] fix(video): give live muxerFailed pump exits a recovery arm MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A live pump death with muxerFailed hit a bare return in handlePumpFinished: no rebuild, no halt (shouldHaltLiveProduction assumed the AE#222 arm covers it), no host signal. The provider kept serving a frozen playlist, AVPlayer parked on it waiting for buffer that never came, and the session zombified until the viewer zapped away. Field trace: a reconnect join with a new encoder session forced a same-PID rotation whose first cut deferred awaiting an audio sample entry (bridged E-AC-3), and the resulting muxerFailed exit was swallowed. The AE#222 assumption was doubly false for live: its live paths returned without doing anything, and even the admitted rebuild went through requestRestart, which bails on the empty live segmentPlan — a second silent zombie path, now also routed into the new arm. handleLiveMuxerFailure rebuilds the producer IN PLACE on the same demuxer and provider at the live continuation point (finishLiveReopen minus the fresh connection — the socket is healthy, and a reopen would double-connect against single-connection portals because it opens the new connection before closing the old demuxer). Bounded by progress, not per session: new segments since the last death reset the budget, three consecutive barren deaths halt production (releasing blocking-reload waiters) and publish onLiveSourceReset so the host can retune. Co-Authored-By: Claude Fable 5 --- .../Video/HLSVideoEngine+LiveReopen.swift | 145 ++++++++++++++++-- .../AetherEngine/Video/HLSVideoEngine.swift | 7 + .../LiveProductionHaltTests.swift | 38 ++++- 3 files changed, 177 insertions(+), 13 deletions(-) diff --git a/Sources/AetherEngine/Video/HLSVideoEngine+LiveReopen.swift b/Sources/AetherEngine/Video/HLSVideoEngine+LiveReopen.swift index ffe8898d..dd1cbc2c 100644 --- a/Sources/AetherEngine/Video/HLSVideoEngine+LiveReopen.swift +++ b/Sources/AetherEngine/Video/HLSVideoEngine+LiveReopen.swift @@ -72,7 +72,9 @@ extension HLSVideoEngine { case .eof, .readError, .keyframeStarvation: return !sourceReopenable case .stopRequested, .muxerFailed, .backpressureWedge, .needsAudioSampleEntryPrime: - // AE#222 rebuilds into the same provider with a primed muxer, so production continues. + // muxerFailed / needsAudioSampleEntryPrime: handleLiveMuxerFailure and the AE#222 arm rebuild + // the producer into the SAME provider, so an eager halt here would 503 the very provider the + // rebuild is about to serve; both arms halt themselves once their budget is exhausted. return false } } @@ -95,6 +97,21 @@ extension HLSVideoEngine { return .none } + /// Live muxerFailed pure decision: the muxer died but the connection is healthy, so the arm rebuilds + /// the producer in place — bounded by PROGRESS, not per session: a rebuild that cut new segments since + /// the last death earns a fresh budget (an hours-long channel legitimately crosses several encoder + /// restarts), while `cap` consecutive barren deaths mean the source is not muxable as-is and the + /// session must halt production and delegate to host retune. + static func liveMuxerRebuildDecision( + continuationIndex: Int, + lastContinuationIndex: Int, + barrenCycles: Int, + cap: Int + ) -> (rebuild: Bool, newBarrenCycles: Int) { + let cycles = continuationIndex == lastContinuationIndex ? barrenCycles + 1 : 0 + return (cycles < cap, cycles) + } + func handlePumpFinished(_ prod: HLSSegmentProducer, reason: HLSSegmentProducer.PumpExitReason) { // #65 (VOD only): a broken backpressure wedge means AVPlayer is stuck behind a parked producer. @@ -111,12 +128,17 @@ extension HLSVideoEngine { handleAudioSampleEntryPrimeNeeded(prod) return } - // #99 failure mode B: a VOD muxer death (e.g. first cut before any bridged audio packet, so + // #99 failure mode B: a muxer death (e.g. first cut before any bridged audio packet, so // mov_write_moov cannot build the dec3 box) previously had NO recovery arm; the session sat - // starved forever. Bounded revive through the normal restart path, which rebuilds the muxer - // and re-arms (post-EOF: rebuilds) the audio bridge. - if case .muxerFailed = reason, !isLiveSession { - handleVODMuxerFailure() + // starved forever. VOD: bounded revive through the normal restart path, which rebuilds the muxer + // and re-arms (post-EOF: rebuilds) the audio bridge. Live: the restart path is VOD-only (empty + // segmentPlan), so the live arm rebuilds the producer in place on the same connection instead. + if case .muxerFailed = reason { + if isLiveSession { + handleLiveMuxerFailure(prod) + } else { + handleVODMuxerFailure() + } return } // #126: a VOD pump that dies on a read error having produced NOTHING (no packets @@ -168,7 +190,7 @@ extension HLSVideoEngine { } switch reason { case .stopRequested, .muxerFailed, .backpressureWedge, .needsAudioSampleEntryPrime: - // needsAudioSampleEntryPrime never reaches here (its arm above returns first). + // muxerFailed and needsAudioSampleEntryPrime never reach here (their arms above return first). return case .sourceReplay: // Server restarted stream from beginning (Jellyfin transcode respawn); URL reopen would replay stale content. Delegate to host for fresh negotiation. @@ -332,8 +354,11 @@ extension HLSVideoEngine { + "falling back to the muxerFailed recovery", category: .session ) - if isLiveSession { return } - handleVODMuxerFailure() + if isLiveSession { + handleLiveMuxerFailure(prod) + } else { + handleVODMuxerFailure() + } return } @@ -350,8 +375,26 @@ extension HLSVideoEngine { + "falling back to the muxerFailed recovery", category: .session ) - if isLiveSession { return } - handleVODMuxerFailure() + if isLiveSession { + handleLiveMuxerFailure(prod) + } else { + handleVODMuxerFailure() + } + return + } + + // Live: requestRestart is VOD-only (performRestart bails on the empty live segmentPlan, so the + // old code path silently rebuilt NOTHING and the session zombified). Rebuild in place instead; + // the stored session prime flows in through makeProducer. + if isLiveSession { + EngineLog.emit( + "[HLSVideoEngine] AE#222 rebuilding live producer in place with a \(prime.count) B " + + "audio moov prime (audio stream-copy preserved)", + category: .session + ) + Task.detached(priority: .userInitiated) { [weak self] in + self?.rebuildLiveProducerInPlace(failed: prod) + } return } @@ -400,6 +443,86 @@ extension HLSVideoEngine { requestRestart(at: idx, authoritative: true) } + /// Live arm for a pump death with `muxerFailed` — previously a bare return: the provider kept serving + /// a frozen playlist, AVPlayer parked on it, and the session zombified until the viewer zapped away. + /// The connection is healthy (the muxer died, not the socket), so the recovery rebuilds the producer + /// IN PLACE on the same demuxer and provider at the live continuation point. A reopen is the wrong + /// tool here: `performLiveReopen` opens its fresh connection BEFORE closing the old demuxer, which + /// double-connects against a healthy socket and breaks single-connection portals. + func handleLiveMuxerFailure(_ prod: HLSSegmentProducer) { + restartLock.lock() + let nextNow = provider?.liveContinuationPoint().nextIndex ?? 0 + let decision = Self.liveMuxerRebuildDecision( + continuationIndex: nextNow, + lastContinuationIndex: lastMuxerRebuildSegmentCount, + barrenCycles: liveMuxerRebuildCycles, + cap: Self.maxLiveMuxerRebuildCycles + ) + liveMuxerRebuildCycles = decision.newBarrenCycles + lastMuxerRebuildSegmentCount = nextNow + restartLock.unlock() + + guard decision.rebuild else { + EngineLog.emit( + "[HLSVideoEngine] live muxer died \(decision.newBarrenCycles) times with no new segment; " + + "halting production and requesting host retune", + category: .session + ) + provider?.markLiveProductionHalted() + onLiveSourceReset?() + return + } + EngineLog.emit( + "[HLSVideoEngine] live pump died with muxerFailed; rebuilding producer in place at " + + "seg\(nextNow) (barren cycle \(decision.newBarrenCycles)/\(Self.maxLiveMuxerRebuildCycles))", + category: .session + ) + // handlePumpFinished runs on the dying pump thread; hop off it like the reopen path does. + Task.detached(priority: .userInitiated) { [weak self] in + self?.rebuildLiveProducerInPlace(failed: prod) + } + } + + /// `finishLiveReopen` minus the fresh connection: same demuxer, same provider, produced timeline + /// continues at the live continuation point behind #EXT-X-DISCONTINUITY (the dead pump dropped its + /// pending look-behind packets; the join gate re-syncs at the next IDR). `sessionAudioMoovPrimeFrame` + /// flows in through `makeProducer`. + func rebuildLiveProducerInPlace(failed: HLSSegmentProducer) { + restartLock.lock() + guard producer === failed, let prov = provider else { + restartLock.unlock() + return + } + let (nextIndex, outputEnd) = prov.liveContinuationPoint() + do { + let newProd = try makeProducer( + baseIndex: nextIndex, + liveReopenOutputEndSeconds: outputEnd + ) + newProd.firstSegmentDiscontinuous = true + newProd.onVideoShiftKnown = { [weak self] shiftPts, _ in + self?.handleLiveTimelineRebase(shiftPts, seamOutputSeconds: outputEnd) + } + producer = newProd + restartLock.unlock() + newProd.start() + EngineLog.emit( + "[HLSVideoEngine] live producer rebuilt in place: continuing at seg\(nextIndex) " + + "(outputEnd=\(String(format: "%.1f", outputEnd))s)", + category: .session + ) + } catch { + restartLock.unlock() + EngineLog.emit( + "[HLSVideoEngine] live in-place producer rebuild failed (\(error)); " + + "halting production and requesting host retune", + category: .session + ) + provider?.markLiveProductionHalted() + onLiveSourceReset?() + } + } + /// #65: re-base the producer onto AVPlayer's real (lagging) position after a VOD backpressure wedge. /// The producer was parked 10 segments ahead of a frozen consumer target; re-anchoring to where AVPlayer /// actually is puts the starved segments back into the producible window so AVPlayer can resume and land. diff --git a/Sources/AetherEngine/Video/HLSVideoEngine.swift b/Sources/AetherEngine/Video/HLSVideoEngine.swift index 4b021cad..94581245 100644 --- a/Sources/AetherEngine/Video/HLSVideoEngine.swift +++ b/Sources/AetherEngine/Video/HLSVideoEngine.swift @@ -1977,6 +1977,13 @@ public final class HLSVideoEngine: @unchecked Sendable { var barrenReopenCycles = 0 var lastReopenSegmentCount = -1 static let maxBarrenReopenCycles = 3 + /// Live muxerFailed in-place rebuilds, bounded by progress like the reopen cycles above: new segments + /// since the last death reset the budget (an hours-long channel legitimately crosses several encoder + /// restarts, so a session-lifetime gate like the VOD #99 one would be wrong here); consecutive barren + /// deaths exhaust it and the session halts + delegates to host retune. + var liveMuxerRebuildCycles = 0 + var lastMuxerRebuildSegmentCount = -1 + static let maxLiveMuxerRebuildCycles = 3 private func handleVideoShiftKnown(_ shiftPts: Int64, firstItemTfdtPts: Int64) { let seconds = shiftPts == Int64.min ? 0 : Double(shiftPts) * sourceVideoTbSeconds diff --git a/Tests/AetherEngineTests/LiveProductionHaltTests.swift b/Tests/AetherEngineTests/LiveProductionHaltTests.swift index efa46f5c..fc30c398 100644 --- a/Tests/AetherEngineTests/LiveProductionHaltTests.swift +++ b/Tests/AetherEngineTests/LiveProductionHaltTests.swift @@ -49,12 +49,46 @@ final class LiveProductionHaltTests: XCTestCase { XCTAssertFalse(HLSVideoEngine.shouldHaltLiveProduction( reason: .stopRequested, sourceReopenable: false)) XCTAssertFalse(HLSVideoEngine.shouldHaltLiveProduction( - reason: .muxerFailed, sourceReopenable: false)) + reason: .muxerFailed, sourceReopenable: false), + "handleLiveMuxerFailure rebuilds the producer into the same provider; an eager halt would 503 the provider the rebuild serves, and the arm halts itself on budget exhaustion") XCTAssertFalse(HLSVideoEngine.shouldHaltLiveProduction( reason: .backpressureWedge, sourceReopenable: false)) XCTAssertFalse(HLSVideoEngine.shouldHaltLiveProduction( reason: .needsAudioSampleEntryPrime, sourceReopenable: false), - "AE#222 rebuilds into the same provider with a primed muxer, so production continues") + "the AE#222 arm rebuilds into the same provider with a primed muxer (in place for live), so production continues") + } + + // MARK: - Live muxerFailed rebuild budget + + func testFirstLiveMuxerDeathAlwaysRebuilds() { + let d = HLSVideoEngine.liveMuxerRebuildDecision( + continuationIndex: 414, lastContinuationIndex: -1, barrenCycles: 0, cap: 3) + XCTAssertTrue(d.rebuild, "a fresh death at a new continuation point starts a fresh budget") + XCTAssertEqual(d.newBarrenCycles, 0) + } + + func testProgressSinceLastDeathResetsTheBudget() { + let d = HLSVideoEngine.liveMuxerRebuildDecision( + continuationIndex: 431, lastContinuationIndex: 414, barrenCycles: 2, cap: 3) + XCTAssertTrue(d.rebuild, + "segments were cut since the last death: an hours-long channel crossing several encoder restarts must not exhaust a session-lifetime budget") + XCTAssertEqual(d.newBarrenCycles, 0) + } + + func testConsecutiveBarrenDeathsExhaustTheBudget() { + var cycles = 0 + var last = -1 + var rebuilds = 0 + // Death after death at the same continuation point: nothing was ever produced in between. + for _ in 0..<10 { + let d = HLSVideoEngine.liveMuxerRebuildDecision( + continuationIndex: 414, lastContinuationIndex: last, barrenCycles: cycles, cap: 3) + cycles = d.newBarrenCycles + last = 414 + if d.rebuild { rebuilds += 1 } else { break } + } + XCTAssertEqual(rebuilds, 3, + "exactly the cap's worth of barren rebuilds, then halt + host retune — never an endless rebuild storm against an unmuxable source") } // MARK: - Provider halt latch