From d5d9244c8137553c45adb0d76f79e6c8ca8f0962 Mon Sep 17 00:00:00 2001 From: Andreas Busslinger Date: Sat, 8 Aug 2026 22:41:53 +0200 Subject: [PATCH] fix(video): prime rotated muxers with the last muxed audio frame MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A mid-session muxer rotation (same-PID parameter-set change after a live reconnect join, SSAI program switch) allocates a brand-new MP4SegmentMuxer, and the only moov-prime carry-in was the construction -time audioMoovPrimeFrame — nil for every session that did not start audio-less. With bridged E-AC-3, video leads audio across the seam and the bridge adds encoder latency, so the rotated muxer's first cut can arrive before any post-seam audio packet: the cut defers awaiting a sample entry (AE#222), the producer turns the deferral into a pump exit, teardown finalize fails on the same precondition, and the live session dies with muxerFailed. The AE#222 exit-scan cannot convert the exit either — it rightly excludes bridged sessions, since a raw source frame cannot prime a bridge-encoded track. Have the producer retain the payload of the last audio frame a muxer accepted (copied before the write — movenc consumes the data ref — and committed only on write success) and hand it to every later muxer allocation, so a rotation is always primed and the deferral cannot fire mid-session. Scoped to AC-3/E-AC-3/TrueHD via the muxer's own parsed-packet predicate, now exposed as a static helper; AAC never copies. One retained frame, pump-thread confined. Tests pin the incident shape (unprimed rotated muxer over a real bridge-encoder track: video-only cut defers, finalize salvages nothing) and the fix's load-bearing claim: a BRIDGE-OUTPUT E-AC-3 frame is a valid dec3 prime, not just a source frame as in AE#222. Co-Authored-By: Claude Fable 5 --- .../Video/HLSSegmentProducer.swift | 38 +++- .../AetherEngine/Video/MP4SegmentMuxer.swift | 21 +- .../LiveRotationAudioPrimeTests.swift | 202 ++++++++++++++++++ 3 files changed, 247 insertions(+), 14 deletions(-) create mode 100644 Tests/AetherEngineTests/LiveRotationAudioPrimeTests.swift diff --git a/Sources/AetherEngine/Video/HLSSegmentProducer.swift b/Sources/AetherEngine/Video/HLSSegmentProducer.swift index 0e66b83b..0669e28b 100644 --- a/Sources/AetherEngine/Video/HLSSegmentProducer.swift +++ b/Sources/AetherEngine/Video/HLSSegmentProducer.swift @@ -249,6 +249,15 @@ final class HLSSegmentProducer: @unchecked Sendable { /// AE#222: real audio frame handed to every muxer this producer builds, so moov (with its packet-derived /// dec3/dac3/dmlp) is written at init instead of at the first cut. nil on a normal session. private let audioMoovPrimeFrame: [UInt8]? + /// Most recent audio frame a muxer accepted, retained so every LATER allocation (same-PID parameter-set + /// rotation, SSAI program switch) starts primed too: video leads audio across a seam, so a rotated + /// muxer's first cut can arrive before any post-seam audio packet exists — unprimed, that cut defers + /// for a sample entry the bridge's encoder latency will never deliver in time and the pump dies with + /// `muxerFailed`. Pump-thread confined; seeded from `audioMoovPrimeFrame`. + private var lastMuxedAudioPrimeFrame: [UInt8]? + /// True when the session's audio codec derives its mp4 sample entry from a parsed packet + /// (AC-3/E-AC-3/TrueHD); only then is a per-frame prime copy worth the memcpy. AAC never copies. + private let capturesAudioPrimeFrames: Bool /// Latched when a cut deferred for want of an audio sample entry; the pump then scans forward for one /// real audio frame and exits with `.needsAudioSampleEntryPrime` so the host can rebuild primed. private var cutDeferredAwaitingAudioSampleEntry: Bool = false @@ -790,6 +799,9 @@ final class HLSSegmentProducer: @unchecked Sendable { ) throws { self.epoch = epoch self.audioMoovPrimeFrame = audioMoovPrimeFrame + self.lastMuxedAudioPrimeFrame = audioMoovPrimeFrame + self.capturesAudioPrimeFrames = + audio.map { MP4SegmentMuxer.audioNeedsParsedPacketForMoov($0.codecpar.pointee.codec_id) } ?? false self.bufferAheadSegments = bufferAheadSegments self.prefetchDiskBudgetBytes = prefetchDiskBudgetBytes self.demuxer = demuxer @@ -1320,8 +1332,10 @@ final class HLSSegmentProducer: @unchecked Sendable { // Floored at 8s (the historical 2 x 4s value): a sub-second fastZap cut target (AE#195) // must not shrink the cap below typical TS A/V interleave skew. maxBufferedFragmentSeconds: max(8.0, 2 * targetSegmentDurationSeconds), - // AE#222: only set on a session that already proved its first segment carries no audio. - audioMoovPrimeFrame: audioMoovPrimeFrame, + // AE#222 + mid-session rotation: prefer the last frame a muxer accepted, so a rotation + // stays primed even when video leads audio across the seam; the construction-time prime + // covers only the first allocation of a session whose first segment carries no audio. + audioMoovPrimeFrame: lastMuxedAudioPrimeFrame ?? audioMoovPrimeFrame, onInitCaptured: { [weak self] initBytes in guard let self = self else { return } if versionedInit { @@ -2831,7 +2845,10 @@ final class HLSSegmentProducer: @unchecked Sendable { } fp.pointee.stream_index = muxer.audioOutputStreamIndex av_packet_rescale_ts(fp, audio.inputTimeBase, muxer.muxerAudioTimeBase) - _ = muxer.writePacket(fp) + let prime = copyAudioPrimeCandidate(fp) + if muxer.writePacket(fp).rc >= 0, let prime { + lastMuxedAudioPrimeFrame = prime + } trackedPacketFree(&fpVar) } if bridgedMuxerGone { @@ -3120,6 +3137,16 @@ final class HLSSegmentProducer: @unchecked Sendable { packet.pointee.size -= headerLen } + /// Copies the payload of an audio packet about to be muxed, BEFORE the write: movenc consumes the + /// packet's data reference, so afterwards there is nothing left to copy. The caller commits the copy + /// into `lastMuxedAudioPrimeFrame` only when the write succeeded. + private func copyAudioPrimeCandidate(_ packet: UnsafeMutablePointer) -> [UInt8]? { + guard capturesAudioPrimeFrames, let data = packet.pointee.data, packet.pointee.size > 0 else { + return nil + } + return [UInt8](UnsafeBufferPointer(start: data, count: Int(packet.pointee.size))) + } + /// Stream-copy audio only; bridge audio bypasses this (FLAC encoder sets durations correctly). private func finalizeAndWriteAudio( _ packet: UnsafeMutablePointer, @@ -3138,7 +3165,10 @@ final class HLSSegmentProducer: @unchecked Sendable { packet.pointee.stream_index = muxer.audioOutputStreamIndex av_packet_rescale_ts(packet, audio.inputTimeBase, muxer.muxerAudioTimeBase) - _ = muxer.writePacket(packet) + let prime = copyAudioPrimeCandidate(packet) + if muxer.writePacket(packet).rc >= 0, let prime { + lastMuxedAudioPrimeFrame = prime + } var pkt: UnsafeMutablePointer? = packet trackedPacketFree(&pkt) diff --git a/Sources/AetherEngine/Video/MP4SegmentMuxer.swift b/Sources/AetherEngine/Video/MP4SegmentMuxer.swift index ba7156e3..83518c1d 100644 --- a/Sources/AetherEngine/Video/MP4SegmentMuxer.swift +++ b/Sources/AetherEngine/Video/MP4SegmentMuxer.swift @@ -140,6 +140,15 @@ final class MP4SegmentMuxer { /// from codecpar alone, so they never wedge, and gating the #64 RAM-cap flush on them would needlessly /// weaken that memory bound. Latched at init from the audio codec_id. private let audioNeedsParsedPacketForMoov: Bool + + /// Only AC-3 / E-AC-3 / TrueHD build their mp4 sample entry from a parsed packet (dac3/dec3/dmlp), + /// so only they can hit the "moov before audio parsed" wedge and need the #64-flush guard. Shared with + /// the producer, which uses it to decide whether retaining a moov-prime frame copy buys anything. + static func audioNeedsParsedPacketForMoov(_ codecID: AVCodecID) -> Bool { + codecID == AV_CODEC_ID_AC3 || + codecID == AV_CODEC_ID_EAC3 || + codecID == AV_CODEC_ID_TRUEHD + } /// Latched when the next staging file open fails; producer must stop the pump. private(set) var isWedged: Bool = false /// Latched after avformat_write_header; mp4 muxer rewrites time_base to its own pick @@ -183,16 +192,8 @@ final class MP4SegmentMuxer { self.currentSegmentIndex = initialSegmentIndex self.sessionDir = sessionDir self.haveAudio = audio != nil - // Only AC-3 / E-AC-3 / TrueHD build their mp4 sample entry from a parsed packet (dac3/dec3/dmlp), - // so only they can hit the "moov before audio parsed" wedge and need the #64-flush guard. - if let audioCodecID = audio?.codecpar.pointee.codec_id { - self.audioNeedsParsedPacketForMoov = - audioCodecID == AV_CODEC_ID_AC3 || - audioCodecID == AV_CODEC_ID_EAC3 || - audioCodecID == AV_CODEC_ID_TRUEHD - } else { - self.audioNeedsParsedPacketForMoov = false - } + self.audioNeedsParsedPacketForMoov = + audio.map { Self.audioNeedsParsedPacketForMoov($0.codecpar.pointee.codec_id) } ?? false let firstPath = Self.stagingPath(forSegmentIndex: initialSegmentIndex, in: sessionDir) diff --git a/Tests/AetherEngineTests/LiveRotationAudioPrimeTests.swift b/Tests/AetherEngineTests/LiveRotationAudioPrimeTests.swift new file mode 100644 index 00000000..87537ee6 --- /dev/null +++ b/Tests/AetherEngineTests/LiveRotationAudioPrimeTests.swift @@ -0,0 +1,202 @@ +import Testing +import Foundation +import Libavformat +import Libavcodec +import Libavutil +@testable import AetherEngine + +/// Live rotation wedge: a mid-session muxer rotation (same-PID parameter-set change after a live reconnect +/// join, or an SSAI program switch) builds a brand-new muxer, and with BRIDGED E-AC-3 the rotated muxer's +/// first cut can arrive before any post-seam audio packet exists — video leads audio across the seam and +/// the bridge adds encoder latency on top. Unprimed, that cut defers for a sample entry (AE#222), the +/// producer converts the deferral into a pump exit, teardown finalize fails on the same precondition, and +/// the live session dies with `muxerFailed`. The exit-scan cannot help either: it rightly excludes bridged +/// sessions (a raw source frame cannot prime a bridge-encoded track). +/// +/// The fix has the producer retain the last audio frame a muxer accepted and prime every later allocation +/// with it. These tests pin the claim that rests on, which AE#222's source-frame tests do not cover: a +/// BRIDGE-OUTPUT E-AC-3 frame is a valid moov prime for a muxer whose audio track IS the bridge encoder. +@Suite("Live rotation: bridge-output audio frame as moov prime") +struct LiveRotationAudioPrimeTests { + + // MARK: - Harness + + /// Little-endian 16-bit PCM WAV with a 440 Hz sine (same shape as Issue99BridgeResumeTests). + private static func makeWAV(sampleRate: Int, channels: Int, seconds: Double) -> Data { + let frames = Int(Double(sampleRate) * seconds) + var pcm = Data(capacity: frames * channels * 2) + for n in 0.. AudioBridge(.surroundCompat) -> E-AC-3 packets, + /// plus the H.264 video fixture, feeding a muxer whose audio track is the bridge's encoder. + private final class Rig { + let videoDemuxer = Demuxer() + let audioDemuxer = Demuxer() + let sessionDir: URL + var bridge: AudioBridge? + var initBytes: Data? + var muxer: MP4SegmentMuxer? + /// Payloads of every E-AC-3 packet the bridge emitted, in order. + var bridgeFrames: [[UInt8]] = [] + + init() throws { + sessionDir = FileManager.default.temporaryDirectory + .appendingPathComponent("live-rotation-prime-\(UUID().uuidString)") + try FileManager.default.createDirectory(at: sessionDir, withIntermediateDirectories: true) + try videoDemuxer.open( + reader: DataIOReader(data: Data( + base64Encoded: AtmosDetectionProbeIntegrationTests.videoOnlyBase64, + options: .ignoreUnknownCharacters) ?? Data()), + formatHint: "mp4" + ) + } + + deinit { + muxer = nil + bridge?.close() + videoDemuxer.close() + audioDemuxer.close() + try? FileManager.default.removeItem(at: sessionDir) + } + + /// Opens the bridge over an in-memory WAV and runs every source packet through it. + func runBridge() throws { + try audioDemuxer.open( + reader: DataIOReader(data: LiveRotationAudioPrimeTests.makeWAV( + sampleRate: 48_000, channels: 2, seconds: 1.0)) + ) + let audioIdx = audioDemuxer.audioStreamIndex + guard audioIdx >= 0, let stream = audioDemuxer.stream(at: audioIdx) else { + throw RigError.noStream + } + let bridge = try AudioBridge( + srcCodecpar: stream.pointee.codecpar, + srcTimeBase: stream.pointee.time_base, + mode: .surroundCompat + ) + self.bridge = bridge + while let packet = try audioDemuxer.readPacket() { + var p: UnsafeMutablePointer? = packet + defer { trackedPacketFree(&p) } + guard packet.pointee.stream_index == audioIdx else { continue } + for out in try bridge.feed(packet: packet) { + var o: UnsafeMutablePointer? = out + defer { trackedPacketFree(&o) } + if let data = out.pointee.data, out.pointee.size > 0 { + bridgeFrames.append( + [UInt8](UnsafeBufferPointer(start: data, count: Int(out.pointee.size)))) + } + } + } + } + + /// The rotated muxer: fresh AVFormatContext whose audio track is the bridge's E-AC-3 encoder, + /// exactly what `rotateMuxerForProgramSwitch` allocates mid-session. + func makeRotatedMuxer(audioPrime: [UInt8]?) throws -> MP4SegmentMuxer { + guard let vStream = videoDemuxer.stream(at: videoDemuxer.videoStreamIndex), + let encoderCodecpar = bridge?.encoderCodecpar, let bridge else { + throw RigError.noStream + } + let m = try MP4SegmentMuxer( + initialSegmentIndex: 414, + sessionDir: sessionDir, + video: MP4SegmentMuxer.VideoConfig( + codecpar: UnsafePointer(vStream.pointee.codecpar), + timeBase: vStream.pointee.time_base, + codecTagOverride: nil + ), + audio: MP4SegmentMuxer.AudioConfig( + codecpar: UnsafePointer(encoderCodecpar), + timeBase: bridge.encoderTimeBase + ), + audioMoovPrimeFrame: audioPrime, + onInitCaptured: { [self] bytes in self.initBytes = bytes } + ) + muxer = m + return m + } + + /// Post-seam video: every fixture packet, rescaled like the producer does. + @discardableResult + func writeAllVideoPackets(into muxer: MP4SegmentMuxer) throws -> Int { + guard let vStream = videoDemuxer.stream(at: videoDemuxer.videoStreamIndex) else { return 0 } + let sourceTb = vStream.pointee.time_base + var written = 0 + while let pkt = try videoDemuxer.readPacket() { + var p: UnsafeMutablePointer? = pkt + defer { trackedPacketFree(&p) } + guard pkt.pointee.stream_index == videoDemuxer.videoStreamIndex else { continue } + pkt.pointee.stream_index = muxer.videoOutputStreamIndex + av_packet_rescale_ts(pkt, sourceTb, muxer.muxerVideoTimeBase) + _ = muxer.writePacket(pkt) + written += 1 + } + return written + } + } + + private enum RigError: Error { case noStream } + + private static func containsBox(_ data: Data?, _ fourCC: String) -> Bool { + guard let data else { return false } + return data.range(of: Data(fourCC.utf8)) != nil + } + + // MARK: - The incident, pinned + + @Test("unprimed rotation with bridged EAC3: video-only cut defers and teardown finalize salvages nothing") + func unprimedRotationDiesLikeTheIncident() throws { + let rig = try Rig() + try rig.runBridge() + let muxer = try rig.makeRotatedMuxer(audioPrime: nil) + + let written = try rig.writeAllVideoPackets(into: muxer) + #expect(written > 0, "fixture must contribute post-seam video packets") + + #expect(muxer.cutFragmentForNextSegment(415) == .deferredAwaitingAudioSampleEntry, + "the rotated muxer's first cut arrives before any post-seam bridge output") + #expect(muxer.finalize() == nil, + "teardown finalize fails on the same precondition — the 'final finalize failed; not adopted' -> muxerFailed pump death") + } + + // MARK: - The fix's load-bearing claim + + @Test("a bridge-output EAC3 frame primes the rotated muxer: moov+dec3 at init, video-only cut completes") + func bridgeOutputFramePrimesRotatedMuxer() throws { + let rig = try Rig() + try rig.runBridge() + #expect(!rig.bridgeFrames.isEmpty, "bridge must emit E-AC-3 packets") + + // What the producer's capture retains: the payload of the last frame a muxer accepted. + let muxer = try rig.makeRotatedMuxer(audioPrime: rig.bridgeFrames.last) + + #expect(Self.containsBox(rig.initBytes, "moov"), "moov is written at init, before any packet") + #expect(Self.containsBox(rig.initBytes, "dec3"), + "dec3 must be derivable from a BRIDGE-OUTPUT frame, not just a source frame") + + try rig.writeAllVideoPackets(into: muxer) + guard case .completed(let path, let bytes) = muxer.cutFragmentForNextSegment(415) else { + Issue.record("primed rotation cut must complete") + return + } + #expect(bytes > 0) + let segment = try Data(contentsOf: path) + #expect(Self.containsBox(segment, "moof"), "the delivered segment is a plain fragment") + #expect(!Self.containsBox(segment, "moov"), "moov already went to the versioned init") + } +}