fix(hls): gate each segment request on FIFO headroom

CHANGELOG.md

@@ -10,6 +10,10 @@ 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
 
 - **The documentation no longer says TrueHD Atmos objects are always lost.** The Known limitations

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

docs/api.md

@@ -1334,7 +1334,7 @@ All flags default to safe values; the table is the full set. Depth for the media
 | `declaredDurationSeconds` | nil | Trusted duration, overriding the container's. Required alongside `sequentialOrigin` on VOD, where the tail read is gone. |
 | `maxConcurrentSourceRequests` | nil | Most requests the reader may have open against this origin at once, across every path it fetches on (pump ranges, detour blocks, size probes, tail prefetch, subtitle side reader, and the HLS live ingest's playlist, segment and key fetches, whose prefetch window narrows to it, AE#678). nil counts without capping and lowers the ceiling on its own after a 429/503/509. Set it when the provider states a limit; `1` also switches off the speculative parallel paths, which exist only to overlap with the pump. Counts **requests**, not TCP connections, because over HTTP/2 a session multiplexes every request onto one connection while the origin still counts each one (AE#377). It is also the only ceiling: several engines playing from one origin are bounded by this value and by what the origin refuses, not by a transport pool underneath it (AE#450). |
 | `sourceOpenPolicy` | `SourceOpenPolicy()` | HTTP VOD opening budgets for the playback reader and its reopens: `firstByteTimeout` (15 s) bounds the first data wait, `sizeProbeTimeout` (25 s) bounds the one retry and the whole size discovery, slot waits included. Nonpositive or nonfinite values take the defaults, anything above 120 s is capped. A request that gets no response is retried once with an open-ended Range and that connection stays the playback connection; two unanswered requests fail as `.sourceOpenFailed` instead of demoting the source to forward-only. A real length-less or Range-ignoring response keeps the ordinary sequential handling. Single-request origins run the size probes serially, others race them and join the losers before playback starts. Live, `sequentialOrigin`, remote HLS and disposable probes keep their own budgets. |
-| `heldSourceConnection` | false | Ask the source **once** and pull it, instead of ending the connection at the reader's window high water and asking again every 8 to 16 MB of drain (AE#377). For an origin that punishes repeated requests rather than concurrency: some CDNs refuse new requests for minutes at a stretch while serving an already open connection at full rate, and against one of those the request cadence is the defect, at any range size (measured at the reporting origin: 32 MB ranges raised to 256 MB, eight times fewer requests, the refusals unchanged). Where `maxConcurrentSourceRequests` bounds how many requests are in flight, this removes the second request. The read path is the engine's own HTTP/1.1 over a demand-driven stream task, so: **HTTP/1.1 only** (no ALPN, an HTTP/2-only origin is out of scope), the **system proxy configuration does not apply**, and TLS is the OS's through the same host trust decision as every other engine session but has not been exercised against a self-signed origin. A viewer who pauses ends the connection after five seconds and resuming costs one request at the frontier, because a held flow nobody reads is the process-wide Network.framework starvation of AE#310. Applies to the playback reader; the subtitle and enrichment side readers keep the default transport, since they park for minutes at a time. The open-time detection of an origin that ignores `Range` (see `sequentialOrigin`) does not cover it, because its open is open-ended. A finite HLS VOD on the HEVC-in-MPEG-TS ingest (AE#268) is covered too (AE#732): playlist, key and segment requests run over one persistent HTTP/1.1 connection per origin, reused across segments and fetched one at a time, for origins that refuse h2 or new connections but serve consecutive requests on one keep-alive connection. Each body is streamed into the ingest as it arrives, and only the wait for a response head has a wall-clock ceiling (45 s); a body ends on its size limit or on 30 s without a byte, never on its duration. A new connection opens only on a server close, an origin change, or a seek that abandons a body with more than 2 MB left or too slow to arrive within 3 s (otherwise the remainder is drained and the connection kept, with the next request waiting for it), and the log names each connection's protocol and request count. **Names the session**: the transport is chosen at open, so a reload that changes it is refused. |
+| `heldSourceConnection` | false | Ask the source **once** and pull it, instead of ending the connection at the reader's window high water and asking again every 8 to 16 MB of drain (AE#377). For an origin that punishes repeated requests rather than concurrency: some CDNs refuse new requests for minutes at a stretch while serving an already open connection at full rate, and against one of those the request cadence is the defect, at any range size (measured at the reporting origin: 32 MB ranges raised to 256 MB, eight times fewer requests, the refusals unchanged). Where `maxConcurrentSourceRequests` bounds how many requests are in flight, this removes the second request. The read path is the engine's own HTTP/1.1 over a demand-driven stream task, so: **HTTP/1.1 only** (no ALPN, an HTTP/2-only origin is out of scope), the **system proxy configuration does not apply**, and TLS is the OS's through the same host trust decision as every other engine session but has not been exercised against a self-signed origin. A viewer who pauses ends the connection after five seconds and resuming costs one request at the frontier, because a held flow nobody reads is the process-wide Network.framework starvation of AE#310. Applies to the playback reader; the subtitle and enrichment side readers keep the default transport, since they park for minutes at a time. The open-time detection of an origin that ignores `Range` (see `sequentialOrigin`) does not cover it, because its open is open-ended. A finite HLS VOD on the HEVC-in-MPEG-TS ingest (AE#268) is covered too (AE#732): playlist, key and segment requests run over one persistent HTTP/1.1 connection per origin, reused across segments and fetched one at a time, for origins that refuse h2 or new connections but serve consecutive requests on one keep-alive connection. Each body is streamed into the ingest as it arrives. A full FIFO lets the current body finish but admits no next segment or key request until consumption makes room; seek, close or cancellation retires that wait. Only the wait for a response head has a wall-clock ceiling (45 s); a body ends on its size limit or on 30 s without a byte, never on its duration. A new connection opens only on a server close, an origin change, or a seek that abandons a body with more than 2 MB left or too slow to arrive within 3 s (otherwise the remainder is drained and the connection kept, with the next request waiting for it), and the log names each connection's protocol and request count. **Names the session**: the transport is chosen at open, so a reload that changes it is refused. |
 | `autoplay` | true | False mounts paused: the load skips the terminal `play()` and settles at `.paused` for a host that resumes later. It describes THIS MOUNT and nothing after it: the rebuilds a session makes on its own (`reloadAtCurrentPosition`, an option correction, the AirPlay LAN swap, an audio-delay nudge) come back in the transport state the session is in, not in this one. Correcting it through `reloadAtCurrentPosition(applying:)` therefore does nothing (the log names it as not applied rather than as applied, AE#464 round 3, and the call returns it as `sessionOwned` with `rebuilt` false rather than spending a teardown on it, AE#464 round 4); call `play()` or `pause()` instead. |
 
 ## Value types