From 7a1cae45892add6962e598de45a0b1541f3bc58f Mon Sep 17 00:00:00 2001 From: Andreas Busslinger Date: Sun, 9 Aug 2026 10:05:39 +0200 Subject: [PATCH 1/4] fix(io): a sequential origin is asked for the stream once, from zero IPTV timeshift/catch-up archives fabricate range answers: any Range: bytes=X- gets a plausible 206 whose body actually sits on a coarse internal chunk boundary, so every 32 MB range rotation spliced ~1.9 s of misplaced content into the pump (mpegts "Packet corrupt", negative audio packet durations, a once-a-minute audio desync on device). Headers cannot expose the lie, so LoadOptions.sequentialOrigin lets the caller declare it: the reader routes onto its existing forward-only streaming mode (one long-lived unranged GET), skipping the suffix/tail prefetch, the optimistic persistent open, the size probe and every detour, and the pb stays non-seekable. With the ranged tail read gone the container resolves no duration (the fabricated tail data probed a 135-min window as 9.5 h), so the paired LoadOptions.declaredDurationSeconds carries the caller-known duration with the highest precedence in the AE#105 chain (declared > custom reader > disc MPLS/IFO > container). A lost connection cannot be resumed at an offset, so the streaming read reports EIO instead of EOF when the body ends short of its advisory Content-Length (EOF would read as end-of-media, which hosts deliberately never retry), and the #169 VOD readError revive is not admitted (its fresh demuxer could only reopen from byte 0 and then fails its anchor seek, burning a slot on connection-capped panels) - the pump loss goes straight to onVODSourceFailed for the host to re-request. The StreamingDelegate also gains the same redirect header preservation as every other delegate in the file; IPTV origins 302 twice before the archive host. aetherctl play grows --sequential-origin / --declared-duration for manual runs against a stub or real portal. Co-Authored-By: Claude Fable 5 --- .../AetherEngine/AetherEngine+Loading.swift | 14 +- Sources/AetherEngine/AetherEngine.swift | 2 + Sources/AetherEngine/Demuxer/AVIOReader.swift | 100 ++++++++++++- Sources/AetherEngine/Demuxer/Demuxer.swift | 60 ++++++-- Sources/AetherEngine/PlayerState.swift | 25 ++++ .../Video/HLSVideoEngine+LiveReopen.swift | 13 +- .../AetherEngine/Video/HLSVideoEngine.swift | 23 ++- Sources/aetherctl/PlaybackCmd.swift | 9 +- Sources/aetherctl/main.swift | 14 +- .../SequentialOriginReaderTests.swift | 138 ++++++++++++++++++ .../ThrottledOriginServer.swift | 31 ++++ 11 files changed, 401 insertions(+), 28 deletions(-) create mode 100644 Tests/AetherEngineTests/SequentialOriginReaderTests.swift diff --git a/Sources/AetherEngine/AetherEngine+Loading.swift b/Sources/AetherEngine/AetherEngine+Loading.swift index 1cd56b35..5248489b 100644 --- a/Sources/AetherEngine/AetherEngine+Loading.swift +++ b/Sources/AetherEngine/AetherEngine+Loading.swift @@ -502,6 +502,8 @@ extension AetherEngine { // Caller-bounded probe budget (#68) for the fallback open / live reopen; the happy path reuses preopenedDemuxer. probesize: loadedOptions.probesize, maxAnalyzeDuration: loadedOptions.maxAnalyzeDuration, + sequentialOrigin: loadedOptions.sequentialOrigin, + declaredDurationSeconds: loadedOptions.declaredDurationSeconds, forwardBufferSegments: loadedOptions.forwardBufferSegments ) // #240: the pump claims the source link through this gate while it is fetching, so the @@ -1292,18 +1294,20 @@ extension AetherEngine { // Capture the caller's probe budget (#68) before the detach: loadedOptions is @MainActor-isolated and unreachable inside the closure. Only used on the fallback open (probe absent). let probesize = loadedOptions.probesize let maxAnalyzeDuration = loadedOptions.maxAnalyzeDuration + let sequentialOrigin = loadedOptions.sequentialOrigin + let declaredDuration = loadedOptions.declaredDurationSeconds // Built on the main actor, captured into the detach: surfaces source stall/reconnect to playbackPhase (#85). let networkPhaseSink: @Sendable (ReaderNetworkPhase) -> Void = { [weak self] phase in Task { @MainActor in self?.setReaderNetworkPhase(phase) } } try await Task.detached(priority: .userInitiated) { - [host, preopenedDemuxer, url, sourceHTTPHeaders, isLive, dvrWindowSeconds, probesize, maxAnalyzeDuration, networkPhaseSink] in + [host, preopenedDemuxer, url, sourceHTTPHeaders, isLive, dvrWindowSeconds, probesize, maxAnalyzeDuration, sequentialOrigin, declaredDuration, networkPhaseSink] in let dem: Demuxer if let pre = preopenedDemuxer { dem = pre } else { dem = Demuxer() - try dem.open(url: url, extraHeaders: sourceHTTPHeaders, profile: .playback.withProbeBudget(probesize: probesize, maxAnalyzeDuration: maxAnalyzeDuration), isLive: isLive) + try dem.open(url: url, extraHeaders: sourceHTTPHeaders, profile: .playback.withProbeBudget(probesize: probesize, maxAnalyzeDuration: maxAnalyzeDuration).withSequentialOrigin(sequentialOrigin, declaredDuration: declaredDuration), isLive: isLive) } dem.onNetworkPhaseChanged = networkPhaseSink try await host.load( @@ -1359,18 +1363,20 @@ extension AetherEngine { // Caller's probe budget (#68) captured before the detach; only used on the fallback open (probe absent). let probesize = loadedOptions.probesize let maxAnalyzeDuration = loadedOptions.maxAnalyzeDuration + let sequentialOrigin = loadedOptions.sequentialOrigin + let declaredDuration = loadedOptions.declaredDurationSeconds // Built on the main actor, captured into the detach: surfaces source stall/reconnect to playbackPhase (#85). let networkPhaseSink: @Sendable (ReaderNetworkPhase) -> Void = { [weak self] phase in Task { @MainActor in self?.setReaderNetworkPhase(phase) } } try await Task.detached(priority: .userInitiated) { - [host, preopenedDemuxer, url, sourceHTTPHeaders, probesize, maxAnalyzeDuration, networkPhaseSink] in + [host, preopenedDemuxer, url, sourceHTTPHeaders, probesize, maxAnalyzeDuration, sequentialOrigin, declaredDuration, networkPhaseSink] in let dem: Demuxer if let pre = preopenedDemuxer { dem = pre } else { dem = Demuxer() - try dem.open(url: url, extraHeaders: sourceHTTPHeaders, profile: .playback.withProbeBudget(probesize: probesize, maxAnalyzeDuration: maxAnalyzeDuration)) + try dem.open(url: url, extraHeaders: sourceHTTPHeaders, profile: .playback.withProbeBudget(probesize: probesize, maxAnalyzeDuration: maxAnalyzeDuration).withSequentialOrigin(sequentialOrigin, declaredDuration: declaredDuration)) } dem.onNetworkPhaseChanged = networkPhaseSink try await host.load( diff --git a/Sources/AetherEngine/AetherEngine.swift b/Sources/AetherEngine/AetherEngine.swift index 0cfc4e3e..800499a6 100644 --- a/Sources/AetherEngine/AetherEngine.swift +++ b/Sources/AetherEngine/AetherEngine.swift @@ -2568,6 +2568,8 @@ public final class AetherEngine: ObservableObject { // demuxer is reused as the session demuxer, so the cap lands on the open that actually pays it. let probeProfile = DemuxerOpenProfile.playback.withProbeBudget( probesize: options.probesize, maxAnalyzeDuration: options.maxAnalyzeDuration) + .withSequentialOrigin(options.sequentialOrigin, + declaredDuration: options.declaredDurationSeconds) switch source { case .url(let u): // isLive configures the AVIOReader for endless-feed mode; must be set at open time because diff --git a/Sources/AetherEngine/Demuxer/AVIOReader.swift b/Sources/AetherEngine/Demuxer/AVIOReader.swift index e494ef19..3c10b3d1 100644 --- a/Sources/AetherEngine/Demuxer/AVIOReader.swift +++ b/Sources/AetherEngine/Demuxer/AVIOReader.swift @@ -190,6 +190,12 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { private var streamEnded = false private let streamLock = NSLock() private let streamDataReady = DispatchSemaphore(value: 0) + /// Advisory body length from the streaming response's Content-Length (-1 unknown). Only the + /// sequential-origin path reads it: a connection that ends short of it (or stalls out) is a + /// LOST source, not end-of-media, and must surface as a read error rather than EOF - the + /// consumer treats EOF as "played to the end" and deliberately never retries it. Guarded by + /// `streamLock`. + private var streamExpectedBytes: Int64 = -1 // MARK: - Persistent Mode (single forward-streaming connection, playback path) @@ -497,6 +503,13 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { /// reports EIO (-5) instead of EOF when the reconnect cap is hit. let isLive: Bool + /// `LoadOptions.sequentialOrigin` (via `DemuxerOpenProfile.avioSequentialOnly`): the origin + /// fabricates range answers, so only byte 0 is addressable. `open()` routes straight onto the + /// forward-only streaming mode (one unranged GET) and never issues a ranged request - no tail + /// prefetch, no optimistic persistent open, no size probe, no detours. `fileSize` stays -1 by + /// construction, which keeps `isStreaming` true and the pb non-seekable (#126 block below). + let sequentialOnly: Bool + /// Detour cache is VOD-only: live feeds have no meaningful random access and a /// non-authoritative size, so they stay on the unchanged reconnect path. private var detourEligible: Bool { !isLive && fileSize > 0 } @@ -541,13 +554,14 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { /// origin at once and the line used to name none of them. private let label: String - init(url: URL, extraHeaders: [String: String] = [:], label: String = "source", chunkSize: Int = 4 * 1024 * 1024, prefetchEnabled: Bool = true, isLive: Bool = false, chunkRequestTimeout: TimeInterval = 35, chunkMaxRetries: Int = 3, boundedInitialFetch: Int64? = nil, connStallTimeout: TimeInterval = AVIOReader.connStallTimeoutDefault, windowHighWater: Int? = nil) { + init(url: URL, extraHeaders: [String: String] = [:], label: String = "source", chunkSize: Int = 4 * 1024 * 1024, prefetchEnabled: Bool = true, isLive: Bool = false, chunkRequestTimeout: TimeInterval = 35, chunkMaxRetries: Int = 3, boundedInitialFetch: Int64? = nil, sequentialOnly: Bool = false, connStallTimeout: TimeInterval = AVIOReader.connStallTimeoutDefault, windowHighWater: Int? = nil) { self.url = url self.label = label self.extraHeaders = extraHeaders self.chunkSize = chunkSize self.prefetchEnabled = prefetchEnabled self.isLive = isLive + self.sequentialOnly = sequentialOnly self.chunkRequestTimeout = chunkRequestTimeout self.chunkMaxRetries = max(1, chunkMaxRetries) self.boundedInitialFetch = boundedInitialFetch.map { max(1, $0) } @@ -603,7 +617,17 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { context = ctx - if prefetchEnabled { + if sequentialOnly { + // Sequential origin: the only trustworthy request shape is one unranged GET from + // byte 0. Everything the branches below would issue on top - the suffix tail + // prefetch, the optimistic `Range: bytes=0-` persistent open, the dedicated size + // probe - is a ranged request the origin would answer with fabricated positions, + // so none of it runs. `fileSize` stays -1: `isStreaming` routes read()/seek() + // onto the forward-only streaming path and the #126 block below marks the pb + // non-seekable. + startStreamingDownload() + _ = streamDataReady.wait(timeout: .now() + .seconds(15)) + } else if prefetchEnabled { // #281: the parse seeks that follow this open are what the retained head exists for. winCond.lock() openPhaseActive = true @@ -1098,7 +1122,27 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { } } - return totalRead > 0 ? Int32(totalRead) : FFmpegErr.eof + if totalRead > 0 { return Int32(totalRead) } + if sequentialOnly { + streamLock.lock() + let ended = streamEnded + let received = streamBytesRead + Int64(streamBuffer.count) + let expected = streamExpectedBytes + streamLock.unlock() + // A sequential origin cannot be resumed at an offset, so a stalled-out wait or a + // body that ended short of its advisory length is a LOST source: report EIO so the + // pump exits on a read error the session can surface. EOF here would read as + // end-of-media, which the consumer deliberately never retries. + if !ended || (expected > 0 && received < expected) { + EngineLog.emit( + "[AVIOReader] sequential stream \(ended ? "ended short" : "stalled out") at " + + "\(received)\(expected > 0 ? "/\(expected)" : "") bytes; reporting EIO", + category: .demux + ) + return AVERROR_EIO_VALUE + } + } + return FFmpegErr.eof } // MARK: - Persistent Read (single forward-streaming connection) @@ -2382,7 +2426,19 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { let semaphore = DispatchSemaphore(value: 0) - let delegate = StreamingDelegate { [weak self] data in + let delegate = StreamingDelegate( + extraHeaders: extraHeaders, + onResponse: { [weak self] response in + // Advisory length for the sequential-origin EOF/EIO distinction; -1 (chunked / + // unknown) leaves the clean-end path as the only EOF source. + guard let self, self.sequentialOnly else { return } + let expected = response.expectedContentLength + guard expected > 0 else { return } + self.streamLock.lock() + self.streamExpectedBytes = expected + self.streamLock.unlock() + } + ) { [weak self] data in guard let self, !self.isClosed else { return } self.streamLock.lock() self.streamBuffer.append(data) @@ -3206,12 +3262,46 @@ private final class ChunkFetchDelegate: NSObject, URLSessionDataDelegate, @unche private final class StreamingDelegate: NSObject, URLSessionDataDelegate { let onData: @Sendable (Data) -> Void let onComplete: @Sendable () -> Void + /// Response hook (advisory Content-Length capture on the sequential-origin path). + let onResponse: (@Sendable (URLResponse) -> Void)? + /// Re-applied across cross-host redirects like every other delegate in this file; + /// IPTV origins routinely 302 twice (portal -> panel -> archive host) and the final + /// host must still see the caller's User-Agent / auth headers. + let extraHeaders: [String: String] - init(onData: @escaping @Sendable (Data) -> Void, onComplete: @escaping @Sendable () -> Void) { + init( + extraHeaders: [String: String] = [:], + onResponse: (@Sendable (URLResponse) -> Void)? = nil, + onData: @escaping @Sendable (Data) -> Void, + onComplete: @escaping @Sendable () -> Void + ) { + self.extraHeaders = extraHeaders + self.onResponse = onResponse self.onData = onData self.onComplete = onComplete } + func urlSession( + _ session: URLSession, + task: URLSessionTask, + willPerformHTTPRedirection response: HTTPURLResponse, + newRequest request: URLRequest, + completionHandler: @escaping (URLRequest?) -> Void + ) { + completionHandler(redirectPreservingHeaders( + task: task, newRequest: request, extraHeaders: extraHeaders)) + } + + func urlSession( + _ session: URLSession, + dataTask: URLSessionDataTask, + didReceive response: URLResponse, + completionHandler: @escaping (URLSession.ResponseDisposition) -> Void + ) { + onResponse?(response) + completionHandler(.allow) + } + func urlSession(_ session: URLSession, dataTask: URLSessionDataTask, didReceive data: Data) { onData(data) } diff --git a/Sources/AetherEngine/Demuxer/Demuxer.swift b/Sources/AetherEngine/Demuxer/Demuxer.swift index bade692c..89427ae7 100644 --- a/Sources/AetherEngine/Demuxer/Demuxer.swift +++ b/Sources/AetherEngine/Demuxer/Demuxer.swift @@ -41,6 +41,17 @@ struct DemuxerOpenProfile: Sendable { /// ~300 ms). nil keeps the open-ended behaviour for every other path (playback streams from 0). var boundedInitialFetch: Int64? = nil + /// `LoadOptions.sequentialOrigin`: the origin fabricates range answers, so the AVIO reader must + /// run its forward-only streaming mode (one unranged GET from byte 0) and never issue a ranged + /// request. Lives in the profile so the probe demuxer, the session demuxer, and every fresh + /// reopen (wedge restart, revive) inherit it together. + var avioSequentialOnly: Bool = false + + /// `LoadOptions.declaredDurationSeconds`: caller-trusted duration override consumed by + /// `Demuxer.duration`. Rides in the profile next to `avioSequentialOnly` because the two are a + /// pair: without the ranged tail read the container resolves no duration of its own. + var declaredDurationSeconds: Double? = nil + /// #240: what to call this demuxer's network reader in the log. Several readers run against the /// same origin at once (the pump, the subtitle forward prefetcher, the native subtitle readers), /// and a connection line without a name cannot say which one opened it: the reporter of #240 read @@ -89,6 +100,16 @@ struct DemuxerOpenProfile: Sendable { return copy } + /// A copy of `self` carrying the sequential-origin declaration and its paired trusted + /// duration (no-op when `sequential` is false and `declaredDuration` is nil), in the style + /// of `withProbeBudget` so call sites can chain it onto their existing profile. + func withSequentialOrigin(_ sequential: Bool, declaredDuration: Double?) -> DemuxerOpenProfile { + var copy = self + copy.avioSequentialOnly = sequential + if let declaredDuration { copy.declaredDurationSeconds = declaredDuration } + return copy + } + /// Open profile for the #79 wedged-restart fresh reopen (#93 residual). The 44 s device /// restart was find_stream_info re-paying the FULL playback probe budget (50 MB / 60 s) /// over an already-starved link, so the reopen shrinks the budget instead of skipping the @@ -298,6 +319,19 @@ public final class Demuxer: @unchecked Sendable { return container } + /// Full duration-precedence chain: a caller-declared duration + /// (`DemuxerOpenProfile.declaredDurationSeconds`, the sequential-origin pair) outranks + /// everything - the non-seekable pb ran no tail estimate, so the container value is 0 or + /// garbage from fabricated range data - then a custom time-seekable reader's own duration, + /// then the disc/container resolution above. + static func effectiveDurationSeconds( + declared: Double?, readerDuration: Double?, discTitle: Double?, container: Double + ) -> Double { + if let declared, declared > 0 { return declared } + if let readerDuration, readerDuration > 0 { return readerDuration } + return effectiveDurationSeconds(discTitle: discTitle, container: container) + } + /// Open a media URL and probe its streams. /// - Parameters: /// - extraHeaders: Attached to every HTTP request (ignored for file:// URLs). @@ -396,7 +430,8 @@ public final class Demuxer: @unchecked Sendable { isLive: isLive, chunkRequestTimeout: openProfile.avioRequestTimeout, chunkMaxRetries: openProfile.avioMaxRetries, - boundedInitialFetch: openProfile.boundedInitialFetch + boundedInitialFetch: openProfile.boundedInitialFetch, + sequentialOnly: openProfile.avioSequentialOnly ) reader.onNetworkPhaseChanged = onNetworkPhaseChanged try openWithProvider(reader, isLive: isLive) @@ -425,7 +460,8 @@ public final class Demuxer: @unchecked Sendable { // URL is nil because pb is already set. var ctxPtr: UnsafeMutablePointer? = ctx var opts: OpaquePointer? = nil - Self.applyDemuxerOptions(&opts, isLive: isLive) + Self.applyDemuxerOptions(&opts, isLive: isLive, + skipDurationEstimate: openProfile.avioSequentialOnly) let ret = avformat_open_input(&ctxPtr, nil, inputFormat, &opts) av_dict_free(&opts) guard ret == 0 else { @@ -456,12 +492,15 @@ public final class Demuxer: @unchecked Sendable { /// (3.24 MB/s -> ~1.7 MB/s). Tried+reverted: +sortdts (worse RSS), +discardcorrupt /// (worse RSS), +igndts (AetherEngine#5: matroska still emits dts=0 on HEVC open-GOP /// CRA B-frames, NOPTS repair stack stayed load-bearing). - private static func applyDemuxerOptions(_ opts: inout OpaquePointer?, isLive: Bool = false) { + private static func applyDemuxerOptions(_ opts: inout OpaquePointer?, isLive: Bool = false, skipDurationEstimate: Bool = false) { av_dict_set(&opts, "fflags", "+genpts", 0) - if isLive { + if isLive || skipDurationEstimate { // Live sources have no Content-Length; stream-info pass seeks SEEK_END, // which latches pb->eof_reached and collapses av_read_frame ~10s in. // skip_estimate_duration_from_pts avoids that SEEK_END entirely. + // A sequential-origin VOD skips it for the same reason from the other + // side: its pb is non-seekable by declaration, and the caller supplies + // the duration (`declaredDurationSeconds`) the estimate would have fed. av_dict_set(&opts, "skip_estimate_duration_from_pts", "1", 0) } } @@ -555,18 +594,17 @@ public final class Demuxer: @unchecked Sendable { } var duration: Double { - if let duration = (avioProvider as? CustomIOReaderBridge)? - .timeSeekableReader?.mediaDuration, - duration > 0 { - return duration - } let container: Double = { guard let ctx = formatContext else { return 0 } let dur = ctx.pointee.duration return dur > 0 ? Double(dur) / Double(AV_TIME_BASE) : 0 }() - // A disc title's MPLS/IFO duration overrides FFmpeg's unreliable mpegts estimate (AE#105). - return Self.effectiveDurationSeconds(discTitle: selectedDiscTitleDurationSeconds, container: container) + // Declared (sequential-origin caller trust) > custom reader > disc MPLS/IFO (AE#105) > container. + return Self.effectiveDurationSeconds( + declared: openProfile.declaredDurationSeconds, + readerDuration: (avioProvider as? CustomIOReaderBridge)?.timeSeekableReader?.mediaDuration, + discTitle: selectedDiscTitleDurationSeconds, + container: container) } /// AVFormatContext.bit_rate in bps, or 0 if unknown. Used by diff --git a/Sources/AetherEngine/PlayerState.swift b/Sources/AetherEngine/PlayerState.swift index 16d25f09..e94ae90c 100644 --- a/Sources/AetherEngine/PlayerState.swift +++ b/Sources/AetherEngine/PlayerState.swift @@ -271,6 +271,27 @@ public struct LoadOptions: Sendable, Equatable { /// Preferred subtitle languages (ISO 639-1/2) used ONLY to choose which native WebVTT rendition is marked DEFAULT=YES in the master, so a host-selected legible track renders (AVKit hides a non-default legible selection as mute-only). Read back as `nativeSubtitleDefaultOrdinal`. Unlike `preferredSubtitleLanguages` this does NOT auto-activate the host-overlay subtitle path, so it won't double up with the native render. Default empty (Sodalite#32). public var nativeSubtitlePreferredLanguages: [String] = [] + /// The origin fabricates range answers: any `Range: bytes=X-` gets a plausible-looking + /// `206 Content-Range: bytes X-.../total`, but the body is positioned on a coarse internal + /// chunk boundary rather than byte X (IPTV timeshift/catch-up archives are the motivating + /// case; a device trace showed ~1.9 s of content lost at every 32 MB range rotation, heard + /// as a once-a-minute audio desync). Headers cannot expose the lie, so this is a caller + /// declaration, not a probe. Only byte 0 is addressable: the reader runs its forward-only + /// streaming mode on one long-lived unranged GET - no bounded-range windowing, no + /// suffix/tail probes, no detour fills, no byte-offset reconnects - and the demuxer's pb is + /// non-seekable, so byte seeking is unavailable and a dropped connection surfaces as a read + /// error (EOF would read as end-of-media) for the host to re-request. FFmpeg's tail-read + /// duration estimate is skipped with the rest of the ranged reads; pair with + /// `declaredDurationSeconds` on VOD or the load fails with `zeroDuration`. Default `false`. + public var sequentialOrigin: Bool = false + + /// Trusted media duration in seconds, overriding the container/estimate-derived value (same + /// trust family as the disc MPLS/IFO override, AE#105). Required alongside + /// `sequentialOrigin` for VOD sources: with the tail read gone the demuxer resolves no + /// duration, and the caller usually knows the real one (an IPTV catch-up request names its + /// window length outright). nil keeps the demuxer's own value. Default nil. + public var declaredDurationSeconds: Double? = nil + /// Caller-bounded demux probe budget in bytes, mapped to `AVFormatContext.probesize` for the main playback open. nil keeps the engine default (50 MB). A smaller value speeds `find_stream_info` on slow remote sources whose sparse streams (PGS, mjpeg cover art) would otherwise read to the full budget. An over-tight budget fails OPEN, not closed: `find_stream_info` still returns success with a logged warning, so the session loads with late-resolving tracks silently missing rather than throwing a load error. The value is written to the context verbatim (FFmpeg's AVOption floor of 32 is bypassed), so validate track presence after load if you set this aggressively. The routing `probe(url:)` API and still extraction keep the full budget; the embedded subtitle side-demuxer caps its own probe (it only needs codec ids, not resolved sparse tracks) and tightens to this value when it is smaller (#76). Default nil (#68). public var probesize: Int64? @@ -374,6 +395,8 @@ public struct LoadOptions: Sendable, Equatable { eagerNativeSubtitleReaders: Bool = false, confirmAtmos: Bool = false, nativeSubtitlePreferredLanguages: [String] = [], + sequentialOrigin: Bool = false, + declaredDurationSeconds: Double? = nil, probesize: Int64? = nil, maxAnalyzeDuration: Int64? = nil, preferredAudioLanguages: [String] = [], @@ -404,6 +427,8 @@ public struct LoadOptions: Sendable, Equatable { self.eagerNativeSubtitleReaders = eagerNativeSubtitleReaders self.confirmAtmos = confirmAtmos self.nativeSubtitlePreferredLanguages = nativeSubtitlePreferredLanguages + self.sequentialOrigin = sequentialOrigin + self.declaredDurationSeconds = declaredDurationSeconds self.probesize = probesize self.maxAnalyzeDuration = maxAnalyzeDuration self.preferredAudioLanguages = preferredAudioLanguages diff --git a/Sources/AetherEngine/Video/HLSVideoEngine+LiveReopen.swift b/Sources/AetherEngine/Video/HLSVideoEngine+LiveReopen.swift index ffe8898d..24426d3f 100644 --- a/Sources/AetherEngine/Video/HLSVideoEngine+LiveReopen.swift +++ b/Sources/AetherEngine/Video/HLSVideoEngine+LiveReopen.swift @@ -129,7 +129,18 @@ extension HLSVideoEngine { // died with the pump and the provider's restart escalation judged by index distance alone, // so the tail request parked 30 s at a time into -12889 (rrgomes' seg719 trace). if case .readError(let code) = reason, !isLiveSession { - if Self.shouldReviveVODAfterReadError( + // A sequential origin admits no revive: the fresh demuxer can only reopen from byte 0 + // and then fails its anchor seek on the non-seekable pb, burning a connection slot on + // origins that are typically connection-capped. Surface the loss to the host, whose + // re-request (a fresh load) is the real recovery path. + if sequentialOrigin { + EngineLog.emit( + "[HLSVideoEngine] sequential-origin VOD pump died (readError \(code)); " + + "revive cannot resume at an offset, surfacing source failure", + category: .session + ) + onVODSourceFailed?(code) + } else if Self.shouldReviveVODAfterReadError( isLive: isLiveSession, packetsWritten: prod.packetsWrittenCount, cachedSegments: cache?.count ?? 0 diff --git a/Sources/AetherEngine/Video/HLSVideoEngine.swift b/Sources/AetherEngine/Video/HLSVideoEngine.swift index 4b021cad..99cfeed2 100644 --- a/Sources/AetherEngine/Video/HLSVideoEngine.swift +++ b/Sources/AetherEngine/Video/HLSVideoEngine.swift @@ -638,14 +638,19 @@ public final class HLSVideoEngine: @unchecked Sendable { companionAudioReader: IOReader? = nil, probesize: Int64? = nil, maxAnalyzeDuration: Int64? = nil, + sequentialOrigin: Bool = false, + declaredDurationSeconds: Double? = nil, forwardBufferSegments: Int? = nil ) { self.sourceURL = url self.sourceHTTPHeaders = sourceHTTPHeaders + self.sequentialOrigin = sequentialOrigin + self.declaredDurationSeconds = declaredDurationSeconds // Caller-bounded find_stream_info budget (#68); nil keeps the .playback default. Applied only to the // fallback open / live reopen here; the happy path reuses the already-budgeted preopenedDemuxer. self.openProfile = DemuxerOpenProfile.playback.withProbeBudget( probesize: probesize, maxAnalyzeDuration: maxAnalyzeDuration) + .withSequentialOrigin(sequentialOrigin, declaredDuration: declaredDurationSeconds) self.dvModeAvailable = dvModeAvailable self.displaySupportsHDR = displaySupportsHDR self.keepDvh1TagWithoutDV = keepDvh1TagWithoutDV @@ -757,6 +762,16 @@ public final class HLSVideoEngine: @unchecked Sendable { /// `+LiveReopen` extension, so it cannot be file-private. let openProfile: DemuxerOpenProfile + /// `LoadOptions.sequentialOrigin` for this session. Gates the VOD readError revive + /// (`+LiveReopen`): a revive's fresh demuxer can only reopen from byte 0 and then fails its + /// anchor seek on the non-seekable pb, burning a connection slot on origins that are typically + /// connection-capped, so the session surfaces `onVODSourceFailed` directly instead. + let sequentialOrigin: Bool + + /// `LoadOptions.declaredDurationSeconds`, threaded into every fresh-demuxer profile this + /// session builds (wedge restart) so a reopened demuxer reports the same trusted duration. + let declaredDurationSeconds: Double? + // MARK: - Public API @@ -2199,7 +2214,13 @@ public final class HLSVideoEngine: @unchecked Sendable { // .restartReopen: bounded find_stream_info budget; the FULL playback budget was // the bulk of a 44 s wedge-reopen over WAN (#93 residual). The pass itself must // run so video_delay resolves, else B-frame dts arrive broken (#93 judder). - try fresh.open(url: sourceURL, extraHeaders: sourceHTTPHeaders, profile: .restartReopen, isLive: false) + // A sequential origin keeps its declaration on the fresh open too: a ranged + // reopen would splice fabricated-position bytes into the new pump. + try fresh.open( + url: sourceURL, extraHeaders: sourceHTTPHeaders, + profile: DemuxerOpenProfile.restartReopen + .withSequentialOrigin(sequentialOrigin, declaredDuration: declaredDurationSeconds), + isLive: false) dem.markClosed() // abort any wedged read now that the replacement is ready freshDemuxer = fresh activeDem = fresh diff --git a/Sources/aetherctl/PlaybackCmd.swift b/Sources/aetherctl/PlaybackCmd.swift index 609e4374..bf09dd14 100644 --- a/Sources/aetherctl/PlaybackCmd.swift +++ b/Sources/aetherctl/PlaybackCmd.swift @@ -12,7 +12,8 @@ import AetherEngine /// plays" reports and for live teletext end-to-end validation (#107). func runPlay(url: URL, seconds: Double, live: Bool, nativeHLS: Bool = false, dvrWindow: Double?, subsPick: String?, hostCalls: [String], audioStats: Bool = false, seekEvery: Double? = nil, seekPattern: [Double] = [], startPosition: Double? = nil, mallocCensus: Bool = false, forceSoftware: Bool = false, censusThresholdMB: Int? = nil, censusHz: Double? = nil, frameTimes: Bool = false, - sidecars: [ExternalSubtitleTrack] = []) -> Int32 { + sidecars: [ExternalSubtitleTrack] = [], + sequentialOrigin: Bool = false, declaredDuration: Double? = nil) -> Int32 { EngineLog.handler = { print($0) } if mallocCensus { AetherEngine.setLargeAllocationCensusEnabled( @@ -26,7 +27,7 @@ func runPlay(url: URL, seconds: Double, live: Bool, nativeHLS: Bool = false, dvr // CFRunLoopRun, not a blocking semaphore: AetherEngine is @MainActor, so parking the main thread would deadlock the executor. let box = UncheckedBox(nil) Task { @MainActor in - box.value = await playSmokeTest(url: url, seconds: seconds, live: live, nativeHLS: nativeHLS, dvrWindow: dvrWindow, subsPick: subsPick, hostCalls: hostCalls, audioStats: audioStats, seekEvery: seekEvery, seekPattern: seekPattern, startPosition: startPosition, frameTimes: frameTimes, sidecars: sidecars) + box.value = await playSmokeTest(url: url, seconds: seconds, live: live, nativeHLS: nativeHLS, dvrWindow: dvrWindow, subsPick: subsPick, hostCalls: hostCalls, audioStats: audioStats, seekEvery: seekEvery, seekPattern: seekPattern, startPosition: startPosition, frameTimes: frameTimes, sidecars: sidecars, sequentialOrigin: sequentialOrigin, declaredDuration: declaredDuration) CFRunLoopStop(CFRunLoopGetMain()) } CFRunLoopRun() @@ -204,7 +205,7 @@ private func seekIntentDrill( } @MainActor -private func playSmokeTest(url: URL, seconds: Double, live: Bool, nativeHLS: Bool = false, dvrWindow: Double?, subsPick: String?, hostCalls: [String], audioStats: Bool, seekEvery: Double? = nil, seekPattern: [Double] = [], startPosition: Double? = nil, frameTimes: Bool = false, sidecars: [ExternalSubtitleTrack] = []) async -> Int32 { +private func playSmokeTest(url: URL, seconds: Double, live: Bool, nativeHLS: Bool = false, dvrWindow: Double?, subsPick: String?, hostCalls: [String], audioStats: Bool, seekEvery: Double? = nil, seekPattern: [Double] = [], startPosition: Double? = nil, frameTimes: Bool = false, sidecars: [ExternalSubtitleTrack] = [], sequentialOrigin: Bool = false, declaredDuration: Double? = nil) async -> Int32 { let engine: AetherEngine do { engine = try AetherEngine() @@ -259,6 +260,8 @@ private func playSmokeTest(url: URL, seconds: Double, live: Bool, nativeHLS: Boo isLive: live, dvrWindowSeconds: dvrWindow, nativeRemoteHLS: nativeHLS, + sequentialOrigin: sequentialOrigin, + declaredDurationSeconds: declaredDuration, externalSubtitles: sidecars ) // #311: installed BEFORE the load on purpose. The engine holds it and arms the host it builds, diff --git a/Sources/aetherctl/main.swift b/Sources/aetherctl/main.swift index da68d86b..49013ce7 100644 --- a/Sources/aetherctl/main.swift +++ b/Sources/aetherctl/main.swift @@ -71,13 +71,15 @@ func printUsage() { aetherctl validate [--no-dv] aetherctl swdecode [--frames N] aetherctl play [--seconds N] [--live] [--dvr-window N] [--subs ] - [--start-position S] + [--start-position S] [--sequential-origin] [--declared-duration S] [--audio-stats] [--host-calls play,extractor,setrate,reloadlive,seekback] (full load+play session smoke test; --subs activates the first matching embedded subtitle track and logs overlay cues; --audio-stats taps decoded PCM and prints per-second audio lead plus PTS-continuity gaps; seekback rewinds 20 s at t=15 and - returns to the live edge at t=30) + returns to the live edge at t=30; --sequential-origin declares a + fake-range origin (one unranged GET, no ranged probes) and needs + --declared-duration on VOD since the tail estimate is skipped) aetherctl segverify [--from N] [--count K] [--no-dv] [--dump ] (#92: SW-decode each segment in isolation; framesDecoded==0 => not independent) aetherctl disc-inspect @@ -471,6 +473,11 @@ if first == "play" { // Resume anchor, the same one load(startPosition:) takes. AE#287 needs it: the reporter's hard // park only reproduces when a rebuilt session opens exactly at the video-exhaustion boundary. let playStartPosition = takeDoubleFlag("--start-position", from: &rest) + // Sequential-origin declaration (LoadOptions.sequentialOrigin): fake-range archives get one + // unranged GET and no ranged probes; pair with --declared-duration on VOD because the tail + // duration estimate is skipped along with the other ranged reads. + let sequentialOrigin = takeFlag("--sequential-origin", from: &rest) + let declaredDuration = takeDoubleFlag("--declared-duration", from: &rest) // #311: install the software frame-time observer and read the presentation timebase, so the // per-frame boundaries and the clock a host would pace an overlay against are both observable. let frameTimes = takeFlag("--frame-times", from: &rest) @@ -499,7 +506,8 @@ if first == "play" { exit(64) } exit(runPlay(url: parseSourceURL(urlArg), seconds: seconds, live: live, nativeHLS: nativeHLS, dvrWindow: dvrWindow, subsPick: subsPick, hostCalls: hostCalls, audioStats: audioStats, seekEvery: seekEvery, seekPattern: seekPattern, startPosition: playStartPosition, mallocCensus: mallocCensus, forceSoftware: playForceSW, - censusThresholdMB: censusThresholdMB, censusHz: censusHz, frameTimes: frameTimes, sidecars: sidecars)) + censusThresholdMB: censusThresholdMB, censusHz: censusHz, frameTimes: frameTimes, sidecars: sidecars, + sequentialOrigin: sequentialOrigin, declaredDuration: declaredDuration)) } if ["probe", "serve", "validate", "swdecode", "extract", "audio", "customio"].contains(first) { diff --git a/Tests/AetherEngineTests/SequentialOriginReaderTests.swift b/Tests/AetherEngineTests/SequentialOriginReaderTests.swift new file mode 100644 index 00000000..b6747c33 --- /dev/null +++ b/Tests/AetherEngineTests/SequentialOriginReaderTests.swift @@ -0,0 +1,138 @@ +import Testing +import Foundation +@testable import AetherEngine + +/// `LoadOptions.sequentialOrigin`: origins that fabricate range answers (IPTV timeshift archives +/// answer any `Range: bytes=X-` with a plausible 206 whose body actually sits on a coarse internal +/// chunk boundary; ~1.9 s of content vanished at every 32 MB range rotation, heard as a +/// once-a-minute audio desync). Headers cannot expose the lie, so the caller declares it and the +/// reader must run ONE unranged GET from byte 0 - no bounded-range windowing, no suffix/tail +/// probe, no size probe, no detour fills - and must report a lost connection as EIO, never EOF +/// (the consumer treats EOF as played-to-the-end and deliberately never retries it). +@Suite("Sequential-origin reader") +struct SequentialOriginReaderTests { + + private func drain(_ reader: AVIOReader, upTo target: Int64, chunk: Int = 256 * 1024) -> (read: Int64, lastReturn: Int32) { + let buf = UnsafeMutablePointer.allocate(capacity: chunk) + defer { buf.deallocate() } + var read: Int64 = 0 + var last: Int32 = 0 + while read < target { + last = reader.read(into: buf, size: Int32(chunk)) + if last <= 0 { break } + read += Int64(last) + } + return (read, last) + } + + @Test("one unranged request serves a read well past the 32 MB rotation point") + func singleUnrangedConnection() throws { + let total: Int64 = 40 * 1024 * 1024 + let server = try #require(ThrottledOriginServer(totalSize: total, throttleUs: 500)) + defer { server.stop() } + let reader = AVIOReader(url: URL(string: "http://127.0.0.1:\(server.port)/archive.ts")!, + sequentialOnly: true) + defer { reader.markClosed(); reader.close() } + try reader.open() + + // Read past where the persistent reader would have rotated its first 32 MB range. + let (read, _) = drain(reader, upTo: 36 * 1024 * 1024) + #expect(read >= 36 * 1024 * 1024) + + // The whole contract: exactly one data request, and it carried no Range header at all + // (which also proves no suffix/tail probe and no size probe ever went out - any of + // those would be a second, ranged, request). + #expect(server.requestLog.count == 1) + #expect(server.rangeHeaderPresence == [false]) + } + + @Test("a body that ends short of its Content-Length reports EIO, not EOF") + func shortBodyReportsEIO() throws { + let total: Int64 = 8 * 1024 * 1024 + let maybeServer = ThrottledOriginServer(totalSize: total, throttleUs: 0, + respond: { _, _, _ in .serveThenDrop(afterBytes: 2 * 1024 * 1024) }) + let server = try #require(maybeServer) + defer { server.stop() } + let reader = AVIOReader(url: URL(string: "http://127.0.0.1:\(server.port)/archive.ts")!, + sequentialOnly: true) + defer { reader.markClosed(); reader.close() } + try reader.open() + + let (read, last) = drain(reader, upTo: total) + #expect(read >= 1 * 1024 * 1024) + #expect(read < total) + // AVERROR(EIO) = -5: a sequential origin cannot be resumed at an offset, so the loss must + // surface as a read error the session can act on. FFmpegErr.eof here would read as + // end-of-media 75 % early and the consumer would never retry it. + #expect(last == -5) + #expect(last != FFmpegErr.eof) + } + + @Test("a complete body still ends in clean EOF") + func completeBodyReportsEOF() throws { + let total: Int64 = 4 * 1024 * 1024 + let server = try #require(ThrottledOriginServer(totalSize: total, throttleUs: 0)) + defer { server.stop() } + let reader = AVIOReader(url: URL(string: "http://127.0.0.1:\(server.port)/archive.ts")!, + sequentialOnly: true) + defer { reader.markClosed(); reader.close() } + try reader.open() + + // Ask for more than the file holds so the final read hits the end-of-stream path. + let (read, last) = drain(reader, upTo: total + 1024) + #expect(read == total) + #expect(last == FFmpegErr.eof) + } + + // MARK: - Profile plumbing + + @Test("withSequentialOrigin carries the pair and touches nothing else") + func profileCopyHelper() { + let base = DemuxerOpenProfile.restartReopen + let p = base.withSequentialOrigin(true, declaredDuration: 8100) + #expect(p.avioSequentialOnly == true) + #expect(p.declaredDurationSeconds == 8100) + #expect(p.probesize == base.probesize) + #expect(p.maxAnalyzeDuration == base.maxAnalyzeDuration) + #expect(p.boundedInitialFetch == base.boundedInitialFetch) + #expect(p.readerLabel == base.readerLabel) + + let off = base.withSequentialOrigin(false, declaredDuration: nil) + #expect(off.avioSequentialOnly == false) + #expect(off.declaredDurationSeconds == nil) + } + + @Test("playback profile defaults to ranged mode") + func playbackDefaultsOff() { + #expect(DemuxerOpenProfile.playback.avioSequentialOnly == false) + #expect(DemuxerOpenProfile.playback.declaredDurationSeconds == nil) + let opts = LoadOptions() + #expect(opts.sequentialOrigin == false) + #expect(opts.declaredDurationSeconds == nil) + } + + // MARK: - Duration precedence + + @Test("a declared duration outranks reader, disc and container values") + func declaredDurationWins() { + // The motivating field case: a 135-min timeshift window whose fabricated tail read + // produced a 9.5 h container estimate. + #expect(Demuxer.effectiveDurationSeconds( + declared: 8100, readerDuration: nil, discTitle: nil, container: 34_213.1) == 8100) + #expect(Demuxer.effectiveDurationSeconds( + declared: 8100, readerDuration: 500, discTitle: 42, container: 34_213.1) == 8100) + } + + @Test("without a declared duration the existing chain is untouched") + func declaredNilKeepsChain() { + #expect(Demuxer.effectiveDurationSeconds( + declared: nil, readerDuration: 500, discTitle: 42, container: 7508) == 500) + #expect(Demuxer.effectiveDurationSeconds( + declared: nil, readerDuration: nil, discTitle: 42, container: 7508) == 42) + #expect(Demuxer.effectiveDurationSeconds( + declared: nil, readerDuration: nil, discTitle: nil, container: 7508) == 7508) + // 0/negative declared values are ignored, not honored. + #expect(Demuxer.effectiveDurationSeconds( + declared: 0, readerDuration: nil, discTitle: nil, container: 7508) == 7508) + } +} diff --git a/Tests/AetherEngineTests/ThrottledOriginServer.swift b/Tests/AetherEngineTests/ThrottledOriginServer.swift index dbd6436c..71d820ea 100644 --- a/Tests/AetherEngineTests/ThrottledOriginServer.swift +++ b/Tests/AetherEngineTests/ThrottledOriginServer.swift @@ -25,6 +25,11 @@ final class ThrottledOriginServer: @unchecked Sendable { /// `afterBytes: 0` is the headers-but-no-body variant, i.e. a generation that never sees a /// first byte. case serveThenGoSilent(afterBytes: Int64) + /// Sequential-origin drop shape: answer the 206 header promising the full remaining body, + /// deliver `afterBytes`, then close the socket outright. The client sees a connection that + /// ended SHORT of its Content-Length - the observable behind the sequential reader's + /// EIO-not-EOF distinction (a lost source must not read as end-of-media). + case serveThenDrop(afterBytes: Int64) } let port: UInt16 @@ -40,6 +45,7 @@ final class ThrottledOriginServer: @unchecked Sendable { private var _stopped = false private var _requestedRanges: [(start: Int64, end: Int64?)] = [] private var _requestLog: [(path: String, start: Int64, end: Int64?)] = [] + private var _rangeHeaderPresent: [Bool] = [] var bytesWritten: Int64 { lock.lock(); defer { lock.unlock() } @@ -65,6 +71,14 @@ final class ThrottledOriginServer: @unchecked Sendable { return _requestLog } + /// Whether each logged request carried a Range header at all. A range-less GET is logged in + /// `requestLog` as (start 0, end nil), indistinguishable from `bytes=0-`; the sequential-origin + /// reader's whole contract is that it never sends Range, so its tests assert on THIS. + var rangeHeaderPresence: [Bool] { + lock.lock(); defer { lock.unlock() } + return _rangeHeaderPresent + } + private var stopped: Bool { lock.lock(); defer { lock.unlock() } return _stopped @@ -174,10 +188,12 @@ final class ThrottledOriginServer: @unchecked Sendable { var offset: Int64 = 0 var rangeEnd: Int64? = nil var isSuffix = false + var hadRangeHeader = false if let rangeLine = request.components(separatedBy: "\r\n") .first(where: { $0.lowercased().hasPrefix("range:") }), let eq = rangeLine.range(of: "bytes="), let dash = rangeLine.range(of: "-", range: eq.upperBound..= dropAfter { + lock.lock() + _connFDs.removeAll { $0 == fd } + lock.unlock() + shutdown(fd, SHUT_RDWR) + close(fd) + return false + } var n = Int(min(Int64(chunkBytes), remaining - served)) if let silentAfter { n = Int(min(Int64(n), silentAfter - served)) } + if let dropAfter { n = Int(min(Int64(n), dropAfter - served)) } guard writeBody(fd, Array(chunk[0.. 0 { usleep(throttleUs) } From 0742e617f5a7f6f3ef61e73d70d1310e6046f1ff Mon Sep 17 00:00:00 2001 From: Andreas Busslinger Date: Sun, 9 Aug 2026 12:13:01 +0200 Subject: [PATCH 2/4] fix(routing): a sequential origin keeps the native path The forward-only rule forced every sequential-origin archive onto the software path, trading AVPlayer's buffering and hardware decode for nothing: the reasons the rule exists (cue prewarm, segment seeks, size probes) are all already answered for an explicit sequential declaration - caller-declared duration, uniform-stride plan, gated restarts, a prewarm that fails fast on the non-seekable pb - and the producer reads the archive linearly from byte 0 exactly like a live session does. Device comparison on the same channel: the 720p50 live stream on the native path plays clean while the forced-software replay of the same provider visibly stutters with every queue metric reading healthy. Declared-interlaced archives still route software via the field-order policy (the #232 refute probe stays seekable-only: its sample cannot be rewound on a sequential reader), and the #126 accidental forward-only case (unknown-length HTTP MP4) keeps the software fallback - nothing declared its timeline trustworthy. Co-Authored-By: Claude Fable 5 --- Sources/AetherEngine/AetherEngine.swift | 22 ++++++++++++++++++++-- 1 file changed, 20 insertions(+), 2 deletions(-) diff --git a/Sources/AetherEngine/AetherEngine.swift b/Sources/AetherEngine/AetherEngine.swift index 800499a6..8a27ad7c 100644 --- a/Sources/AetherEngine/AetherEngine.swift +++ b/Sources/AetherEngine/AetherEngine.swift @@ -2998,9 +2998,27 @@ public final class AetherEngine: ObservableObject { // the forward-only streaming reader (#126: unknown-length HTTP MP4 produced zero segments). // Live sources are exempt: the live producer never seeks backward, scrub previews come from // the DVR segment cache, and audio-switch is already no-op for forward-only sources. + // An explicit sequential-origin declaration is exempt for the same reasons a live session + // is: the producer reads the archive linearly from byte 0, the duration is caller-declared + // (no tail estimate), the segment plan is the uniform-stride fallback, restarts/revives are + // gated off, and the cue prewarm fails fast on the non-seekable pb. Forcing those archives + // onto the software path traded AVPlayer's buffering and hardware decode for nothing + // (device trace: a 720p50 timeshift archive played clean on the native path and visibly + // stuttered on the software one). Declared-interlaced archives still route software via + // the field-order policy above - the #232 refute probe cannot run without a rewind. if !probe.isSourceSeekable && !options.isLive { - useSoftwarePath = true - EngineLog.emit("[AetherEngine] source is forward-only, forcing software path", category: .engine) + if options.sequentialOrigin { + if !useSoftwarePath { + EngineLog.emit( + "[AetherEngine] sequential origin keeps the native path (linear read, " + + "declared duration, seeks unavailable)", + category: .engine + ) + } + } else { + useSoftwarePath = true + EngineLog.emit("[AetherEngine] source is forward-only, forcing software path", category: .engine) + } } // TEST-ONLY: forces SW path for aetherctl live --sw; unset in shipping builds. if Self.forceSoftwarePathForTesting { From 0d72393c2c50efd211d8c33a28e700cbf05cae57 Mon Sep 17 00:00:00 2001 From: Andreas Busslinger Date: Sun, 9 Aug 2026 12:55:01 +0200 Subject: [PATCH 3/4] fix(hls): sequential archives serve an append playlist with real durations The static VOD plan advertises uniform EXTINF values the muxer cannot honor: cuts snap to real keyframes, so an archive whose GOP cadence does not divide the cut target puts every segment's media outside its advertised window (device trace: 1.92 s GOPs against the 4.0 s plan - eleven 3.84 s segments and one 5.76 s segment per 48 s cycle, all declared as 4.000; AVPlayer on tvOS 26 visibly jumped ~1.8 s at every resync while every queue metric read clean; a channel whose keyframes happen to sit on the 4 s grid played the same pipeline flawlessly). A sequential-origin session now renders its media playlist the way the live path always has - only finalized segments, each with the duration actually muxed - as an append-only EVENT playlist that completes with ENDLIST at true source EOF (a torn-down session never marks it). The producer records each VOD segment's item-axis start at the #65 ledger site and reports real durations at rotation; the provider grows the visible set contiguously and the server holds the first playlist until the startup segments exist, mirroring the live gate. Two host-facing seams from the growing playlist are closed at the engine: a caller-declared duration outranks the item's (the scrubber scales to the window length, not the produced span), and a sequential session with no resume anchor queues a post-readiness seek to 0 through the #127 replay, because AVPlayer defaults an EVENT playlist to edge-minus-holdback and re-anchors there after the load-time seek. Co-Authored-By: Claude Fable 5 --- .../AetherEngine/AetherEngine+Loading.swift | 18 +++- .../AetherEngine/Network/HLSLocalServer.swift | 11 +++ .../Video/HLSSegmentProducer.swift | 35 ++++++++ .../Video/HLSVideoEngine+LiveReopen.swift | 6 ++ .../AetherEngine/Video/HLSVideoEngine.swift | 9 ++ .../Video/VideoSegmentProvider.swift | 85 ++++++++++++++++++- 6 files changed, 162 insertions(+), 2 deletions(-) diff --git a/Sources/AetherEngine/AetherEngine+Loading.swift b/Sources/AetherEngine/AetherEngine+Loading.swift index 5248489b..1a6d9a60 100644 --- a/Sources/AetherEngine/AetherEngine+Loading.swift +++ b/Sources/AetherEngine/AetherEngine+Loading.swift @@ -126,7 +126,15 @@ extension AetherEngine { } duration .sink { [weak self] value in - if value > 0 { self?.duration = value } + guard let self else { return } + // A caller-declared duration outranks the host's: a sequential session's append + // playlist grows while it plays, so the item duration is the produced span, not + // the window length the host UI should scale its scrubber to. + if let declared = self.loadedOptions.declaredDurationSeconds, declared > 0 { + self.duration = declared + } else if value > 0 { + self.duration = value + } } .store(in: &cancellables) isReady @@ -1125,6 +1133,14 @@ extension AetherEngine { // forwardBufferDuration default (4 s): deep buffer lets AVPlayer race to the live edge and hit the transcode warm-up gap head-on (-12888); 4 s PACES consumption. Verified: 8 s worsened startup pause (8-10 s vs ~1 s). // Live REJOIN: skip initial seek so AVPlayer picks edge-minus-holdback instead; seek-to-0 against the re-served backlog wedged the reloaded item in waitingToPlay (device repro: tvOS 26, Jellyfin stream.ts). See LiveReloadPolicy. lastNativeVideoStartPosition = startPosition ?? 0 + // Sequential append playlist: AVPlayer treats the growing playlist as an EVENT and + // defaults to edge-minus-holdback (~6 s in on a fresh session, more once the producer + // has raced ahead). The load-time seek to 0 fires before readyToPlay and the item + // re-anchors to the edge default afterwards, so queue a post-readiness seek through + // the #127 replay instead - every segment stays retained, so 0 is always reachable. + if !isLive, loadedOptions.sequentialOrigin, startPosition == nil { + pendingPreReadySeekSeconds = 0.0 + } // AE#158: consume-and-reset so only the load() that armed the handover swaps in place; audio-switch // and recovery reloads keep their own contracts. let inPlaceHandover = pendingInPlaceItemHandover diff --git a/Sources/AetherEngine/Network/HLSLocalServer.swift b/Sources/AetherEngine/Network/HLSLocalServer.swift index ff514ebf..f43c4f36 100644 --- a/Sources/AetherEngine/Network/HLSLocalServer.swift +++ b/Sources/AetherEngine/Network/HLSLocalServer.swift @@ -93,6 +93,10 @@ protocol HLSSegmentProvider: AnyObject { /// LL-HLS blocking reload: block until segment at absolute index exists. Holds AVPlayer's ?_HLS_msn= reload open so it receives the new segment the instant it is cut, not a poll-interval late. func waitForLiveSegment(index: Int, timeout: TimeInterval) -> Bool + /// Sequential append playlist: block until the startup segments are finalized (or timeout). + /// Same rationale as the live gate - AVPlayer treats an empty first playlist as a broken asset. + func waitForSequentialStartupSegments(timeout: TimeInterval) -> Bool + /// Upper bound on how long a blocking reload may hold before the 503. Production providers derive /// it from the sealed TARGETDURATION (3 x TD, the HOLD-BACK depth) so a fastZap session (TD=2) /// times out in 6 s instead of 18 s — a hold that outlives AVPlayer's forward buffer guarantees @@ -133,6 +137,7 @@ extension HLSSegmentProvider { } func waitForFirstLiveSegment(timeout: TimeInterval) -> Bool { true } func waitForLiveSegment(index: Int, timeout: TimeInterval) -> Bool { true } + func waitForSequentialStartupSegments(timeout: TimeInterval) -> Bool { true } var liveBlockingReloadHoldSeconds: TimeInterval { 18.0 } func notePlaylistBuild() -> (visibleCount: Int, firstVisible: Int, refreshCounter: Int, endlistAdded: Bool, discontinuitySequence: Int) { return (visibleCount: segmentCount, firstVisible: 0, refreshCounter: 0, endlistAdded: false, discontinuitySequence: 0) @@ -700,6 +705,12 @@ final class HLSLocalServer: @unchecked Sendable { } else { _ = p.waitForFirstLiveSegment(timeout: 30.0) } + } else if let p = provider, p.playlistType == .event { + // Sequential append playlist: hold until the startup segments exist. A fast + // archive origin cuts them within ~a second; the timeout only covers a source + // that dies before its first cut (the playlist then renders empty and AVPlayer + // surfaces the failure instead of hanging). + _ = p.waitForSequentialStartupSegments(timeout: 30.0) } let body = buildMediaPlaylist() stateLock.lock() diff --git a/Sources/AetherEngine/Video/HLSSegmentProducer.swift b/Sources/AetherEngine/Video/HLSSegmentProducer.swift index 0e66b83b..af402054 100644 --- a/Sources/AetherEngine/Video/HLSSegmentProducer.swift +++ b/Sources/AetherEngine/Video/HLSSegmentProducer.swift @@ -174,6 +174,17 @@ final class HLSSegmentProducer: @unchecked Sendable { /// Fires synchronously on the pump thread per finalized live segment (index, duration, startSeconds, discontinuous). var onLiveSegmentFinalized: (@Sendable (Int, Double, Double, Bool) -> Void)? + /// Sequential-VOD twin of `onLiveSegmentFinalized` (index, real duration in seconds): feeds + /// the append playlist whose EXTINF must match the media actually muxed. Set only for + /// sequential-origin sessions; nil keeps the historical VOD behavior byte-identical. + var onSequentialSegmentFinalized: (@Sendable (Int, Double) -> Void)? + /// Fired once when the pump reaches true source EOF (not a stop/teardown): the append + /// playlist completes with ENDLIST. + var onSequentialSourceEnded: (@Sendable () -> Void)? + /// Item-axis start (seconds) per VOD segment, recorded at the #65 ledger site as each + /// segment opens; consumed by `reportSequentialSegmentFinalized`. Pump-thread only. + private var vodSegmentStartByIndex: [Int: Double] = [:] + /// Forward discontinuity threshold. Distinct from NOPTS-dts repair (+1 tick scale); only fires on genuine multi-second leaps. static let discontinuityThresholdSeconds: Double = 10.0 @@ -1460,6 +1471,9 @@ final class HLSSegmentProducer: @unchecked Sendable { if isLive { reportLiveSegmentFinalized(index: currentMuxerSegmentIndex, nextIndex: newIdx) + } else if onSequentialSegmentFinalized != nil { + reportSequentialSegmentFinalized(index: currentMuxerSegmentIndex, + nextIndex: newIdx) } // Cut succeeded but muxer failed to open the next staging fd: silently discards every subsequent byte. if muxer.isWedged { @@ -1510,6 +1524,22 @@ final class HLSSegmentProducer: @unchecked Sendable { return muxer } + /// Sequential-VOD finalize report: real duration = next segment's item-axis start minus this + /// one's, both recorded at the #65 ledger site. Falls back to the cut target when a start is + /// missing (NOPTS dts at the boundary) - one estimated EXTINF beats a stalled playlist. + private func reportSequentialSegmentFinalized(index: Int, nextIndex: Int?) { + let duration: Double + if let start = vodSegmentStartByIndex[index], + let nextIndex, let nextStart = vodSegmentStartByIndex[nextIndex], + nextStart > start { + duration = nextStart - start + } else { + duration = targetSegmentDurationSeconds + } + vodSegmentStartByIndex.removeValue(forKey: index) + onSequentialSegmentFinalized?(index, duration) + } + private func reportLiveSegmentFinalized(index: Int, nextIndex: Int?) { guard let startSeconds = liveSegmentStartByIndex[index] else { EngineLog.emit( @@ -1550,6 +1580,8 @@ final class HLSSegmentProducer: @unchecked Sendable { byteCount: result.bytesWritten) if isLive { reportLiveSegmentFinalized(index: idx, nextIndex: nil) + } else if onSequentialSegmentFinalized != nil { + reportSequentialSegmentFinalized(index: idx, nextIndex: nil) } } else { EngineLog.emit( @@ -2759,6 +2791,9 @@ final class HLSSegmentProducer: @unchecked Sendable { vodLedgerLastRoutedSeg = prevSeg let shiftTicks = videoShiftPts == Int64.min ? 0 : videoShiftPts let outDts = prev.pointee.dts + if onSequentialSegmentFinalized != nil { + vodSegmentStartByIndex[prevSeg] = Double(outDts) * sourceVideoTbSeconds + } let srcDts = outDts &+ shiftTicks let localI = prevSeg - baseIndex let planSrc: Int64? = (localI >= 0 && localI < segmentBoundaries.count) diff --git a/Sources/AetherEngine/Video/HLSVideoEngine+LiveReopen.swift b/Sources/AetherEngine/Video/HLSVideoEngine+LiveReopen.swift index 24426d3f..cc6509dd 100644 --- a/Sources/AetherEngine/Video/HLSVideoEngine+LiveReopen.swift +++ b/Sources/AetherEngine/Video/HLSVideoEngine+LiveReopen.swift @@ -170,6 +170,12 @@ extension HLSVideoEngine { handleVODGateStarvationExit(prod) return } + // Sequential append playlist: TRUE source EOF (not a stop, not a re-anchor) completes + // the playlist with ENDLIST so AVPlayer can reach end-of-media - a growing playlist + // without ENDLIST never ends. + if case .eof = reason, !isLiveSession, sequentialOrigin { + provider?.markSequentialEnded() + } guard isLiveSession else { return } let reopenTransport = Self.liveReopenTransport( sourceReopenableByURL: sourceReopenableByURL, diff --git a/Sources/AetherEngine/Video/HLSVideoEngine.swift b/Sources/AetherEngine/Video/HLSVideoEngine.swift index 99cfeed2..180b38d1 100644 --- a/Sources/AetherEngine/Video/HLSVideoEngine.swift +++ b/Sources/AetherEngine/Video/HLSVideoEngine.swift @@ -1414,6 +1414,11 @@ public final class HLSVideoEngine: @unchecked Sendable { hdcpLevel: hdcpLevel, sourceBitrate: sourceBitrate, isLive: isLiveSession, + // Sequential archives: playlist grows with the producer's REAL cut durations. The + // static plan's uniform EXTINF lies whenever the archive's GOP cadence does not + // divide the cut target (1.92 s GOPs vs a 4.0 s plan put every segment's media up + // to 1.9 s outside its advertised window; AVPlayer visibly jumped at each resync). + sequentialAppendPlaylist: sequentialOrigin && !isLiveSession, liveWindowSizing: LiveWindowSizing( targetSegmentDurationSeconds: liveCutTargetSeconds, dvrWindowSeconds: dvrWindowSeconds @@ -1454,6 +1459,10 @@ public final class HLSVideoEngine: @unchecked Sendable { durationSeconds: durationSeconds, discontinuous: discontinuous) } + } else if sequentialOrigin { + prod.onSequentialSegmentFinalized = { [weak prov] index, durationSeconds in + prov?.appendSequentialSegmentDuration(index: index, durationSeconds: durationSeconds) + } } EngineLog.emit( diff --git a/Sources/AetherEngine/Video/VideoSegmentProvider.swift b/Sources/AetherEngine/Video/VideoSegmentProvider.swift index 18e0ddd7..068df0b0 100644 --- a/Sources/AetherEngine/Video/VideoSegmentProvider.swift +++ b/Sources/AetherEngine/Video/VideoSegmentProvider.swift @@ -149,6 +149,8 @@ final class VideoSegmentProvider: HLSSegmentProvider, @unchecked Sendable { /// Immutable for VOD; grows under stateLock for live (producer appends via appendLiveSegment). private var segments: [HLSVideoEngine.Segment] private let isLive: Bool + /// Sequential-origin session: playlist grows with finalized real durations (see _seqDurations). + private let sequentialAppendPlaylist: Bool /// Drives both playlist firstVisible and cache eviction cutoff so they never drift. private let liveWindowSizing: LiveWindowSizing /// Only `.fastZap` sessions may serve a shallow first window after a bounded grace. @@ -269,6 +271,15 @@ final class VideoSegmentProvider: HLSSegmentProvider, @unchecked Sendable { /// One-shot latch for `noteWindowSlideRelativeToConsumer`. Guarded by stateLock. private var _liveConsumerOutsideWindowLatched = false + /// Sequential-origin append playlist: real EXTINF per finalized segment, index-aligned and + /// contiguous from 0. The static VOD plan's uniform durations are a lie for archives whose + /// GOP cadence does not divide the cut target (device trace: 1.92 s GOPs against a 4.0 s + /// plan put every segment's media up to 1.9 s outside its advertised window; AVPlayer + /// showed a content jump at every resync). Guarded by stateLock. + private var _seqDurations: [Double] = [] + /// Producer reached true source EOF: the next playlist build appends ENDLIST. Guarded by stateLock. + private var _seqEnded = false + init( cache: SegmentCache, segments: [HLSVideoEngine.Segment], @@ -280,6 +291,7 @@ final class VideoSegmentProvider: HLSSegmentProvider, @unchecked Sendable { hdcpLevel: String?, sourceBitrate: Int64, isLive: Bool = false, + sequentialAppendPlaylist: Bool = false, liveWindowSizing: LiveWindowSizing = LiveWindowSizing(targetSegmentDurationSeconds: 4.0, dvrWindowSeconds: nil), allowsBoundedDegradedStart: Bool = false, blockingReloadOverride: Bool? = nil, @@ -305,6 +317,7 @@ final class VideoSegmentProvider: HLSSegmentProvider, @unchecked Sendable { self.cache = cache self.segments = segments self.isLive = isLive + self.sequentialAppendPlaylist = sequentialAppendPlaylist self.liveWindowSizing = liveWindowSizing self.allowsBoundedDegradedStart = allowsBoundedDegradedStart self.blockingReloadOverride = blockingReloadOverride @@ -368,6 +381,56 @@ final class VideoSegmentProvider: HLSSegmentProvider, @unchecked Sendable { firstSegmentCondition.unlock() } + /// Append the real duration of a finalized sequential-VOD segment (index-contiguous from 0; + /// out-of-order appends are ignored like the live path's). The playlist's visible count and + /// EXTINF values follow these, so AVPlayer only ever sees segments whose advertised duration + /// matches the media inside them. + func appendSequentialSegmentDuration(index: Int, durationSeconds: Double) { + stateLock.lock() + guard index == _seqDurations.count else { + stateLock.unlock() + EngineLog.emit( + "[HLSVideoEngine] sequential segment append out of order: got index=\(index), " + + "expected \(_seqDurations.count); ignoring", + category: .session + ) + return + } + _seqDurations.append(max(0.001, durationSeconds)) + stateLock.unlock() + firstSegmentCondition.lock() + firstSegmentCondition.broadcast() + firstSegmentCondition.unlock() + } + + /// Producer reached true source EOF: the next playlist build renders as a completed VOD + /// asset (ENDLIST). Never called for a torn-down session, whose playlist simply stops + /// being served. + func markSequentialEnded() { + stateLock.lock() + _seqEnded = true + stateLock.unlock() + firstSegmentCondition.lock() + firstSegmentCondition.broadcast() + firstSegmentCondition.unlock() + } + + /// Hold the first media playlist until the startup segments exist (mirrors the live gate: + /// an empty playlist is a broken asset to AVPlayer, which never re-polls it). A fast + /// archive origin cuts the first segments within a second. + func waitForSequentialStartupSegments(timeout: TimeInterval) -> Bool { + let deadline = Date(timeIntervalSinceNow: timeout) + firstSegmentCondition.lock() + defer { firstSegmentCondition.unlock() } + while true { + stateLock.lock() + let ready = _seqDurations.count >= LiveEdgePolicy.minStartupSegments || _seqEnded + stateLock.unlock() + if ready { return true } + if !firstSegmentCondition.wait(until: deadline) { return false } + } + } + /// Called on each playlist build. For live: advances firstVisible to max(0, highWater - window), /// evicts cache below it, and increments _discontinuitySequence for each dropped discontinuous segment. /// VOD: returns full count so AVPlayer sees a complete asset (EVENT experiment that reported @@ -408,6 +471,13 @@ final class VideoSegmentProvider: HLSSegmentProvider, @unchecked Sendable { } return (total, _liveFirstVisible, refreshCounter, false, _discontinuitySequence) } + if sequentialAppendPlaylist { + // Only finalized segments are visible (their EXTINF is real); the playlist grows as + // the producer cuts and completes with ENDLIST at true source EOF. The plan bounds + // the count so a source running past the declared window cannot outgrow the asset. + let visible = min(_seqDurations.count, segments.count) + return (visible, 0, refreshCounter, _seqEnded, 0) + } return (segments.count, 0, refreshCounter, false, 0) } @@ -819,6 +889,14 @@ final class VideoSegmentProvider: HLSSegmentProvider, @unchecked Sendable { guard index >= 0, index < segments.count else { return 0 } return segments[index].durationSeconds } + if sequentialAppendPlaylist { + stateLock.lock() + defer { stateLock.unlock() } + if index >= 0, index < _seqDurations.count { return _seqDurations[index] } + // Not yet finalized (restart-mapping callers only; the playlist never shows these). + guard index >= 0, index < segments.count else { return 0 } + return segments[index].durationSeconds + } guard index >= 0, index < segments.count else { return 0 } return segments[index].durationSeconds } @@ -859,7 +937,12 @@ final class VideoSegmentProvider: HLSSegmentProvider, @unchecked Sendable { /// regardless of playlist type); side effects: Control Center showed "LIVE" (asset.duration NaN), /// replay-from-beginning landed ~2 min in. .live is the only spec-correct shape for a sliding window /// (EVENT forbids segment removal; VOD stops playback). VOD stays .vod. - var playlistType: HLSPlaylistType { isLive ? .live : .vod } + var playlistType: HLSPlaylistType { + if isLive { return .live } + // Sequential archives grow append-only with real durations and never remove segments - + // exactly EVENT's contract; the completed playlist (ENDLIST) renders as plain VOD. + return sequentialAppendPlaylist ? .event : .vod + } /// Stable TARGETDURATION from the first manifest; avoids -12888 startup race for high-bitrate live. var liveTargetSegmentDuration: Double? { isLive ? liveWindowSizing.targetSegmentDurationSeconds : nil From 3c0932e74c43918a40c9fa00a4ea16e804eabc1b Mon Sep 17 00:00:00 2001 From: Andreas Busslinger Date: Sun, 9 Aug 2026 13:10:13 +0200 Subject: [PATCH 4/4] fix(hls): pair sequential finalize reports order-independently The first cut of the append playlist still advertised the plan's 4.000: the muxer rotation can fire off an AUDIO packet crossing the boundary before the first video packet of the new segment has recorded its start, so the duration lookup at capture time found nothing and fell back to the cut target on every single segment. Durations are now computed at the ledger site (where the next start becomes known) and reports fire once BOTH halves - capture and duration - are in, in either order. Two more shapes the first version missed: a long GOP can skip plan indices outright (a 5.76 s cut spans two 4 s boundaries), so skipped holes report duration 0 and the playlist renderer omits zero-duration entries; and since capture, holes and the EOF tail can resolve out of order while the provider's append API is strictly contiguous, all reports funnel through a small reorder buffer. Verified against the failing archive: the served playlist now reads EXTINF 5.760 / 3.840 matching the muxed media exactly, TARGETDURATION 6. Co-Authored-By: Claude Fable 5 --- .../AetherEngine/Network/HLSLocalServer.swift | 3 + .../Video/HLSSegmentProducer.swift | 94 +++++++++++++++---- .../Video/VideoSegmentProvider.swift | 4 +- 3 files changed, 83 insertions(+), 18 deletions(-) diff --git a/Sources/AetherEngine/Network/HLSLocalServer.swift b/Sources/AetherEngine/Network/HLSLocalServer.swift index f43c4f36..1b764889 100644 --- a/Sources/AetherEngine/Network/HLSLocalServer.swift +++ b/Sources/AetherEngine/Network/HLSLocalServer.swift @@ -1354,6 +1354,9 @@ final class HLSLocalServer: @unchecked Sendable { lastInitVersion = v } let dur = provider.segmentDuration(at: i) + // A zero-duration entry is a plan index the producer skipped outright (sequential + // sessions: a long GOP spanning two boundaries); no media file exists for it. + guard dur > 0 else { continue } lines.append("#EXTINF:\(String(format: "%.3f", dur)),") lines.append(segURI(i)) } diff --git a/Sources/AetherEngine/Video/HLSSegmentProducer.swift b/Sources/AetherEngine/Video/HLSSegmentProducer.swift index af402054..70f1bb22 100644 --- a/Sources/AetherEngine/Video/HLSSegmentProducer.swift +++ b/Sources/AetherEngine/Video/HLSSegmentProducer.swift @@ -182,8 +182,35 @@ final class HLSSegmentProducer: @unchecked Sendable { /// playlist completes with ENDLIST. var onSequentialSourceEnded: (@Sendable () -> Void)? /// Item-axis start (seconds) per VOD segment, recorded at the #65 ledger site as each - /// segment opens; consumed by `reportSequentialSegmentFinalized`. Pump-thread only. + /// segment opens. Pump-thread only. private var vodSegmentStartByIndex: [Int: Double] = [:] + /// Capture and duration arrive in either order: the muxer rotation can fire off an AUDIO + /// packet crossing the boundary before the first video packet of the new segment has + /// recorded its start (the duration source). The report fires once BOTH are in. Pump-thread only. + private var seqCapturedAwaitingDuration: Set = [] + private var seqDurationAwaitingCapture: [Int: Double] = [:] + /// The most recent segment index the ledger recorded a start for: a long GOP can SKIP plan + /// indices entirely (a 5.76 s cut spans two 4 s boundaries), so duration pairing runs against + /// the last RECORDED index, and the skipped holes report duration 0 (the playlist renderer + /// omits zero-duration entries). Pump-thread only. + private var lastSeqLedgerSeg = Int.min + /// The provider's append API is strictly contiguous; captures, holes and the EOF tail can + /// resolve out of order, so reports funnel through this small reorder buffer. Pump-thread only. + private var seqNextReportIndex: Int? = nil + private var seqReadyReports: [Int: Double] = [:] + + /// Order-preserving funnel for sequential finalize reports. + private func emitSequentialReport(index: Int, duration: Double) { + let base = seqNextReportIndex ?? index + seqNextReportIndex = base + seqReadyReports[index] = duration + var next = base + while let d = seqReadyReports.removeValue(forKey: next) { + onSequentialSegmentFinalized?(next, d) + next += 1 + } + seqNextReportIndex = next + } /// Forward discontinuity threshold. Distinct from NOPTS-dts repair (+1 tick scale); only fires on genuine multi-second leaps. static let discontinuityThresholdSeconds: Double = 10.0 @@ -1472,8 +1499,7 @@ final class HLSSegmentProducer: @unchecked Sendable { reportLiveSegmentFinalized(index: currentMuxerSegmentIndex, nextIndex: newIdx) } else if onSequentialSegmentFinalized != nil { - reportSequentialSegmentFinalized(index: currentMuxerSegmentIndex, - nextIndex: newIdx) + reportSequentialSegmentFinalized(index: currentMuxerSegmentIndex, isFinal: false) } // Cut succeeded but muxer failed to open the next staging fd: silently discards every subsequent byte. if muxer.isWedged { @@ -1524,22 +1550,40 @@ final class HLSSegmentProducer: @unchecked Sendable { return muxer } - /// Sequential-VOD finalize report: real duration = next segment's item-axis start minus this - /// one's, both recorded at the #65 ledger site. Falls back to the cut target when a start is - /// missing (NOPTS dts at the boundary) - one estimated EXTINF beats a stalled playlist. - private func reportSequentialSegmentFinalized(index: Int, nextIndex: Int?) { - let duration: Double - if let start = vodSegmentStartByIndex[index], - let nextIndex, let nextStart = vodSegmentStartByIndex[nextIndex], - nextStart > start { - duration = nextStart - start + /// Sequential-VOD duration became known (next segment's start recorded at the ledger). + /// Reports immediately when the segment is already captured; otherwise parks until the + /// capture side arrives. + private func noteSequentialDurationKnown(index: Int, duration: Double) { + if seqCapturedAwaitingDuration.remove(index) != nil { + emitSequentialReport(index: index, duration: duration) } else { - duration = targetSegmentDurationSeconds + seqDurationAwaitingCapture[index] = duration + } + } + + /// Sequential-VOD capture side of the pairing. `isFinal` (EOF finalize) reports with the cut + /// target when no next ledger entry will ever supply the real duration - one estimated tail + /// EXTINF beats a playlist that never completes. + private func reportSequentialSegmentFinalized(index: Int, isFinal: Bool) { + if let dur = seqDurationAwaitingCapture.removeValue(forKey: index) { + emitSequentialReport(index: index, duration: dur) + } else if isFinal { + let dur: Double + if let start = vodSegmentStartByIndex[index], lastMuxedItemAxisSeconds > start { + dur = lastMuxedItemAxisSeconds - start + } else { + dur = targetSegmentDurationSeconds + } + emitSequentialReport(index: index, duration: dur) + } else { + seqCapturedAwaitingDuration.insert(index) } - vodSegmentStartByIndex.removeValue(forKey: index) - onSequentialSegmentFinalized?(index, duration) } + /// Newest item-axis video segment start muxed (seconds); floors the final segment's EXTINF + /// estimate at EOF. Pump-thread only. + private var lastMuxedItemAxisSeconds: Double = 0 + private func reportLiveSegmentFinalized(index: Int, nextIndex: Int?) { guard let startSeconds = liveSegmentStartByIndex[index] else { EngineLog.emit( @@ -1581,7 +1625,7 @@ final class HLSSegmentProducer: @unchecked Sendable { if isLive { reportLiveSegmentFinalized(index: idx, nextIndex: nil) } else if onSequentialSegmentFinalized != nil { - reportSequentialSegmentFinalized(index: idx, nextIndex: nil) + reportSequentialSegmentFinalized(index: idx, isFinal: true) } } else { EngineLog.emit( @@ -2792,7 +2836,23 @@ final class HLSSegmentProducer: @unchecked Sendable { let shiftTicks = videoShiftPts == Int64.min ? 0 : videoShiftPts let outDts = prev.pointee.dts if onSequentialSegmentFinalized != nil { - vodSegmentStartByIndex[prevSeg] = Double(outDts) * sourceVideoTbSeconds + let startSec = Double(outDts) * sourceVideoTbSeconds + vodSegmentStartByIndex[prevSeg] = startSec + lastMuxedItemAxisSeconds = startSec + // The previous segment's REAL duration is final the moment this + // one's start is known; the report pairs with its capture. Plan + // indices a long GOP skipped report as zero-duration holes. + if lastSeqLedgerSeg != Int.min, lastSeqLedgerSeg < prevSeg, + let priorStart = vodSegmentStartByIndex[lastSeqLedgerSeg], + startSec > priorStart { + vodSegmentStartByIndex.removeValue(forKey: lastSeqLedgerSeg) + noteSequentialDurationKnown(index: lastSeqLedgerSeg, + duration: startSec - priorStart) + for hole in (lastSeqLedgerSeg + 1)..