test(subtitles): cover independent secondary bitmap decoding

CHANGELOG.md

@@ -10,8 +10,14 @@ the public-API contract.
 
 ## [Unreleased]
 
+### Fixed
+
+- Persistent HTTP/1.1 HLS ingest checks FIFO headroom before every segment request, including when the previous body’s final chunk filled the buffer with no pending remainder. Seek, close and cancellation retire the waiting producer.
+
 ### Changed
 
+- Correct the secondary subtitle contract: embedded bitmap tracks already use independent channel decoders. Add real dual-PGS drain regression coverage for reused palette/object IDs, clear/off, and cached seek reconstruction. No playback behavior change.
+
 - **The documentation no longer says TrueHD Atmos objects are always lost.** The Known limitations
   entry in `docs/formats.md`, the `AudioDelivery.bridged` reference and the channel bridge's notes
   predate `objectAudioRendering` (#731) and said objects and heights never survive. They now say the

Sources/AetherEngine/AetherEngine+Subtitles.swift

@@ -167,7 +167,9 @@ extension AetherEngine {
         selectSubtitleTrack(index: Int(index), startAt: anchor > 0 ? anchor : sourceTime)
     }
 
-    /// Activate an embedded subtitle stream as the secondary companion track (issue #47). Text-only; bitmap codecs are rejected. Runs a second side demuxer concurrently. External ids (#88) route onto the secondary sidecar decode.
+    /// Activate an embedded subtitle stream as the secondary companion track (issue #47).
+    /// Text and supported bitmap codecs use an independent decoder and cursor over the shared
+    /// packet store. External ids (#88) route onto the secondary sidecar decode.
     public func selectSecondarySubtitleTrack(index: Int) {
         selectSecondarySubtitleTrack(index: index, startAt: sourceTime)
     }
@@ -1704,7 +1706,7 @@ extension AetherEngine {
     }
 
     /// Turn the secondary subtitle off and clear its cues. Tears down
-    /// the secondary sidecar decode task and the secondary side reader.
+    /// the secondary sidecar decode task and its packet-store drain target.
     public func clearSecondarySubtitle() {
         hostExplicitSubtitleAction = true
         cancelSidecarTask(channel: .secondary)

Sources/AetherEngine/AetherEngine.swift

@@ -1415,7 +1415,7 @@ public final class AetherEngine: ObservableObject {
     /// Nil for SRT/VTT and when markup preservation is off.
     @Published public internal(set) var sidecarASSHeader: String? = nil
 
-    /// Cues for the secondary subtitle track (#47). Text-only (bitmap rejected); independent of primary.
+    /// Cues for the secondary subtitle track (#47). Text and supported bitmap codecs; independent of primary.
     @Published public internal(set) var secondarySubtitleCues: [SubtitleCue] = []
     @Published public internal(set) var isLoadingSecondarySubtitles: Bool = false
     @Published public internal(set) var isSecondarySubtitleActive: Bool = false

Sources/AetherEngine/IO/HLSIngest/HLSVODIngestReader.swift

@@ -475,6 +475,9 @@ final class HLSVODIngestReader: TimeSeekableIOReader, @unchecked Sendable {
         defer { primed?.body.abandon() }
         for (offset, (segment, url)) in segments.enumerated() {
             try Task.checkCancellation()
+            // Runs on the producer's BlockingWork executor. The last offered chunk can fill the
+            // FIFO with no pending remainder, so admission must not depend on append() parking.
+            guard waitForSegmentAdmission(generation: generation) else { return }
             if let crypt = segment.crypt {
                 let bytes = try await decrypt(
                     try await fetch(url, limit: BoundedFetch.segmentLimit(forDuration: segment.duration)),
@@ -574,6 +577,18 @@ final class HLSVODIngestReader: TimeSeekableIOReader, @unchecked Sendable {
         return body
     }
 
+    /// Wait between requests, after the previous body has completed. Seek/close wake the producer
+    /// and invalidate its generation; a superseded producer must never open the next segment.
+    private func waitForSegmentAdmission(generation: UInt64) -> Bool {
+        condition.lock()
+        defer { condition.unlock() }
+        while self.generation == generation, !closed, !failed,
+              bufferedBytes >= Self.maximumBufferedBytes {
+            condition.wait()
+        }
+        return self.generation == generation && !closed && !failed
+    }
+
     private func append(_ data: Data, generation: UInt64) -> Bool {
         condition.lock()
         defer { condition.unlock() }

Tests/AetherEngineTests/Issue732PersistentHTTP1Tests.swift

@@ -1,6 +1,7 @@
 import CommonCrypto
 import Foundation
 import XCTest
+import Testing
 @testable import AetherEngine
 
 /// AE#732: under `heldSourceConnection` the HLS VOD ingest carries its playlist, probe and segment
@@ -426,14 +427,14 @@ final class Issue732PersistentHTTP1Tests: XCTestCase {
 
     static let markerOffset = 100
 
-    private static func mediaPlaylist(_ segments: [String]) -> Data {
+    fileprivate static func mediaPlaylist(_ segments: [String]) -> Data {
         var text = "#EXTM3U\n#EXT-X-TARGETDURATION:6\n"
         for segment in segments { text += "#EXTINF:6,\n\(segment)\n" }
         return Data((text + "#EXT-X-ENDLIST").utf8)
     }
 
     /// TS packets whose first PMT declares HEVC (0x24); the marker sits in the PMT packet's filler.
-    private static func transportStream(marker: UInt8, packets: Int = 3) -> Data {
+    fileprivate static func transportStream(marker: UInt8, packets: Int = 3) -> Data {
         var bytes = [UInt8](repeating: 0xFF, count: 188 * packets)
         for packet in 0..<packets {
             bytes[packet * 188] = 0x47
@@ -450,7 +451,7 @@ final class Issue732PersistentHTTP1Tests: XCTestCase {
         var bytes: [UInt8] = []
     }
 
-    private static func aes128CBC(_ data: Data, key: Data) -> Data {
+    fileprivate static func aes128CBC(_ data: Data, key: Data) -> Data {
         var out = Data(count: data.count + kCCBlockSizeAES128)
         var written = 0
         let iv = Data(count: kCCBlockSizeAES128)
@@ -684,3 +685,95 @@ final class KeepAliveOrigin: @unchecked Sendable {
         }
     }
 }
+
+@Suite("Persistent HLS segment-boundary admission", .serialized, .timeLimit(.minutes(1)))
+struct PersistentHLSSegmentAdmissionTests {
+    enum Exit: String, CaseIterable, Sendable { case consume, seek, close, cancel }
+
+    @Test("a final chunk filling the FIFO holds the next request until consumption or retirement",
+          arguments: Exit.allCases, [false, true])
+    func finalChunkAdmission(exit: Exit, encryptedNext: Bool) async throws {
+        let server = try #require(KeepAliveOrigin())
+        defer { server.stop() }
+        let segments = (0..<4).map { "seg\($0).ts" }
+        // After the one 188-byte read below, these two segments leave 16,777,308 bytes:
+        // just above the 16 MiB ceiling. The final offered chunk can empty `pending` while
+        // filling the FIFO, which used to skip append() and start the next request.
+        let packetCounts = [64_000, 25_242, 3, 3]
+        let key = Data(repeating: 0x43, count: 16)
+        var playlist = String(decoding: Issue732PersistentHTTP1Tests.mediaPlaylist(segments), as: UTF8.self)
+        if encryptedNext {
+            playlist = playlist.replacingOccurrences(of: "#EXTINF:6,\nseg2.ts",
+                with: "#EXT-X-KEY:METHOD=AES-128,URI=\"/key\",IV=0x00000000000000000000000000000000\n#EXTINF:6,\nseg2.ts")
+        }
+        let playlistBody = Data(playlist.utf8)
+        server.route = { path in
+            if path == "/vod.m3u8" {
+                return .init(body: playlistBody)
+            }
+            if path == "/key" { return .init(body: key) }
+            guard let index = segments.firstIndex(where: { "/" + $0 == path }) else {
+                return .init(status: 404)
+            }
+            let clear = Issue732PersistentHTTP1Tests.transportStream(
+                marker: UInt8(index), packets: packetCounts[index])
+            return .init(body: encryptedNext && index >= 2
+                ? Issue732PersistentHTTP1Tests.aes128CBC(clear, key: key) : clear)
+        }
+        let reader = try #require(try await HLSVODIngestReader.makeIfHEVCMPEGTS(
+            playlistURL: server.url("/vod.m3u8"), httpHeaders: [:], persistentHTTP1: true))
+        defer { reader.close() }
+        let firstRead = ProbeTestJob { () -> Int32 in
+            var bytes = [UInt8](repeating: 0, count: 188)
+            return bytes.withUnsafeMutableBufferPointer { reader.read($0.baseAddress, size: 188) }
+        }
+        #expect(try await firstRead.outcome().get() == 188)
+        try await waitFor { server.completed.contains("/seg1.ts") }
+        let escaped = try await waitFor(upTo: .milliseconds(500)) {
+            server.requests.contains { $0.path == "/seg2.ts" || $0.path == "/key" }
+        }
+        #expect(!escaped, "the final chunk must not bypass the next request's admission")
+
+        switch exit {
+        case .consume: break
+        case .seek: #expect(reader.seek(to: 20)) // starts at segment 2, before the containing segment 3
+        case .close: reader.close()
+        case .cancel: reader.cancel()
+        }
+        let drain = ProbeTestJob { () -> (bytes: Int, marker: UInt8?, terminal: Int32) in
+            var bytes = [UInt8](repeating: 0, count: 4096)
+            var total = 0
+            var marker: UInt8?
+            while true {
+                let n = bytes.withUnsafeMutableBufferPointer { reader.read($0.baseAddress, size: 4096) }
+                if n <= 0 { return (total, marker, n) }
+                if marker == nil, n > Issue732PersistentHTTP1Tests.markerOffset {
+                    marker = bytes[Issue732PersistentHTTP1Tests.markerOffset]
+                }
+                total += Int(n)
+            }
+        }
+        let result = try await drain.outcome().get()
+        switch exit {
+        case .consume:
+            #expect(result.bytes + 188 == packetCounts.reduce(0, +) * 188)
+            #expect(result.terminal == 0)
+        case .seek:
+            #expect(result.bytes == 6 * 188)
+            #expect(result.marker == 2)
+            #expect(result.terminal == 0)
+        case .close, .cancel:
+            #expect(result.bytes == 0)
+            #expect(result.terminal == -1)
+            let lateRequest = try await waitFor(upTo: .milliseconds(500)) {
+                server.requests.contains { $0.path == "/seg2.ts" || $0.path == "/key" }
+            }
+            #expect(!lateRequest, "a retired producer must not issue the next request")
+        }
+        if exit == .consume || exit == .seek {
+            #expect(server.requests.filter { $0.path == "/seg2.ts" }.count == 1)
+            #expect(server.connectionCount == 1)
+            #expect(server.maxInFlight == 1)
+        }
+    }
+}

Tests/AetherEngineTests/SecondarySubtitleChannelTests.swift

@@ -1,11 +1,14 @@
 import XCTest
+import Testing
+import AetherLibavformat
+import AetherLibavutil
 import AetherLibavcodec
 @testable import AetherEngine
 
 final class SecondarySubtitleChannelTests: XCTestCase {
 
-    // Documents the basis of the secondary text-only guard: the four
-    // bitmap subtitle codecs must classify as bitmap, text codecs must not.
+    // Bitmap classification is shared by both channels. Decoder availability still depends
+    // on the shipped FFmpeg build (in particular, XSUB is not built).
     func testBitmapCodecClassification() {
         XCTAssertTrue(EmbeddedSubtitleDecoder.isBitmapCodec(AV_CODEC_ID_HDMV_PGS_SUBTITLE))
         XCTAssertTrue(EmbeddedSubtitleDecoder.isBitmapCodec(AV_CODEC_ID_DVB_SUBTITLE))
@@ -29,3 +32,94 @@ final class SecondarySubtitleChannelTests: XCTestCase {
         XCTAssertFalse(engine.isSecondarySubtitleActive)
     }
 }
+
+/// Generated PGS packets deliberately reuse palette/object id zero on two streams.
+/// Drive the engine's actual packet-store drain with real shipped FFmpeg decoders.
+@MainActor
+struct SecondaryBitmapSubtitleTests {
+    static func segment(_ type: UInt8, _ body: [UInt8]) -> Data {
+        Data([type, UInt8(body.count >> 8), UInt8(body.count & 255)] + body)
+    }
+    static func pcs(x: Int, state: UInt8, clear: Bool = false) -> Data {
+        var body: [UInt8] = [7, 128, 4, 56, 16, 0, 1, state, 0, 0, clear ? 0 : 1]
+        if !clear { body += [0, 0, 0, 0, UInt8(x >> 8), UInt8(x & 255), 3, 132] }
+        return segment(0x16, body)
+    }
+    static func displaySet(x: Int, luminance: UInt8) -> Data {
+        let palette = segment(0x14, [0, 0, 0, 16, 128, 128, 0, 1, luminance, 128, 128, 255])
+        let rle = Array(repeating: [0, 0x88, 1, 0, 0] as [UInt8], count: 8).flatMap { $0 }
+        let object = segment(0x15, [0, 0, 0, 0xc0, 0, 0, 44, 0, 8, 0, 8] + rle)
+        return pcs(x: x, state: 0x80) + palette + object + segment(0x80, [])
+    }
+
+    @Test("PGS channels isolate palettes, clear, off and cached reconstruction", .timeLimit(.minutes(1)))
+    func independentChannels() async throws {
+        let format = try #require(avformat_alloc_context())
+        defer { avformat_free_context(format) }
+        for _ in 0..<2 {
+            let stream = try #require(avformat_new_stream(format, nil))
+            stream.pointee.time_base = AVRational(num: 1, den: 90_000)
+            stream.pointee.codecpar.pointee.codec_type = AVMEDIA_TYPE_SUBTITLE
+            stream.pointee.codecpar.pointee.codec_id = AV_CODEC_ID_HDMV_PGS_SUBTITLE
+        }
+        let engine = try AetherEngine()
+        defer { engine.clearSubtitle(); engine.clearSecondarySubtitle() }
+        engine.loadedURL = URL(string: "https://example.invalid/movie.mkv")!
+        func drain() async {
+            engine.requestSubtitleDrainTick()
+            while let pending = engine.subtitleDrainTickInFlight { await pending.value }
+        }
+        let store = SubtitlePacketStore()
+        engine.softwareSubtitlePacketStore = store
+        engine.subtitleDrainDecoderFactoryForTesting = { index in
+            guard (0..<2).contains(index), let stream = format.pointee.streams[Int(index)] else { return nil }
+            return EmbeddedSubtitleDecoder(stream: stream, sourceVideoWidth: 1920, sourceVideoHeight: 1080)
+        }
+        func harvest(_ stream: Int32, _ time: Double, _ data: Data) {
+            store.harvestChunk(streamIndex: stream, ptsSeconds: time, durationSeconds: 0,
+                               flags: 0, payload: data, assembleSplitDisplaySets: true)
+        }
+        func image(_ cues: [SubtitleCue], at time: Double) throws -> SubtitleImage {
+            let cue = try #require(cues.first { $0.startTime <= time && $0.endTime > time }, "no bitmap at \(time); intervals \(cues.map { [$0.startTime, $0.endTime] })")
+            guard case .image(let image) = cue.body else {
+                Issue.record("expected a decoded bitmap"); throw NSError(domain: "fixture", code: 1)
+            }
+            return image
+        }
+        func pixels(_ image: SubtitleImage) -> Data { image.cgImage.dataProvider!.data! as Data }
+        harvest(0, 10, Self.displaySet(x: 100, luminance: 235))
+        harvest(1, 10, Self.displaySet(x: 300, luminance: 100))
+        // Composition-only packets require the decoder's previously cached palette/object.
+        harvest(0, 12, Self.pcs(x: 100, state: 0) + Self.segment(0x80, []))
+        harvest(1, 11, Self.pcs(x: 300, state: 0, clear: true) + Self.segment(0x80, []))
+        engine.clock.sourceTime = 10
+        engine.selectSubtitleTrack(index: 0)
+        engine.selectSecondarySubtitleTrack(index: 1)
+        await drain()
+        let primary = try image(engine.subtitleCues, at: 10)
+        let secondary = try image(engine.secondarySubtitleCues, at: 10)
+        #expect(abs(primary.position.origin.x * 1920 - 100) < 0.01)
+        #expect(abs(secondary.position.origin.x * 1920 - 300) < 0.01)
+        #expect(pixels(primary) != pixels(secondary))
+        #expect(!engine.secondarySubtitleCues.contains { $0.startTime <= 11.5 && $0.endTime > 11.5 })
+        #expect(pixels(try image(engine.subtitleCues, at: 12)) == pixels(primary))
+        let primaryDecoder = try #require(engine.subtitleDrainDecoders[.primary])
+        let secondaryDecoder = try #require(engine.subtitleDrainDecoders[.secondary])
+        #expect(primaryDecoder !== secondaryDecoder)
+        engine.clearSecondarySubtitle()
+        #expect(engine.subtitleDrainDecoders[.primary] === primaryDecoder)
+        #expect(engine.subtitleDrainDecoders[.secondary] == nil)
+        #expect(engine.secondarySubtitleCues.isEmpty)
+        #expect(pixels(try image(engine.subtitleCues, at: 12)) == pixels(primary))
+        engine.selectSecondarySubtitleTrack(index: 1)
+        await drain()
+        #expect(pixels(try image(engine.secondarySubtitleCues, at: 10)) == pixels(secondary))
+        // A seek resets both channel cursors and reconstructs from retained packets.
+        engine.clock.sourceTime = 200
+        await drain()
+        engine.clock.sourceTime = 10
+        await drain()
+        #expect(pixels(try image(engine.subtitleCues, at: 10)) == pixels(primary))
+        #expect(pixels(try image(engine.secondarySubtitleCues, at: 10)) == pixels(secondary))
+    }
+}