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