From 98b15a02c4777d4cda23b3f415ac2243d7f184e5 Mon Sep 17 00:00:00 2001 From: JonLuca DeCaro Date: Sun, 25 Oct 2026 09:26:15 -0700 Subject: [PATCH 1/2] fix(hls): keep empty FIFO writes from signaling EOF --- CHANGELOG.md | 4 ++ .../AetherEngine/IO/HLSIngest/ByteFIFO.swift | 5 +- Tests/AetherEngineTests/ByteFIFOTests.swift | 65 +++++++++++++++++++ docs/architecture.md | 2 +- 4 files changed, 74 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8f7a9fb8f..4f709f8f6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -168,6 +168,10 @@ the public-API contract. the bound port while `stop()` reset it under the lock. Host-only redirects now take the same lock, and the port stays private to the server. Explicit `--redirect-port` targets are unchanged. +- **Empty HLS FIFO writes no longer look like EOF.** Zero-byte writes leave readers waiting for + real data and do not wait for capacity. Nonpositive read lengths fail without consuming bytes; + zero remains reserved for a finished, drained queue. + - **Source failures keep their cause and request ownership.** Initial live probes preserve HTTP, URL and TLS failures without changing the live-start `liveSourceUnavailable` kind (HTTP and certificate classifications stay specific). A transport failure at open (a reset or a timeout from diff --git a/Sources/AetherEngine/IO/HLSIngest/ByteFIFO.swift b/Sources/AetherEngine/IO/HLSIngest/ByteFIFO.swift index 7ee1e936b..e9c55651d 100644 --- a/Sources/AetherEngine/IO/HLSIngest/ByteFIFO.swift +++ b/Sources/AetherEngine/IO/HLSIngest/ByteFIFO.swift @@ -43,6 +43,8 @@ final class ByteFIFO: @unchecked Sendable { func write(_ data: Data) -> Bool { condition.lock() defer { condition.unlock() } + // Empty writes neither occupy capacity nor make a reader's EOF condition true. + if data.isEmpty { return !finished && !cancelled } parked += 1 while pendingCount >= capacity && !finished && !cancelled { condition.wait() @@ -55,8 +57,9 @@ final class ByteFIFO: @unchecked Sendable { return true } - /// Blocking read. Returns: >0 bytes copied; 0 = EOF (finished + drained); -1 = cancelled. + /// Blocking read. Returns: >0 bytes copied; 0 = EOF (finished + drained); -1 = cancelled or invalid length. func read(into buffer: UnsafeMutablePointer, maxLength: Int) -> Int { + guard maxLength > 0 else { return -1 } condition.lock() defer { condition.unlock() } parked += 1 diff --git a/Tests/AetherEngineTests/ByteFIFOTests.swift b/Tests/AetherEngineTests/ByteFIFOTests.swift index 3924ab946..ce2ee383a 100644 --- a/Tests/AetherEngineTests/ByteFIFOTests.swift +++ b/Tests/AetherEngineTests/ByteFIFOTests.swift @@ -108,3 +108,68 @@ final class ByteFIFOTests: XCTestCase { wait(for: [expectation], timeout: 300) } } + +// Empty writes must not masquerade as the EOF that the ingest reader returns to FFmpeg. +// Run potentially blocking calls on real threads so cancellation can still release them. +import Testing +import Synchronization + +@Suite(.timeLimit(.minutes(1))) +struct ByteFIFOEmptyOperationsTests { + @Test("an empty write leaves a reader waiting for bytes") + func emptyWriteDoesNotSignalEOF() async throws { + let fifo = ByteFIFO(capacity: 4) + defer { fifo.cancel() } + #expect(fifo.write(Data())) + let result = Mutex(nil) + Thread.detachNewThread { + var byte: UInt8 = 0 + let count = fifo.read(into: &byte, maxLength: 1) + result.withLock { $0 = count } + } + try await waitFor { result.withLock { $0 != nil } || fifo.parkedWaiterCount > 0 } + #expect(result.withLock { $0 == nil }, "an empty write must not return zero (EOF)") + #expect(fifo.isEmptyWithReaderParked) + #expect(fifo.write(Data([42]))) + try await waitFor { result.withLock { $0 != nil } } + #expect(result.withLock { $0 == 1 }) + } + + @Test("an empty write completes even when capacity is occupied") + func emptyWriteDoesNotWaitForCapacity() async throws { + let fifo = ByteFIFO(capacity: 1) + defer { fifo.cancel() } + #expect(fifo.write(Data([42]))) + let result = Mutex(nil) + Thread.detachNewThread { + let accepted = fifo.write(Data()) + result.withLock { $0 = accepted } + } + try await waitFor { fifo.parkedWaiterCount > 0 || result.withLock { $0 != nil } } + #expect(fifo.parkedWaiterCount == 0) + fifo.cancel() + try await waitFor { result.withLock { $0 != nil } } + #expect(result.withLock { $0 == true }) + } + + @Test("empty writes still reject a finished or cancelled FIFO") + func emptyWriteHonorsTerminalState() { + let finished = ByteFIFO(capacity: 1) + finished.finish() + #expect(!finished.write(Data())) + let cancelled = ByteFIFO(capacity: 1) + cancelled.cancel() + #expect(!cancelled.write(Data())) + } + + @Test("a nonpositive read length is an error and preserves queued bytes", arguments: [0, -1, Int.min]) + func invalidLengthPreservesBytes(length: Int) { + let fifo = ByteFIFO(capacity: 1) + #expect(fifo.write(Data([42]))) + var byte: UInt8 = 99 + #expect(fifo.read(into: &byte, maxLength: length) == -1) + #expect(byte == 99) + #expect(fifo.read(into: &byte, maxLength: 1) == 1) + #expect(byte == 42) + } +} diff --git a/docs/architecture.md b/docs/architecture.md index 994601894..1490bfede 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -542,7 +542,7 @@ Sources/AetherEngine/ │ ├── HLSPlaylistTracker.swift Pure segment cursor: duration-capped edge join, window-slide rejoin, stall budget │ ├── HLSSegmentDecryptor.swift AES-128-CBC clear-key segment decryption (key fetch + memoise, PKCS7) │ ├── PackedAudioSegments.swift Packed-audio rendition support: LiveSegmentFormat classification + ID3 PRIV timestamp parser (raw ADTS segments) -│ ├── ByteFIFO.swift Bounded blocking byte queue between the fetch loop and the demux thread +│ ├── ByteFIFO.swift Bounded blocking byte queue between the fetch loop and the demux thread; empty writes are no-ops, invalid read lengths return an error, and only a finished, drained queue returns EOF │ ├── HLSIngestError.swift Typed terminal errors (encrypted, fMP4, unreachable, invalid, stalled) │ ├── LiveIngestSourceInfo.swift Internal seam: upstream segment cadence (shapes TARGETDURATION + blocking-reload eligibility) and DualSourceMergeOrder for the dual-source DTS merge │ ├── HLSVODIngestReader.swift Ingest reader for an immutable ENDLIST playlist: seeks restart the source at the segment preceding the requested playlist time, on an elapsed-media-time axis (the EXTINF sum) rather than the container's PTS From a78c49d75c72aedf75a388e64ec0898d538a0760 Mon Sep 17 00:00:00 2001 From: JonLuca DeCaro Date: Mon, 26 Oct 2026 07:44:26 -0700 Subject: [PATCH 2/2] docs(hls): describe FIFO guards as internal hardening --- CHANGELOG.md | 7 ++++--- Tests/AetherEngineTests/ByteFIFOTests.swift | 5 ++--- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4f709f8f6..514ce81bd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -168,9 +168,10 @@ the public-API contract. the bound port while `stop()` reset it under the lock. Host-only redirects now take the same lock, and the port stays private to the server. Explicit `--redirect-port` targets are unchanged. -- **Empty HLS FIFO writes no longer look like EOF.** Zero-byte writes leave readers waiting for - real data and do not wait for capacity. Nonpositive read lengths fail without consuming bytes; - zero remains reserved for a finished, drained queue. +- **The internal HLS byte FIFO defensively rejects operations that could look like EOF.** + Empty writes complete without occupying capacity, and nonpositive read lengths fail without + consuming bytes. Current ingest callers already filter both cases, so this hardens the internal + contract for future callers rather than changing existing playback. - **Source failures keep their cause and request ownership.** Initial live probes preserve HTTP, URL and TLS failures without changing the live-start `liveSourceUnavailable` kind (HTTP and diff --git a/Tests/AetherEngineTests/ByteFIFOTests.swift b/Tests/AetherEngineTests/ByteFIFOTests.swift index ce2ee383a..310f9026b 100644 --- a/Tests/AetherEngineTests/ByteFIFOTests.swift +++ b/Tests/AetherEngineTests/ByteFIFOTests.swift @@ -1,4 +1,6 @@ import XCTest +import Testing +import Synchronization @testable import AetherEngine final class ByteFIFOTests: XCTestCase { @@ -111,9 +113,6 @@ final class ByteFIFOTests: XCTestCase { // Empty writes must not masquerade as the EOF that the ingest reader returns to FFmpeg. // Run potentially blocking calls on real threads so cancellation can still release them. -import Testing -import Synchronization - @Suite(.timeLimit(.minutes(1))) struct ByteFIFOEmptyOperationsTests { @Test("an empty write leaves a reader waiting for bytes")