diff --git a/Sources/AetherEngine/AetherEngine+Loading.swift b/Sources/AetherEngine/AetherEngine+Loading.swift index 1cd56b35..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 @@ -502,6 +510,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 @@ -1123,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 @@ -1292,18 +1310,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 +1379,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..8a27ad7c 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 @@ -2996,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 { 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/Network/HLSLocalServer.swift b/Sources/AetherEngine/Network/HLSLocalServer.swift index ff514ebf..1b764889 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() @@ -1343,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/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/HLSSegmentProducer.swift b/Sources/AetherEngine/Video/HLSSegmentProducer.swift index 0e66b83b..70f1bb22 100644 --- a/Sources/AetherEngine/Video/HLSSegmentProducer.swift +++ b/Sources/AetherEngine/Video/HLSSegmentProducer.swift @@ -174,6 +174,44 @@ 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. 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 @@ -1460,6 +1498,8 @@ final class HLSSegmentProducer: @unchecked Sendable { if isLive { reportLiveSegmentFinalized(index: currentMuxerSegmentIndex, nextIndex: newIdx) + } else if onSequentialSegmentFinalized != nil { + reportSequentialSegmentFinalized(index: currentMuxerSegmentIndex, isFinal: false) } // Cut succeeded but muxer failed to open the next staging fd: silently discards every subsequent byte. if muxer.isWedged { @@ -1510,6 +1550,40 @@ final class HLSSegmentProducer: @unchecked Sendable { return muxer } + /// 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 { + 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) + } + } + + /// 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( @@ -1550,6 +1624,8 @@ final class HLSSegmentProducer: @unchecked Sendable { byteCount: result.bytesWritten) if isLive { reportLiveSegmentFinalized(index: idx, nextIndex: nil) + } else if onSequentialSegmentFinalized != nil { + reportSequentialSegmentFinalized(index: idx, isFinal: true) } } else { EngineLog.emit( @@ -2759,6 +2835,25 @@ final class HLSSegmentProducer: @unchecked Sendable { vodLedgerLastRoutedSeg = prevSeg let shiftTicks = videoShiftPts == Int64.min ? 0 : videoShiftPts let outDts = prev.pointee.dts + if onSequentialSegmentFinalized != nil { + 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)..= 0 && localI < segmentBoundaries.count) diff --git a/Sources/AetherEngine/Video/HLSVideoEngine+LiveReopen.swift b/Sources/AetherEngine/Video/HLSVideoEngine+LiveReopen.swift index ffe8898d..cc6509dd 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 @@ -159,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 4b021cad..180b38d1 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 @@ -1399,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 @@ -1439,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( @@ -2199,7 +2223,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/AetherEngine/Video/VideoSegmentProvider.swift b/Sources/AetherEngine/Video/VideoSegmentProvider.swift index 18e0ddd7..a4c43dd1 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,58 @@ 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 + } + // Zero marks a plan index a long GOP skipped entirely (no media file exists); the + // playlist renderer omits those entries. Negative values are producer bugs, clamp them. + _seqDurations.append(max(0, 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 +473,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 +891,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 +939,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 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) }