From 3bf8acc850411b9960db9fd48ff26df701658ca9 Mon Sep 17 00:00:00 2001 From: orut34iop Date: Tue, 13 Oct 2026 15:29:16 +0800 Subject: [PATCH 1/4] fix(cache): publish retained software VOD resident ranges --- CHANGELOG.md | 6 + .../AetherEngine/AetherEngine+Loading.swift | 7 + Sources/AetherEngine/AetherEngine.swift | 3 +- .../Native/SoftwarePacketCoverage.swift | 26 +++ .../Native/SoftwarePacketDiskFIFO.swift | 1 + .../Native/SoftwarePacketReadAhead.swift | 89 ++++++++ .../Native/SoftwarePlaybackHost.swift | 62 ++++++ .../Native/SoftwareVideoPacketCoverage.swift | 6 + .../SoftwareResidentRangesTests.swift | 204 ++++++++++++++++++ docs/api.md | 19 +- 10 files changed, 420 insertions(+), 3 deletions(-) create mode 100644 Tests/AetherEngineTests/SoftwareResidentRangesTests.swift diff --git a/CHANGELOG.md b/CHANGELOG.md index 73d956cdd..d02c73bff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,12 @@ the public-API contract. ### Fixed +- **Software VOD exposes retained cache ranges to host timelines.** `residentRanges` now covers + compressed video packets together with the selected audio stream, including retained history, + gaps and disk eviction. Cached seeks keep the same spans; cold seeks and teardown clear them. + Sampling continues while paused and runs off the main actor, with stale seek/stop results + discarded. Hosts no longer need to infer a cache band's start from the moving playhead. + - **A host with its own Objective-C and C++ links again (#752).** Since 7.34.0 LibTrueHD and LibDovi each brought a `rust_eh_personality`, and with the host's `__objc_personality_v0` and `__gxx_personality_v0` that made four personality routines, one more than compact unwind encodes diff --git a/Sources/AetherEngine/AetherEngine+Loading.swift b/Sources/AetherEngine/AetherEngine+Loading.swift index c6a71dc6b..f88276bbf 100644 --- a/Sources/AetherEngine/AetherEngine+Loading.swift +++ b/Sources/AetherEngine/AetherEngine+Loading.swift @@ -1941,6 +1941,13 @@ extension AetherEngine { // #353: the settled picture size, wired next to the frame times because a host laying out an // overlay needs the rectangle as well as the clock, and both come off this renderer. mirrorSoftwareDisplaySize(from: host.$videoDisplaySize, storeIn: &softwareCancellables) + host.$residentRanges + .sink { [weak self, weak host] ranges in + guard let self, let host, self.softwareHost === host, + self.loadGeneration == generation, !self.isLive else { return } + self.residentRanges = ranges + } + .store(in: &softwareCancellables) // SW-PiP: publish the bridge once the session owns its layer (the layer object is stable for // the session; the host attaches it to the view and, on PiP start, to the system window). softwarePiPSource = SoftwarePiPSource(layer: host.displayLayer, isLive: isLive, engine: self) diff --git a/Sources/AetherEngine/AetherEngine.swift b/Sources/AetherEngine/AetherEngine.swift index 295e5c0e7..8c22f2a17 100644 --- a/Sources/AetherEngine/AetherEngine.swift +++ b/Sources/AetherEngine/AetherEngine.swift @@ -83,7 +83,8 @@ public final class AetherEngine: ObservableObject { didSet { recomputePlaybackPhase() } } - /// Where the loopback segment cache holds picture right now, on the same 0-based display axis as + /// Where the loopback segment cache or software VOD packet cache holds picture right now, + /// with selected audio coverage required for software A/V, on the same 0-based display axis as /// `currentTime`, `duration` and `seek(to:)`. An empty array is the nil-equivalent, and live /// sessions publish one always: their rewind depth is `clock.seekableLiveRange`, a different /// contract. Residency is not a promise that a seek inside a range will be instant; the player may diff --git a/Sources/AetherEngine/Native/SoftwarePacketCoverage.swift b/Sources/AetherEngine/Native/SoftwarePacketCoverage.swift index bd1e2998c..20408a1e1 100644 --- a/Sources/AetherEngine/Native/SoftwarePacketCoverage.swift +++ b/Sources/AetherEngine/Native/SoftwarePacketCoverage.swift @@ -22,6 +22,32 @@ struct SoftwarePacketCoverage: Sendable { var rangeCount: Int { ranges.count } var isFull: Bool { ranges.count >= maximumRangeCount } + /// Immutable presentation spans for a background residency snapshot. Never round over a gap. + func rangesSeconds(timeBaseNumerator: Int32, timeBaseDenominator: Int32) -> [ClosedRange] { + guard timeBaseNumerator > 0, timeBaseDenominator > 0 else { return [] } + return ranges.compactMap { range in + guard Int64(exactly: Double(range.lowerBound)) == range.lowerBound, + Int64(exactly: Double(range.upperBound)) == range.upperBound else { return nil } + let start = Double(range.lowerBound) * Double(timeBaseNumerator) / Double(timeBaseDenominator) + let end = Double(range.upperBound) * Double(timeBaseNumerator) / Double(timeBaseDenominator) + guard start.isFinite, end.isFinite, start < end else { return nil } + return start...end + } + } + + /// Linear intersection of sorted, disjoint spans; selected audio is required even when muted. + static func intersection(_ video: [ClosedRange], _ audio: [ClosedRange]) -> [ClosedRange] { + var result: [ClosedRange] = [] + var v = 0, a = 0 + while v < video.count, a < audio.count { + let start = max(video[v].lowerBound, audio[a].lowerBound) + let end = min(video[v].upperBound, audio[a].upperBound) + if start < end { result.append(start...end) } + if video[v].upperBound < audio[a].upperBound { v += 1 } else { a += 1 } + } + return result + } + /// Adds [pts, pts + duration). Invalid or capacity-exceeding input leaves existing coverage /// unchanged. Rejecting a sparse extension is conservative: it never claims a missing packet. @discardableResult diff --git a/Sources/AetherEngine/Native/SoftwarePacketDiskFIFO.swift b/Sources/AetherEngine/Native/SoftwarePacketDiskFIFO.swift index 2126cbf78..c665727fe 100644 --- a/Sources/AetherEngine/Native/SoftwarePacketDiskFIFO.swift +++ b/Sources/AetherEngine/Native/SoftwarePacketDiskFIFO.swift @@ -57,6 +57,7 @@ final class SoftwarePacketDiskFIFO: @unchecked Sendable { let staleSweepResult: StaleSweepResult private let chunkTargetBytes: Int private let retainConsumed: Bool + var retainsConsumedPackets: Bool { retainConsumed } private let lock = NSLock() private var writer: FileHandle? private var reader: FileHandle? diff --git a/Sources/AetherEngine/Native/SoftwarePacketReadAhead.swift b/Sources/AetherEngine/Native/SoftwarePacketReadAhead.swift index 3a7ee53d2..5d4e533cf 100644 --- a/Sources/AetherEngine/Native/SoftwarePacketReadAhead.swift +++ b/Sources/AetherEngine/Native/SoftwarePacketReadAhead.swift @@ -82,6 +82,7 @@ final class SoftwarePacketReadAhead: @unchecked Sendable { } private var keyframes: [Keyframe] = [] private let maximumKeyframes = 65_536 + private var residency = SoftwarePacketResidency() /// Construct only off-main: creating the FIFO touches the temporary volume. init(video: Stream, audio: Stream?, byteBudget: Int, forwardSeconds: Double, @@ -134,6 +135,35 @@ final class SoftwarePacketReadAhead: @unchecked Sendable { cacheSeekMisses: cacheSeekMisses) } + /// Called only by a background sampler. Copy bounded metadata under the short state lock; + /// timestamp conversion/intersection and keyframe selection happen after releasing it. No I/O. + var residentRanges: [ClosedRange] { + condition.lock() + guard !closed, !sourceRepositioning, !resetPending, failure == nil, + fifo.retainsConsumedPackets else { condition.unlock(); return [] } + let videoCoverage = self.videoCoverage + let presentationCoverage = self.presentationCoverage + let audioCoverage = self.audioCoverage + let keyframes = self.keyframes + let videoFloor = residency.videoFloor + let audioFloor = residency.audioFloor + condition.unlock() + let videoRanges = presentationCoverage?.rangesSeconds ?? videoCoverage.rangesSeconds( + timeBaseNumerator: video.numerator, timeBaseDenominator: video.denominator) + let ranges: [ClosedRange] + if let audio { + ranges = SoftwarePacketCoverage.intersection(videoRanges, audioCoverage.rangesSeconds( + timeBaseNumerator: audio.numerator, timeBaseDenominator: audio.denominator)) + } else { ranges = videoRanges } + guard let keyframe = keyframes.lazy.map(\.seconds) + .filter({ $0 >= videoFloor }).min() else { return [] } + let floor = max(keyframe, audio == nil ? -.infinity : audioFloor) + return ranges.compactMap { range in + let start = max(range.lowerBound, floor) + return start < range.upperBound ? start...range.upperBound : nil + } + } + /// Main-thread safe: metadata only. Decode/render backpressure still belongs to the old loop. func updatePlayhead(_ seconds: Double) { guard seconds.isFinite else { return } @@ -405,6 +435,7 @@ final class SoftwarePacketReadAhead: @unchecked Sendable { videoCoverage.reset(); audioCoverage.reset(); presentationCoverage?.reset() storedVideoSeconds = nil; consumedVideoSeconds = nil keyframes.removeAll(keepingCapacity: true) + residency = SoftwarePacketResidency() } /// Presentation seconds of a stored packet, or nil when this stream cannot express them. @@ -420,6 +451,7 @@ final class SoftwarePacketReadAhead: @unchecked Sendable { residentBytes = state.residentByteCount if let floor = state.oldestRetainedChunkID { keyframes.removeAll { $0.cursor.chunkID < floor } + residency.evict(before: floor) } else { keyframes.removeAll(keepingCapacity: true) } } @@ -497,6 +529,12 @@ final class SoftwarePacketReadAhead: @unchecked Sendable { condition.lock() if token == sourceEpoch, !closed, !sourceRepositioning { copyDiskStateLocked(state) + if packet.streamIndex == video.index || packet.streamIndex == audio?.index { + let stream = packet.streamIndex == video.index ? video : audio! + residency.insert(chunkID: cursor.chunkID, pts: packet.pts, + duration: packet.duration, numerator: stream.numerator, + denominator: stream.denominator, isVideo: packet.streamIndex == video.index) + } if packet.streamIndex == video.index { if presentationCoverage != nil { if presentationCoverage?.isFull == true, let tick = playheadTick(video) { @@ -640,3 +678,54 @@ final class SoftwarePacketReadAhead: @unchecked Sendable { + "qos=\(QoSClass.name(reportedProducerQoS))" } } + +/// Constant-size summaries per retained disk chunk, never per packet. The cumulative discarded +/// high-water marks also cover reordered PTS: clipping at the first packet in the new chunk alone +/// could otherwise claim an evicted B picture or audio packet still exists. Metadata pressure may +/// conservatively shorten a displayed range; it never changes disk retention or seek policy. +struct SoftwarePacketResidency: Sendable { + private struct Chunk: Sendable { + let id: UInt64 + var videoEnd = -Double.infinity + var audioEnd = -Double.infinity + } + private var chunks: [Chunk] = [] + private(set) var videoFloor = -Double.infinity + private(set) var audioFloor = -Double.infinity + private let maximumChunks: Int + + init(maximumChunks: Int = 4096) { self.maximumChunks = max(1, maximumChunks) } + + mutating func insert(chunkID: UInt64, pts: Int64, duration: Int64, + numerator: Int32, denominator: Int32, isVideo: Bool) { + if chunks.last?.id != chunkID { + if chunks.count == maximumChunks { discardFirst(1) } + chunks.append(Chunk(id: chunkID)) + } + // A successor-duration video with no packet duration still excludes the evicted PTS. + let (end, overflow) = pts.addingReportingOverflow(max(1, duration)) + let seconds: Double + if pts == Int64.min || overflow || numerator <= 0 || denominator <= 0 + || Int64(exactly: Double(end)) != end { + seconds = .infinity // unknown discarded timestamps cannot prove later residency + } else { + seconds = Double(end) * Double(numerator) / Double(denominator) + } + let index = chunks.count - 1 + if isVideo { chunks[index].videoEnd = max(chunks[index].videoEnd, seconds) } + else { chunks[index].audioEnd = max(chunks[index].audioEnd, seconds) } + } + + mutating func evict(before chunkID: UInt64) { + discardFirst(chunks.prefix { $0.id < chunkID }.count) + } + + private mutating func discardFirst(_ count: Int) { + guard count > 0 else { return } + for chunk in chunks.prefix(count) { + videoFloor = max(videoFloor, chunk.videoEnd) + audioFloor = max(audioFloor, chunk.audioEnd) + } + chunks.removeFirst(count) + } +} diff --git a/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift b/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift index a383b4d29..58199214d 100644 --- a/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift +++ b/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift @@ -138,6 +138,8 @@ final class SoftwarePlaybackHost { private var vodPacketReadAhead: SoftwarePacketReadAhead? /// #687: this host's entries in the process-wide retention ledger, released on `stop()`. private var retentionClaims: [RetentionClaims.Claim] = [] + @Published private(set) var residentRanges: [ClosedRange] = [] + private let residentRangePublication = SoftwareResidentRangePublication() /// The same public buffered-position axis as the live and native hosts, but backed by /// actual compressed A/V packet coverage. nil when no continuous cache span contains the clock. @@ -1473,6 +1475,7 @@ final class SoftwarePlaybackHost { // reposition can tell on `seekQueue` whether a newer seek has already taken over. bumpSeekGeneration() let generation = seekGeneration + residentRangePublication.invalidate() didReachEnd = false didParkClockAtEnd = false didEmitParkedDiag = false @@ -1538,6 +1541,7 @@ final class SoftwarePlaybackHost { switch preparation { case .success(let hit): cacheHit = hit + if !hit { residentRanges = [] } EngineLog.emit( "[SWHost] packet cache seek generation=\(generation) " + "result=\(hit ? "hit" : "miss") " @@ -1553,6 +1557,7 @@ final class SoftwarePlaybackHost { // a normal cache miss. Leave playback parked and publish an explicit failure. seekInFlight = false packetSource.close() + residentRanges = [] noteSeekSettled(generation) EngineLog.emit("[SWHost] packet cache seek generation=\(generation) result=error", category: .swPlayback) @@ -1842,6 +1847,8 @@ final class SoftwarePlaybackHost { noteSeekSettled(seekGeneration) timeTimer?.cancel() timeTimer = nil + residentRangePublication.invalidate() + residentRanges = [] vodPacketReadAhead?.close() vodPacketReadAhead = nil renderer.subtitleCompositor.reset() @@ -3285,6 +3292,32 @@ final class SoftwarePlaybackHost { // MARK: - Time updates + nonisolated static func displayResidentRanges(_ ranges: [ClosedRange], + sessionZeroSeconds: Double) -> [ClosedRange] { + guard sessionZeroSeconds.isFinite else { return [] } + let zero = max(0, sessionZeroSeconds) + return ranges.compactMap { range in + let start = max(0, range.lowerBound - zero), end = range.upperBound - zero + return start.isFinite && end.isFinite && start < end ? start...end : nil + } + } + + /// Also runs while paused, independently of rendered-time events. At most one bounded + /// metadata calculation is pending. Cached seeks keep the last valid spans; a miss clears + /// them at the actual discard decision. Seek/stop invalidate any pre-operation result. + private func sampleResidentRanges() { + guard !isLive, !stopRequested, !seekInFlight, + let source = vodPacketReadAhead else { return } + let zero = clockSessionZero + residentRangePublication.sample { + Self.displayResidentRanges(source.residentRanges, sessionZeroSeconds: zero) + } publish: { [weak self] ranges in + guard let self, self.vodPacketReadAhead === source, !self.stopRequested, + self.clockSessionZero == zero else { return } + if self.residentRanges != ranges { self.residentRanges = ranges } + } + } + private func startTimeUpdates() { timeTimer = Timer.publish(every: 0.25, on: .main, in: .common) .autoconnect() @@ -3292,6 +3325,7 @@ final class SoftwarePlaybackHost { guard let self else { return } // Independent of the clock below: a seek in flight must not defer the one check. self.checkSurfaceVisibilityIfDue() + self.sampleResidentRanges() guard let aOut = self.audioOutput else { return } // #254: a reposition in flight holds `currentTime` at its target; the synchronizer is // still on the pre-seek anchor and would drag the published position backwards. @@ -3576,3 +3610,31 @@ extension SoftwarePlaybackHost: LiveRecordingHost { ) } } + +/// A single background snapshot at a time. Invalidating at seek/stop prevents a delayed snapshot +/// from restoring a discarded range or from clearing a newer request. Main never waits for work. +@MainActor +final class SoftwareResidentRangePublication { + private var task: Task? + private var revision: UInt64 = 0 + + func invalidate() { + revision &+= 1 + task?.cancel() + task = nil + } + + @discardableResult + func sample(_ read: @escaping @Sendable () async -> [ClosedRange], + publish: @escaping @MainActor ([ClosedRange]) -> Void) -> Task? { + guard task == nil else { return nil } + let token = revision + task = Task { [weak self] in + let ranges = await Task.detached(priority: .utility, operation: read).value + guard let self, !Task.isCancelled, self.revision == token else { return } + self.task = nil + publish(ranges) + } + return task + } +} diff --git a/Sources/AetherEngine/Native/SoftwareVideoPacketCoverage.swift b/Sources/AetherEngine/Native/SoftwareVideoPacketCoverage.swift index 136b77b1a..ae44a14e9 100644 --- a/Sources/AetherEngine/Native/SoftwareVideoPacketCoverage.swift +++ b/Sources/AetherEngine/Native/SoftwareVideoPacketCoverage.swift @@ -48,6 +48,12 @@ struct SoftwareVideoPacketCoverage: Sendable { var rangeCount: Int { coverage.rangeCount } var isFull: Bool { coverage.isFull } + var rangesSeconds: [ClosedRange] { + guard !isInvalidated else { return [] } + return coverage.rangesSeconds(timeBaseNumerator: timeBaseNumerator, + timeBaseDenominator: timeBaseDenominator) + } + /// Duration/DTS are deliberately not inputs. Duplicate pending PTS do not advance the reorder /// watermark. At most depth + 1 timestamps are held transiently, then the minimum is emitted. @discardableResult diff --git a/Tests/AetherEngineTests/SoftwareResidentRangesTests.swift b/Tests/AetherEngineTests/SoftwareResidentRangesTests.swift new file mode 100644 index 000000000..62e0573aa --- /dev/null +++ b/Tests/AetherEngineTests/SoftwareResidentRangesTests.swift @@ -0,0 +1,204 @@ +import Foundation +import Testing +@testable import AetherEngine + +@Suite("Software VOD resident ranges", .timeLimit(.minutes(2))) +struct SoftwareResidentRangesTests { + private final class Input: @unchecked Sendable { + private let lock = NSLock() + private var packets: [SoftwareStoredPacket] + private var index = 0 + init(_ packets: [SoftwareStoredPacket]) { self.packets = packets } + func next() -> SoftwareStoredPacket? { + lock.lock(); defer { lock.unlock() } + guard index < packets.count else { return nil } + defer { index += 1 }; return packets[index] + } + } + private func packet(_ pts: Int64, stream: Int32 = 0, duration: Int64 = 40, + key: Bool = true) -> SoftwareStoredPacket { + .init(pts: pts, dts: pts, duration: duration, position: 0, streamIndex: stream, + flags: key ? 1 : 0, timeBaseNumerator: 1, timeBaseDenominator: 1000, + bytes: Data(repeating: 7, count: 100), sideData: []) + } + private func source(_ packets: [SoftwareStoredPacket], audio: Bool = false, + reorder: Int? = nil, budget: Int = 1_000_000, + chunkBytes: Int = 1024) throws -> SoftwarePacketReadAhead { + let input = Input(packets) + let fifo = try SoftwarePacketDiskFIFO(chunkTargetBytes: chunkBytes, retainConsumed: true) + let source = SoftwarePacketReadAhead(video: .init(index: 0, numerator: 1, denominator: 1000), + audio: audio ? .init(index: 1, numerator: 1, denominator: 1000) : nil, + byteBudget: budget, forwardSeconds: 3600, initialSourceClock: 0, + fifo: fifo, videoReorderDepth: reorder) { _ in input.next() } + source.start() + return source + } + + @Test("Backward and forward cache hits preserve the same spans through begin, prepare and settle") + func cachedSeeks() async throws { + let source = try source((0..<100).map { packet(Int64($0) * 40) }) + defer { source.close() } + try await waitFor { source.snapshot.sourceEnded } + #expect(source.residentRanges == [0...4]) + for target in [3.0, 1, 2.5, 0.5] { + source.updatePlayhead(target) + let token = source.beginSeek(to: target) + #expect(source.residentRanges == [0...4]) + #expect(try source.prepareSeek(token, to: target)) + #expect(source.residentRanges == [0...4]) + source.endSeek(token, sourceClock: target) + #expect(source.residentRanges == [0...4]) + } + let miss = source.beginSeek(to: 8) + #expect(try !source.prepareSeek(miss, to: 8)) + #expect(source.residentRanges.isEmpty) + source.close() + #expect(source.residentRanges.isEmpty) + } + + @Test("Required audio intersects video, keeps holes, and missing selected audio proves nothing") + func audioIntersection() async throws { + let video = (0..<100).map { packet(Int64($0) * 40) } + let audio = [packet(500, stream: 1, duration: 500), packet(2000, stream: 1, duration: 1000)] + let source = try source(video + audio, audio: true) + defer { source.close() } + try await waitFor { source.snapshot.sourceEnded } + #expect(source.residentRanges == [0.5...1, 2...3]) + let missing = try self.source(video, audio: true) + defer { missing.close() } + try await waitFor { missing.snapshot.sourceEnded } + #expect(missing.residentRanges.isEmpty) + } + + @Test("Successor PTS retains reorder semantics, a discontinuity, and no invented final frame") + func reorderedVideo() async throws { + let pts: [Int64] = [0, 120, 40, 80, 160, 2000, 2120, 2040, 2080] + let source = try source(pts.map { packet($0, duration: 0) }, reorder: 4) + defer { source.close() } + try await waitFor { source.snapshot.sourceEnded } + #expect(source.residentRanges == [0...0.16, 2...2.12]) + let invalid = try self.source(([Int64](stride(from: 0, to: 2000, by: 40)) + [0]) + .map { packet($0) }, reorder: 4) + defer { invalid.close() } + try await waitFor { invalid.snapshot.sourceEnded } + #expect(invalid.residentRanges.isEmpty) + } + + @Test("Real FIFO eviction shortens backward coverage without changing the read bytes") + func diskEviction() async throws { + let packets = (0..<100).map { packet(Int64($0) * 40) } + let source = try source(packets, budget: 1400, chunkBytes: 512) + defer { source.close() } + let retained = try await withTaskCancellationHandler { + try await withCheckedThrowingContinuation { (continuation: CheckedContinuation<[SoftwareStoredPacket], Error>) in + Thread.detachNewThread { + do { + var values: [SoftwareStoredPacket] = [] + while let packet = try source.read() { + values.append(packet) + source.updatePlayhead(Double(packet.pts) / 1000) + } + continuation.resume(returning: values) + } catch { continuation.resume(throwing: error) } + } + } + } onCancel: { source.close() } + #expect(retained == packets) + try await waitFor { source.snapshot.sourceEnded } + let spans = source.residentRanges + #expect(!spans.isEmpty) + #expect((spans.first?.lowerBound ?? 0) > 3) + #expect(spans.last?.upperBound == 4) + let miss = source.beginSeek(to: 0.1) + #expect(try !source.prepareSeek(miss, to: 0.1)) + #expect(source.residentRanges.isEmpty) + } + + @Test("Evicted reordered timestamps and metadata pressure only shorten coverage") + func residencyWatermarks() { + var value = SoftwarePacketResidency(maximumChunks: 2) + value.insert(chunkID: 0, pts: 900, duration: 40, numerator: 1, denominator: 1000, isVideo: true) + value.insert(chunkID: 1, pts: 100, duration: 40, numerator: 1, denominator: 1000, isVideo: true) + value.insert(chunkID: 1, pts: 1100, duration: 40, numerator: 1, denominator: 1000, isVideo: false) + value.evict(before: 1) + #expect(value.videoFloor == 0.94) + #expect(value.audioFloor == -.infinity) + value.insert(chunkID: 2, pts: 1200, duration: 0, numerator: 1, denominator: 1000, isVideo: true) + value.insert(chunkID: 3, pts: 1300, duration: 0, numerator: 1, denominator: 1000, isVideo: true) + #expect(value.videoFloor == 0.94) + #expect(value.audioFloor == 1.14) + value.evict(before: 4) + #expect(value.videoFloor == 1.301) + } + + @Test("Invalid discarded timestamps fail closed and reset begins a separate residency lifetime") + func invalidWatermark() { + var value = SoftwarePacketResidency() + value.insert(chunkID: 0, pts: .max, duration: 40, numerator: 1, denominator: 1000, isVideo: true) + value.evict(before: 1) + #expect(value.videoFloor == .infinity) + value = SoftwarePacketResidency() + #expect(value.videoFloor == -.infinity) + } + + @Test("Source origin mapping preserves gaps, clips negatives and rejects unknown origins") + func displayMapping() { + let spans = [90.0...105, 110...120] + #expect(SoftwarePlaybackHost.displayResidentRanges(spans, sessionZeroSeconds: 100) == [0...5, 10...20]) + #expect(SoftwarePlaybackHost.displayResidentRanges(spans, sessionZeroSeconds: .nan).isEmpty) + #expect(SoftwarePlaybackHost.displayResidentRanges(spans, sessionZeroSeconds: 200).isEmpty) + } + + @Test("Large inexact timestamps do not manufacture display ranges") + func exactness() { + var coverage = SoftwarePacketCoverage() + coverage.insert(pts: 9_007_199_254_740_993, duration: 4) + #expect(coverage.rangesSeconds(timeBaseNumerator: 1, timeBaseDenominator: 1).isEmpty) + } +} + +@MainActor +@Suite("Software resident range publication", .timeLimit(.minutes(2))) +struct SoftwareResidentRangePublicationTests { + private actor Gate { + private(set) var entered = false + private var waiter: CheckedContinuation? + func wait() async { + await withCheckedContinuation { waiter = $0; entered = true } + } + func release() { waiter?.resume(); waiter = nil } + } + + @Test("A delayed sampler leaves main responsive, coalesces ticks, and cannot publish after seek/stop") + func delayedSnapshots() async throws { + let publisher = SoftwareResidentRangePublication() + let gate = Gate() + var received: [[ClosedRange]] = [] + let old = publisher.sample { + await gate.wait() + return [0...100] + } publish: { received.append($0) } + #expect(publisher.sample({ [0...999] }, publish: { received.append($0) }) == nil) + try await waitFor { await gate.entered } + // This main-actor turn can run while the background read is suspended. A new seek or stop + // invalidates it; a new session/seek can immediately publish without waiting for that read. + publisher.invalidate() + let replacement = publisher.sample({ [200...220] }, publish: { received.append($0) }) + await replacement?.value + #expect(received == [[200...220]]) + await gate.release() + await old?.value + #expect(received == [[200...220]]) + } + + @Test("Paused growth, eviction and a reset publish without a clock tick") + func pausedUpdates() async { + let publisher = SoftwareResidentRangePublication() + var received: [[ClosedRange]] = [] + for spans: [ClosedRange] in [[10...30], [10...60], [20...60], [], [80...90]] { + let work = publisher.sample({ spans }, publish: { received.append($0) }) + await work?.value + } + #expect(received == [[10...30], [10...60], [20...60], [], [80...90]]) + } +} diff --git a/docs/api.md b/docs/api.md index ecf343240..7ecfb3ae4 100644 --- a/docs/api.md +++ b/docs/api.md @@ -585,10 +585,25 @@ Time lives on `player.clock`, a separate `ObservableObject`, so ~10 Hz ticks nev | `clock.$progress` | `currentTime / duration` | | `clock.$bufferedPosition` | source-axis position buffered ahead | | `clock.$liveEdgeTime`, `clock.$seekableLiveRange`, `clock.$behindLiveSeconds`, `clock.$isAtLiveEdge` | live-window surfaces. `seekableLiveRange` is the intersection of the DVR window (policy) and what the segment cache actually holds and can play forward from (fact), so it is honest to scale a rewind strip on and `seek(to:)` clamps to the same floor (AE#441). The two diverge for the whole first `dvrWindowSeconds` of a session and again whenever retention evicts faster than the window slides. Software live sessions hold a packet ring instead, which evicts by the same byte budget as the native cache (a quarter of the volume's free space, at most 2 GiB, at least 64 MiB), so the range's floor follows the ring once eviction has moved it past the window. | -| `$residentRanges` | where the loopback segment cache holds picture right now, as disjoint ascending spans on the `currentTime` axis. This is the cache's own truth, not AVPlayer's `loadedTimeRanges`: a measured session held 64 segments over four minutes across several islands while AVPlayer exposed roughly twelve seconds around the playhead and forgot a seeked-ahead island as soon as the playhead left it. Empty is the nil-equivalent, and a live session publishes empty always (its rewind depth is `seekableLiveRange`, which answers a different question). Residency is not a promise that a seek inside a span is instant: the player may still re-anchor and decode at the target, and a segment can start mid-GOP (AE#412). Coalesced to at most four updates a second, cleared on `load()` and teardown. | +| `$residentRanges` | where the loopback segment cache or software VOD packet cache holds picture right now, as disjoint ascending spans on the `currentTime` axis. This is the cache's own truth, not AVPlayer's `loadedTimeRanges`: a measured session held 64 segments over four minutes across several islands while AVPlayer exposed roughly twelve seconds around the playhead and forgot a seeked-ahead island as soon as the playhead left it. Empty is the nil-equivalent, and a live session publishes empty always (its rewind depth is `seekableLiveRange`, which answers a different question). Residency is not a promise that a seek inside a span is instant: the player may still re-anchor and decode at the target, and a segment can start mid-GOP (AE#412). Coalesced to at most four updates a second, cleared on `load()` and teardown. | | `player.currentTime`, `sourceTime`, `sourceTimeFollowsPicture`, `progress`, `bufferedPosition`, `liveEdgeTime`, `seekableLiveRange`, `behindLiveSeconds`, `isAtLiveEdge` | non-published mirrors of the same values for one-shot reads | | `$duration` | seconds; a `LoadOptions.declaredDurationSeconds` outranks the container's | +For software VOD, `residentRanges` intersects video presentation coverage with the selected audio +stream, even if muted; a video-only source does not require audio. H.264/HEVC use verified +successor-PTS intervals, so the final picture with no known successor does not acquire an invented +end. Ranges are clipped against retained keyframes and discarded packet timestamps; unknown or +invalid coverage stays empty. Metadata limits can conservatively omit old spans rather than claim +packets that no longer exist. + +The software sampler computes an immutable snapshot in the background and publishes on the main +actor, including while paused. Cached seeks preserve retained spans; a cache miss clears them when +the old store is discarded. Seek, replacement and teardown guards reject a late snapshot. Display +these spans directly on the `currentTime` axis: `bufferedPosition` alone is a forward frontier and +cannot establish a backward cache start. Live playback continues to expose its separate seekable +window rather than VOD resident ranges. + + **`sourceTime` on `nativeRemoteHLS` (AE#616).** The engine sees no segment on that route, so the clock is AVPlayer's item time. An origin whose playlist places a segment at its slot while the segment starts at the keyframe before it (a Jellyfin transcode restarted by `-ss -noaccurate_seek -copyts`) makes @@ -1128,7 +1143,7 @@ All flags default to safe values; the table is the full set. Depth for the media | `deinterlaceFieldRate` | `.field` | A `DeinterlaceFieldRate`: the hardware deinterlacer emits one frame per field (25i to 50p) or per frame. The software fallback is always frame rate, because doubling a CPU bwdif is the wrong trade and a fallback should not change cost class. | | `probesize`, `maxAnalyzeDuration` | nil | Caller-bounded open-time probe budget (defaults 50 MB / 60 s). They fail **open**: an over-tight budget loads with late-resolving tracks silently missing rather than throwing, so validate track presence if you tighten them. Do not pass `0` for `maxAnalyzeDuration`; FFmpeg maps it to a shorter heuristic. | | `progressiveSegmentDelivery` | false | Serve each VOD loopback segment while it is being written: the muxer flushes a fragment about every half second and the loopback server sends each one as it lands, so AVPlayer shows and starts on a segment's first fragments instead of waiting for its cut. Changes nothing on a source fast enough to cut segments before AVPlayer asks; on a slow link it removes a whole segment's download from startup and from a seek outside the buffer. A segment the producer abandons (restart, failed cut) ends its connection without the final chunk, so AVPlayer asks for it again. VOD only; ignored on live, `nativeRemoteHLS` and the software path. | -| `forwardBufferSegments` | nil (10, about 40 s) | How far the producer may race ahead and how much the cache keeps resident. Clamped to 4...2700; past the historical 150 the real bound is the session's disk budget, so a "buffer without limit" option can pass `Int.max`. Ignored on `nativeRemoteHLS`. Native HLS retention is readable as `$residentRanges` (see Time). Software VOD uses the same forward-window and volume-budget policy for compressed packet read-ahead; its continuous selected A/V frontier is `bufferedPosition`, not native `residentRanges`. | +| `forwardBufferSegments` | nil (10, about 40 s) | How far the producer may race ahead and how much the cache keeps resident. Clamped to 4...2700; past the historical 150 the real bound is the session's disk budget, so a "buffer without limit" option can pass `Int.max`. Ignored on `nativeRemoteHLS`. Native HLS retention is readable as `$residentRanges` (see Time). Software VOD uses the same forward-window and volume-budget policy for compressed packet read-ahead; its continuous selected A/V frontier is `bufferedPosition`, and its retained spans (including history and gaps) are `$residentRanges`. | | `sequentialOrigin` | false | Declare an origin that fabricates range answers: one long-lived unranged GET, no ranged probes, non-seekable pb. **Seeking is unavailable**; re-request the archive at a shifted start instead. A URL source the reader finds forward-only by itself (`Range` ignored, so it can only stream front to back; the open confirms a 200 answer with a one byte range, because some CDNs answer a cache miss that way) is served this way without the declaration when its container states a duration, which a stated `Content-Length` lets an MPEG-TS source estimate; `loadedOptions` then reads `true` with that duration declared, and without a duration it stays on the software path. An AirPlay hop on such a session swaps the item onto the LAN address instead of reopening the source, since a reopen could only start again from byte 0. | | `declaredDurationSeconds` | nil | Trusted duration, overriding the container's. Required alongside `sequentialOrigin` on VOD, where the tail read is gone. | | `maxConcurrentSourceRequests` | nil | Most requests the reader may have open against this origin at once, across every path it fetches on (pump ranges, detour blocks, size probes, tail prefetch, subtitle side reader, and the HLS live ingest's playlist, segment and key fetches, whose prefetch window narrows to it, AE#678). nil counts without capping and lowers the ceiling on its own after a 429/503/509. Set it when the provider states a limit; `1` also switches off the speculative parallel paths, which exist only to overlap with the pump. Counts **requests**, not TCP connections, because over HTTP/2 a session multiplexes every request onto one connection while the origin still counts each one (AE#377). It is also the only ceiling: several engines playing from one origin are bounded by this value and by what the origin refuses, not by a transport pool underneath it (AE#450). | From 761b031bcf406ada527d82fad251a25bb2415567 Mon Sep 17 00:00:00 2001 From: orut34iop Date: Tue, 13 Oct 2026 15:43:56 +0800 Subject: [PATCH 2/4] fix(cache): use upstream blocking executor for residency sampling --- Sources/AetherEngine/Native/SoftwarePlaybackHost.swift | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift b/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift index 58199214d..82f584abd 100644 --- a/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift +++ b/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift @@ -3630,7 +3630,7 @@ final class SoftwareResidentRangePublication { guard task == nil else { return nil } let token = revision task = Task { [weak self] in - let ranges = await Task.detached(priority: .utility, operation: read).value + let ranges = await BlockingWork.detached(priority: .utility, operation: read).value guard let self, !Task.isCancelled, self.revision == token else { return } self.task = nil publish(ranges) From 4a5eae0eefb750923355152f9326df33f5f94f08 Mon Sep 17 00:00:00 2001 From: orut34iop Date: Tue, 13 Oct 2026 17:08:55 +0800 Subject: [PATCH 3/4] test(cache): guard residency sampling executor --- .../SoftwareResidentRangesTests.swift | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/Tests/AetherEngineTests/SoftwareResidentRangesTests.swift b/Tests/AetherEngineTests/SoftwareResidentRangesTests.swift index 62e0573aa..8cd47c854 100644 --- a/Tests/AetherEngineTests/SoftwareResidentRangesTests.swift +++ b/Tests/AetherEngineTests/SoftwareResidentRangesTests.swift @@ -160,6 +160,25 @@ struct SoftwareResidentRangesTests { @MainActor @Suite("Software resident range publication", .timeLimit(.minutes(2))) struct SoftwareResidentRangePublicationTests { + private nonisolated static func isMainThread() -> Bool { + Thread.isMainThread + } + + @Test("Residency sampling leaves both main and the cooperative executor") + @available(visionOS 2, *) + func samplingExecutor() async { + let publisher = SoftwareResidentRangePublication() + var received: [ClosedRange] = [] + let work = publisher.sample { + let label = String(cString: __dispatch_queue_get_label(nil)) + #expect(!Self.isMainThread()) + #expect(!label.contains("cooperative"), "residency read ran on \(label)") + return [1...2] + } publish: { received = $0 } + await work?.value + #expect(received == [1...2]) + } + private actor Gate { private(set) var entered = false private var waiter: CheckedContinuation? From a905385a1ebc22130664c700239e63b68532bde6 Mon Sep 17 00:00:00 2001 From: orut34iop Date: Wed, 14 Oct 2026 16:06:06 +0800 Subject: [PATCH 4/4] fix(software): preserve valid residency and honor bounded positioning outcomes --- CHANGELOG.md | 5 + Sources/AetherEngine/AetherEngine.swift | 13 ++ Sources/AetherEngine/Demuxer/AVIOReader.swift | 7 +- Sources/AetherEngine/Demuxer/Demuxer.swift | 6 +- .../Native/SoftwarePacketReadAhead.swift | 51 +++++- .../Native/SoftwarePlaybackHost.swift | 71 +++++--- .../Native/SoftwareVideoPacketCoverage.swift | 9 +- Sources/AetherEngine/SeekEvent.swift | 5 +- .../DetourOriginTicketTests.swift | 50 ++++++ .../DocumentedConstantsTests.swift | 5 + .../Issue613PacketCoverageTests.swift | 108 ++++++++++++- .../SoftwareRepositionResultTests.swift | 152 ++++++++++++++++++ .../SoftwareResidentRangesTests.swift | 118 ++++++++++++++ docs/api.md | 21 ++- 14 files changed, 583 insertions(+), 38 deletions(-) create mode 100644 Tests/AetherEngineTests/SoftwareRepositionResultTests.swift diff --git a/CHANGELOG.md b/CHANGELOG.md index d02c73bff..06f1cb524 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,11 @@ the public-API contract. gaps and disk eviction. Cached seeks keep the same spans; cold seeks and teardown clear them. Sampling continues while paused and runs off the main actor, with stale seek/stop results discarded. Hosts no longer need to infer a cache band's start from the moving playhead. + VC-1 uses verified successor timestamps rather than rounded packet-duration islands. Valid + committed packets remain seekable after a source read error, which is still reported after drain; + storage failures remain unusable. Failed resume/seek positioning can no longer publish a false + successful landing. Bounded cold probes avoid speculative detour fetches, and foreground software + positioning has a finite 30 s budget for slow multi-probe remote keyframe searches. - **A host with its own Objective-C and C++ links again (#752).** Since 7.34.0 LibTrueHD and LibDovi each brought a `rust_eh_personality`, and with the host's `__objc_personality_v0` and diff --git a/Sources/AetherEngine/AetherEngine.swift b/Sources/AetherEngine/AetherEngine.swift index 8c22f2a17..acb4b31b4 100644 --- a/Sources/AetherEngine/AetherEngine.swift +++ b/Sources/AetherEngine/AetherEngine.swift @@ -6184,6 +6184,19 @@ public final class AetherEngine: ObservableObject { // A superseding seek owns the final state. guard loadGeneration == gen, seekGeneration == seekGen else { return } setPendingRecoverySeekTarget(nil) + if softwareHost != nil, hostReposition != .landed { + setProgrammaticSeek(inFlight: false, target: nil) + closeSeekTicket(&programmaticSeekTicket, + with: hostReposition == .superseded ? .superseded : .stalled) + // Host failure sinks may already have published .error. Never overwrite it with + // .playing/.paused or publish the requested position as though it was reached. + if hostReposition == .stalled { + if case .error = state {} else { + publishError(.softwarePipelineFailed, "Playback source could not reposition.") + } + } + return + } // Publish the measured native landing, not the optimistic request. The two can differ // slightly even after a successful callback; subtitle anchors and SeekEvent must agree. let landedClock = nativeOnly ? (nativeHost?.currentTime ?? clockTarget) : clockTarget diff --git a/Sources/AetherEngine/Demuxer/AVIOReader.swift b/Sources/AetherEngine/Demuxer/AVIOReader.swift index 126617cd6..6c8c3f2f4 100644 --- a/Sources/AetherEngine/Demuxer/AVIOReader.swift +++ b/Sources/AetherEngine/Demuxer/AVIOReader.swift @@ -2731,7 +2731,12 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { if let n = detourCache.serveCached(into: dst, maxLen: maxLen, at: offset) { return .served(n, fetched: false) } - guard allowFetch else { return .miss } + // A bounded position search may probe several distant offsets. Waiting for a whole + // speculative 4 MB block at each offset can spend the seek's entire deadline before + // the existing streaming reconnect gets a chance to deliver the requested bytes. + // Keep resident hits above, but let a bounded miss re-anchor immediately. Ordinary + // parse-time ping-pong still fills the detour cache and preserves its connection. + guard allowFetch, readDeadline == .distantFuture else { return .miss } let blockStart = (offset / Int64(Self.detourBlockSize)) * Int64(Self.detourBlockSize) let blockLen = Int(min(Int64(Self.detourBlockSize), fileSize - blockStart)) diff --git a/Sources/AetherEngine/Demuxer/Demuxer.swift b/Sources/AetherEngine/Demuxer/Demuxer.swift index 847ccc0fd..4d1d2dd99 100644 --- a/Sources/AetherEngine/Demuxer/Demuxer.swift +++ b/Sources/AetherEngine/Demuxer/Demuxer.swift @@ -2409,7 +2409,11 @@ public final class Demuxer: @unchecked Sendable { // matroska may return success with a partial index after abort; deadline flag // is authoritative, not ret. let capped = avioProvider?.readDeadlineFired ?? false - return ret >= 0 && !capped && probeControl?.isStopped != true + let stopped = probeControl?.isStopped == true + // Fixed result flags only: FFmpeg refusal, read deadline, cancellation. No source identity. + let flags = (ret < 0 ? 1 : 0) | (capped ? 2 : 0) | (stopped ? 4 : 0) + EngineLog.emit("[DemuxReposition] flags=\(flags)", category: .swPlayback) + return ret >= 0 && !capped && !stopped } /// How an off-actor reposition ended (#254). Named to mirror `SeekEvent.Outcome` so the engine's diff --git a/Sources/AetherEngine/Native/SoftwarePacketReadAhead.swift b/Sources/AetherEngine/Native/SoftwarePacketReadAhead.swift index 5d4e533cf..c0cce31b3 100644 --- a/Sources/AetherEngine/Native/SoftwarePacketReadAhead.swift +++ b/Sources/AetherEngine/Native/SoftwarePacketReadAhead.swift @@ -63,6 +63,17 @@ final class SoftwarePacketReadAhead: @unchecked Sendable { private var started = false private var ended = false private var failure: Error? + private enum FailureOrigin: Int { + case none = 0, sourceRead = 1024, storage = 2048, envelope = 4096 + } + private var failureOrigin: FailureOrigin = .none + // Called under condition. An upstream read failure says nothing about the integrity of + // envelopes already committed to the FIFO. Keep those usable; read() still throws the + // original failure after they drain, and the producer stays parked until a cold seek. + private var retainedPacketsUsableLocked: Bool { + failure == nil || failureOrigin == .sourceRead + } + private var lastCoverageDiagnosticFlags: Int? private var count = 0 private var bytes = 0 private var residentBytes = 0 @@ -139,7 +150,28 @@ final class SoftwarePacketReadAhead: @unchecked Sendable { /// timestamp conversion/intersection and keyframe selection happen after releasing it. No I/O. var residentRanges: [ClosedRange] { condition.lock() - guard !closed, !sourceRepositioning, !resetPending, failure == nil, + // Fixed numeric state only, emitted on edges by this background sampler. No URLs, + // error descriptions or packet content. Bits: closed/reposition/reset/failure, + // nonretaining/invalid-video/EOF/empty-audio/no-keyframe/unknown-floor, + // source-failure/storage-failure/envelope-failure/empty-video. + var flags = failureOrigin.rawValue + if closed { flags |= 1 } + if sourceRepositioning { flags |= 2 } + if resetPending { flags |= 4 } + if failure != nil { flags |= 8 } + if !fifo.retainsConsumedPackets { flags |= 16 } + if presentationCoverage?.isInvalidated == true { flags |= 32 } + if ended { flags |= 64 } + if audio != nil && audioCoverage.rangeCount == 0 { flags |= 128 } + if keyframes.isEmpty { flags |= 256 } + if residency.videoFloor == .infinity || residency.audioFloor == .infinity { flags |= 512 } + if (presentationCoverage?.rangeCount ?? videoCoverage.rangeCount) == 0 { flags |= 8192 } + let changed = lastCoverageDiagnosticFlags != flags + lastCoverageDiagnosticFlags = flags + defer { + if changed { EngineLog.emit("[SWCacheCoverage] flags=\(flags)", category: .swPlayback) } + } + guard !closed, !sourceRepositioning, !resetPending, retainedPacketsUsableLocked, fifo.retainsConsumedPackets else { condition.unlock(); return [] } let videoCoverage = self.videoCoverage let presentationCoverage = self.presentationCoverage @@ -208,7 +240,7 @@ final class SoftwarePacketReadAhead: @unchecked Sendable { guard token == generation, seeking, !closed else { condition.unlock(); throw ReadError.interrupted } - let hasCoverage = !sourceRepositioning && !resetPending && failure == nil + let hasCoverage = !sourceRepositioning && !resetPending && retainedPacketsUsableLocked && seconds.isFinite && (frontierLocked(at: seconds).map { $0 > seconds } ?? false) let candidates = hasCoverage ? keyframes.filter { $0.seconds <= seconds } .sorted { $0.seconds < $1.seconds } : [] @@ -272,7 +304,7 @@ final class SoftwarePacketReadAhead: @unchecked Sendable { guard seconds.isFinite, maxPackets > 0, isCancelled?() != true else { return nil } let deadline = ContinuousClock.now.advanced(by: .milliseconds(750)) condition.lock() - let eligible = !closed && !sourceRepositioning && !resetPending && failure == nil + let eligible = !closed && !sourceRepositioning && !resetPending && retainedPacketsUsableLocked && (frontierLocked(at: seconds).map { $0 > seconds } ?? false) let anchor = eligible ? keyframes.filter { $0.seconds <= seconds }.max { $0.seconds < $1.seconds } : nil @@ -431,7 +463,7 @@ final class SoftwarePacketReadAhead: @unchecked Sendable { private func clearSourceMetadataLocked() { count = 0; bytes = 0; residentBytes = 0 - ended = false; failure = nil + ended = false; failure = nil; failureOrigin = .none videoCoverage.reset(); audioCoverage.reset(); presentationCoverage?.reset() storedVideoSeconds = nil; consumedVideoSeconds = nil keyframes.removeAll(keepingCapacity: true) @@ -487,7 +519,7 @@ final class SoftwarePacketReadAhead: @unchecked Sendable { if token == sourceEpoch { resetPending = false } condition.broadcast() condition.unlock() - } catch { recordFailure(error, token: token) } + } catch { recordFailure(error, token: token, origin: .storage) } operations.unlock() continue } @@ -499,6 +531,7 @@ final class SoftwarePacketReadAhead: @unchecked Sendable { } private func producePacket(token: UInt64) { + var failureOrigin: FailureOrigin = .sourceRead do { let packet = try readSource { [self] in condition.lock(); defer { condition.unlock() } @@ -514,6 +547,7 @@ final class SoftwarePacketReadAhead: @unchecked Sendable { condition.unlock() return } + failureOrigin = .envelope let data = try packet.encoded() operations.lock() condition.lock() @@ -563,10 +597,10 @@ final class SoftwarePacketReadAhead: @unchecked Sendable { } condition.broadcast() condition.unlock() - } catch { recordFailure(error, token: token) } + } catch { recordFailure(error, token: token, origin: .storage) } } operations.unlock() - } catch { recordFailure(error, token: token) } + } catch { recordFailure(error, token: token, origin: failureOrigin) } } private func shouldParkLocked() -> Bool { @@ -587,10 +621,11 @@ final class SoftwarePacketReadAhead: @unchecked Sendable { return frontierLocked().map { $0 - sourceClock >= forwardSeconds } ?? false } - private func recordFailure(_ error: Error, token: UInt64) { + private func recordFailure(_ error: Error, token: UInt64, origin: FailureOrigin) { condition.lock(); defer { condition.unlock() } guard token == sourceEpoch, !closed else { return } failure = error + failureOrigin = origin resetPending = false condition.broadcast() } diff --git a/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift b/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift index 82f584abd..aa33c98ed 100644 --- a/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift +++ b/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift @@ -14,11 +14,14 @@ final class SoftwarePlaybackHost { /// Codecs whose presentation reordering has an established bound get the read-ahead's /// successor-PTS coverage; every other codec keeps strict packet-duration coverage. HEVC's bound /// is FFmpeg's own: `sps_max_num_reorder_pics > HEVC_MAX_DPB_SIZE - 1` (15) is rejected - /// (n8.1.2 hevc/ps.c:1397-1403), inside the 32-timestamp queue. On the duration model a Matroska + /// (n8.1.2 hevc/ps.c:1397-1403), inside the 32-timestamp queue. VC-1's max_b_frames is at most + /// 7 (n8.1.2 vc1.c:346,426), also within that queue including a field-picture margin. Its + /// 33 ms packet durations over 33/34 ms timestamps otherwise create the same tiny holes. + /// On the duration model a Matroska /// file muxed with 41 ms durations against 41/42 ms deltas split at every 42 ms step (#613). nonisolated static func presentationReorderDepth(codecID: UInt32) -> Int? { switch codecID { - case AV_CODEC_ID_H264.rawValue, AV_CODEC_ID_HEVC.rawValue: return 32 + case AV_CODEC_ID_H264.rawValue, AV_CODEC_ID_HEVC.rawValue, AV_CODEC_ID_VC1.rawValue: return 32 default: return nil } } @@ -175,8 +178,12 @@ final class SoftwarePlaybackHost { /// the wrong thread for it. Serial and separate from `demuxQueue`, whose block never returns. private let seekQueue = DispatchQueue(label: "engine.sw.seek", qos: .userInitiated) - /// Read-deadline budget for a reposition (#254). Matches the side reader's positioning budget. - private static let seekBudgetSeconds: TimeInterval = 8.0 + /// Bound foreground VOD positioning independently of speculative side-reader work. + /// Index-less containers can require several remote keyframe probes; cloud-mounted + /// sources may take 2-3 seconds per successful range response. The former 8-second + /// side-reader budget aborted these progressing seeks. Keep a finite total bound; + /// repositioning still runs off-main and failed/cancelled results remain failures. + static let seekBudgetSeconds: TimeInterval = 30.0 /// True while a reposition is awaiting `seekQueue`. Gates the 0.25 s time tick: the synchronizer /// still carries the pre-seek anchor, so an ungated tick would walk the OLD position forward for @@ -746,7 +753,16 @@ final class SoftwarePlaybackHost { // MARK: - Init - init(videoGravity: AVLayerVideoGravity = .resizeAspect) { + typealias Reposition = @Sendable (Demuxer, Double, TimeInterval, DispatchQueue, + @escaping @Sendable () -> Bool) async -> Demuxer.RepositionOutcome + private let reposition: Reposition + + init(videoGravity: AVLayerVideoGravity = .resizeAspect, + reposition: @escaping Reposition = { dem, target, budget, queue, superseded in + await dem.seekBounded(to: target, timeout: budget, on: queue, + isSuperseded: superseded) + }) { + self.reposition = reposition self.renderer = SampleBufferRenderer(videoGravity: videoGravity) // Default to the software decoder; load() swaps it for the // VT-backed one when the source's video codec is HEVC. @@ -1082,11 +1098,16 @@ final class SoftwarePlaybackHost { let sourceStart = SWClockAnchorPolicy.sourceSeconds(forSession: start, sessionZeroSeconds: zero) // #254: same off-main, deadline-bounded reposition the transport seek uses. A resume into a // remote source that has to scan for its landing would otherwise block the main thread here. - _ = await dem.seekBounded(to: sourceStart, timeout: Self.seekBudgetSeconds, on: seekQueue) + let outcome = await reposition(dem, sourceStart, Self.seekBudgetSeconds, seekQueue, + { [weak self] in self?.stopRequested ?? true }) // This is load()'s only suspension point, so it is also the only place a stop() can land // mid-load. Arming the clock and publishing isReady on a session already torn down would // hand the engine a host it has stopped. - guard !stopRequested else { return } + guard !stopRequested, !Task.isCancelled, outcome != .superseded else { + throw CancellationError() + } + EngineLog.emit("[SWReposition] flags=\(outcome == .landed ? 0 : 1)", category: .swPlayback) + guard outcome == .landed else { throw HostError.repositionFailed } // Mirror seek() skip-PTS + clock alignment so demux drops pre-keyframe frames and synchronizer starts at the resume offset. let startTime = CMTime(seconds: sourceStart, preferredTimescale: 90000) videoDecoder.skipUntilPTS = startTime @@ -1161,6 +1182,7 @@ final class SoftwarePlaybackHost { // MARK: - Transport func play() { + guard !stopRequested, failure == nil else { return } // Resume after pause(): gate on clockArmed, not demuxLoopStarted. A rate change on the // un-anchored synchronizer (no media at its clock time yet) wedges the delayed-rate-change // machinery permanently frozen; the arming seekClock applies the current lastRate (#107). @@ -1470,7 +1492,7 @@ final class SoftwarePlaybackHost { @discardableResult func seek(to seconds: Double) async -> Demuxer.RepositionOutcome { guard !stopRequested else { return .superseded } - guard let dem = demuxer else { return .stalled } + guard failure == nil, let dem = demuxer else { return .stalled } // Stop loop + bump generation to invalidate in-flight packets. Captured right after, so the // reposition can tell on `seekQueue` whether a newer seek has already taken over. bumpSeekGeneration() @@ -1572,23 +1594,31 @@ final class SoftwarePlaybackHost { // existing frontier; seeking it too would duplicate or skip already retained packets. outcome = .landed } else { - outcome = await dem.seekBounded( - to: sourceSeconds, timeout: Self.seekBudgetSeconds, on: seekQueue, - isSuperseded: { [weak self] in + outcome = await reposition(dem, sourceSeconds, Self.seekBudgetSeconds, seekQueue, + { [weak self] in self?.seekGeneration != generation || (self?.stopRequested ?? true) }) } // A newer seek owns the state from here: it published its own target and clears the hold itself. // stop() can also land in the await now that this suspends, and re-arming the clock or flipping // isPlaying on a torn-down session would resurrect a loop that has already been told to quit. - guard seekGeneration == generation, !stopRequested else { return .superseded } + guard seekGeneration == generation, !stopRequested, outcome != .superseded else { return .superseded } seekInFlight = false - if outcome == .stalled { - EngineLog.emit( - "[SWHost] reposition to \(String(format: "%.2f", seconds))s did not complete within " - + "\(String(format: "%.0f", Self.seekBudgetSeconds))s; read position is undefined", - category: .swPlayback - ) + EngineLog.emit("[SWReposition] flags=\(outcome == .landed ? 2 : 3)", category: .swPlayback) + guard outcome == .landed else { + // No target clock, decoder run or cache producer may follow an undefined read position. + // Keep the last observed clock; the engine must not finalize this request as a landing. + isPlaying = false + pausedByHost = true + audioOutput?.setRate(0) + currentTime = clockSessionZero > 0 + ? max(0, sourceClockSeconds - clockSessionZero) : sourceClockSeconds + packetSource?.close() + residentRanges = [] + noteSeekSettled(generation) + failure = PlaybackErrorInfo(kind: .softwarePipelineFailed, + message: "Playback source could not reposition.") + return .stalled } // Re-armed after the landing: a pre-seek frame PAST the target (a backward seek) clears the @@ -3329,7 +3359,7 @@ final class SoftwarePlaybackHost { guard let aOut = self.audioOutput else { return } // #254: a reposition in flight holds `currentTime` at its target; the synchronizer is // still on the pre-seek anchor and would drag the published position backwards. - guard !self.seekInFlight else { return } + guard !self.seekInFlight, self.failure == nil else { return } let raw = aOut.currentTimeSeconds self.emitDiagIfDue(clock: raw) // AE#724: until the first sample arms it, the synchronizer still reads 0, and a paused @@ -3496,11 +3526,12 @@ final class SoftwarePlaybackHost { // MARK: - Errors enum HostError: Error, LocalizedError { - case noVideoStream + case noVideoStream, repositionFailed var errorDescription: String? { switch self { case .noVideoStream: return "Source has no video stream" + case .repositionFailed: return "Playback source could not resume at the requested position." } } } diff --git a/Sources/AetherEngine/Native/SoftwareVideoPacketCoverage.swift b/Sources/AetherEngine/Native/SoftwareVideoPacketCoverage.swift index ae44a14e9..f57c93467 100644 --- a/Sources/AetherEngine/Native/SoftwareVideoPacketCoverage.swift +++ b/Sources/AetherEngine/Native/SoftwareVideoPacketCoverage.swift @@ -1,6 +1,6 @@ import Foundation -/// H.264 and HEVC compressed-packet presentation coverage matching the renderer's successor-PTS +/// H.264, HEVC and VC-1 compressed-packet presentation coverage matching the renderer's successor-PTS /// duration. /// /// AVPacket.duration may describe decode cadence rather than how long a variable-rate picture is @@ -8,14 +8,17 @@ import Foundation /// up to one second). This model publishes only those successor intervals, without rewriting any /// packet timestamp or duration. A larger gap remains a discontinuity, not a tolerance to bridge. /// -/// H.264/HEVC caller contract: complete selected-stream packets, with valid presentation timestamps, +/// H.264/HEVC/VC-1 caller contract: complete selected-stream packets, with valid presentation timestamps, /// enter in demux/decode order after they are retained in the packet store. FFmpeg n8.1.2 h264_ps.c /// rejects num_reorder_frames > 16 and hevc/ps.c rejects sps_max_num_reorder_pics > 15; the default -/// holds 32 distinct timestamps as a conservative field-picture margin. That codec picture bound +/// holds 32 distinct timestamps as a conservative field-picture margin. VC-1's max_b_frames is +/// either read from three bits or set to 7, also within that queue. That codec picture bound /// does NOT prove arbitrary container PTS obey the bound. A timestamp behind the emitted watermark /// therefore invalidates all coverage until reset. /// Sources: https://github.com/FFmpeg/FFmpeg/blob/n8.1.2/libavcodec/h264_ps.c#L174-L180 /// https://github.com/FFmpeg/FFmpeg/blob/n8.1.2/libavcodec/hevc/ps.c#L1397-L1403 +/// https://github.com/FFmpeg/FFmpeg/blob/n8.1.2/libavcodec/vc1.c#L346 +/// https://github.com/FFmpeg/FFmpeg/blob/n8.1.2/libavcodec/vc1.c#L426 /// /// This metadata-only value neither owns packets nor synchronizes access. The owner must reset it /// together with its packet store on seek/flush/discard. Other codecs must keep their strict packet diff --git a/Sources/AetherEngine/SeekEvent.swift b/Sources/AetherEngine/SeekEvent.swift index 15179bf8e..784ee7412 100644 --- a/Sources/AetherEngine/SeekEvent.swift +++ b/Sources/AetherEngine/SeekEvent.swift @@ -51,8 +51,9 @@ public struct SeekEvent: Sendable, Equatable { /// The picture is at the target: AVPlayer reported the landing, or the rendered frame reached the /// scrub target. This is the only outcome that means "safe to read the clock". case landed(renderedTime: Double) - /// The engine spent its recovery budget without the source serving the target. The clock is held - /// AT the target and the seek stays alive, so a `.landed` under this `id` can still follow. + /// Positioning did not complete within its budget. Native AVPlayer recovery may keep the + /// target pending and later emit `.landed` under this `id`. A software positioning failure + /// instead reports a terminal pipeline error without finalizing the requested clock. case stalled /// A newer seek took over before this one settled; the newer one's `.began` carries its target. case superseded diff --git a/Tests/AetherEngineTests/DetourOriginTicketTests.swift b/Tests/AetherEngineTests/DetourOriginTicketTests.swift index 1b8a5567a..b3a0f4f48 100644 --- a/Tests/AetherEngineTests/DetourOriginTicketTests.swift +++ b/Tests/AetherEngineTests/DetourOriginTicketTests.swift @@ -58,5 +58,55 @@ struct DetourOriginTicketTests { // exactly that range: the pump's refills are 32 MB and the tail prefetch is a suffix. #expect(server.requestedRanges.contains { $0.start == 4 * 1024 * 1024 && $0.end == 8 * 1024 * 1024 - 1 }, "no detour block went out, so this measured nothing: \(server.requestedRanges)") + + // Arming a seek deadline must still reuse that resident block without network I/O. + let requestsBeforeHit = server.rangeRequestCount + reader.beginReadDeadline(secondsFromNow: 8) + #expect(reader.seek(offset: target, whence: Int32(SEEK_SET)) == target) + #expect(reader.read(into: buf, size: Int32(chunk)) == chunk) + #expect(server.rangeRequestCount == requestsBeforeHit) + reader.endReadDeadline() + } + + @Test("bounded cold probing reanchors without a speculative block, then restores normal detours", + .timeLimit(.minutes(2))) + func boundedProbeSkipsColdDetour() throws { + let server = try #require(ThrottledOriginServer(totalSize: 64 * 1024 * 1024, + throttleUs: 40_000, patternedBody: true)) + defer { server.stop() } + let reader = AVIOReader(url: URL(string: "http://127.0.0.1:\(server.port)/bounded.bin")!) + defer { reader.markClosed(); reader.close() } + try reader.open() + let chunk = 256 * 1024 + let buf = UnsafeMutablePointer.allocate(capacity: chunk) + defer { buf.deallocate() } + func readToThirteenMB(from start: Int) throws { + var position = start + while position < 13 * 1024 * 1024 { + let n = Int(reader.read(into: buf, size: Int32(chunk))) + try #require(n > 0) + position += n + } + } + try readToThirteenMB(from: 0) + #expect(reader.hasLiveConnectionForTesting) + let target: Int64 = 5 * 1024 * 1024 + reader.beginReadDeadline(secondsFromNow: 8) + #expect(reader.seek(offset: target, whence: Int32(SEEK_SET)) == target) + let n = Int(reader.read(into: buf, size: Int32(chunk))) + #expect(n == chunk) + #expect(!reader.readDeadlineFired) + #expect((0..(_ body: (URL) async throws -> T) async throws -> T { + let root = FileManager.default.temporaryDirectory + .appendingPathComponent("vc1-coverage-\(UUID().uuidString)", isDirectory: true) + try FileManager.default.createDirectory(at: root, withIntermediateDirectories: true) + defer { try? FileManager.default.removeItem(at: root) } + return try await body(root) + } + + /// Synthetic VC-1 timing only: one reference followed by up to seven B pictures, + /// 30 fps rounded to milliseconds, and presentation-ordered audio with explicit duration. + /// Missing audio duration is covered separately by upstream #775, not this VC-1 regression. + /// No private media, packet payloads or source identifiers are needed to reproduce it. + private func vc1ReadAhead(root: URL, count: Int, strict: Bool = false, + discontinuity: Bool = false) async throws -> SoftwarePacketReadAhead { + var order = [0] + for first in stride(from: 1, to: count, by: 8) { + let reference = min(first + 7, count - 1) + order.append(reference) + order.append(contentsOf: first.. Int64 { + Int64(index) * 100 / 3 + (discontinuity && index >= 150 ? 2000 : 0) + } + for (decodeIndex, picture) in order.enumerated() { + packets.append(SoftwareStoredPacket(pts: timestamp(picture), dts: timestamp(decodeIndex), + duration: 33, position: 0, streamIndex: 0, flags: picture % 240 == 0 ? 1 : 0, + timeBaseNumerator: 1, timeBaseDenominator: 1000, + bytes: Data(repeating: 7, count: 100), sideData: [])) + if decodeIndex % 5 == 0 { + packets.append(SoftwareStoredPacket(pts: timestamp(decodeIndex), dts: timestamp(decodeIndex), + duration: Int64(decodeIndex + 5) * 100 / 3 - Int64(decodeIndex) * 100 / 3, + position: 0, streamIndex: 1, flags: 1, + timeBaseNumerator: 1, timeBaseDenominator: 1000, + bytes: Data(repeating: 3, count: 16), sideData: [])) + } + } + let all = packets + let counter = Counter() + let fifo = try SoftwarePacketDiskFIFO(chunkTargetBytes: 64 * 1024, retainConsumed: true, + parentDirectory: root) + let source = SoftwarePacketReadAhead(video: .init(index: 0, numerator: 1, denominator: 1000), + audio: .init(index: 1, numerator: 1, denominator: 1000), byteBudget: 50_000_000, + forwardSeconds: 10_000, initialSourceClock: 0, fifo: fifo, + videoReorderDepth: strict ? nil : SoftwarePlaybackHost.presentationReorderDepth( + codecID: AV_CODEC_ID_VC1.rawValue)) { _ in + let index = counter.next() - 1 + return index < all.count ? all[index] : nil + } + source.start() + do { try await waitFor { source.snapshot.sourceEnded } } + catch { source.close(); throw error } + return source + } + + @Test("VC-1 rounded durations and seven-picture reorder retain one long A/V range", .timeLimit(.minutes(2))) + func vc1RoundedDurationsAndCacheSeek() async throws { + try await withAsyncRoot { root in + let old = try await vc1ReadAhead(root: root, count: 15_000, strict: true) + #expect(old.residentRanges.count == 4096, "old codec selection saturates coverage") + #expect((old.residentRanges.last?.upperBound ?? 0) < 420) + old.close() + + let source = try await vc1ReadAhead(root: root, count: 15_000) + defer { source.close() } + #expect(source.residentRanges.count == 1) + #expect((source.residentRanges.last?.upperBound ?? 0) > 499) + // A former one-millisecond hole, then a retained position beyond the old cap. + for target in [0.0995, 450.0] { + let token = source.beginSeek(to: target) + #expect(try source.prepareSeek(token, to: target)) + } + source.close() + #expect(source.residentRanges.isEmpty, "discard must clear the published cache") + } + } + + @Test("VC-1 successor coverage preserves a genuine two-second A/V discontinuity", .timeLimit(.minutes(2))) + func vc1DiscontinuityRemainsUncached() async throws { + try await withAsyncRoot { root in + let source = try await vc1ReadAhead(root: root, count: 600, discontinuity: true) + defer { source.close() } + #expect(source.residentRanges.count == 2) + #expect(!source.residentRanges.contains { $0.contains(6) }) + let token = source.beginSeek(to: 6) + #expect(try source.prepareSeek(token, to: 6) == false) + } + } + + @Test("VC-1 admission retains the existing fail-closed late-timestamp guard") + func vc1LateTimestampInvalidates() throws { + let depth = try #require(SoftwarePlaybackHost.presentationReorderDepth(codecID: AV_CODEC_ID_VC1.rawValue)) + var model = SoftwareVideoPacketCoverage(timeBaseNumerator: 1, timeBaseDenominator: 1000, + reorderDepth: depth) + for index: Int64 in 0..<100 { model.insert(pts: index * 100 / 3) } + #expect(!model.rangesSeconds.isEmpty) + let admitted = model.insert(pts: 1) + #expect(!admitted) + #expect(model.isInvalidated) + #expect(model.rangesSeconds.isEmpty) + } + // MARK: - 2. A full coverage forgets history instead of freezing /// Consumes `count` packets, moving the playhead onto each, and answers the frontier at the end. diff --git a/Tests/AetherEngineTests/SoftwareRepositionResultTests.swift b/Tests/AetherEngineTests/SoftwareRepositionResultTests.swift new file mode 100644 index 000000000..16792043c --- /dev/null +++ b/Tests/AetherEngineTests/SoftwareRepositionResultTests.swift @@ -0,0 +1,152 @@ +import Foundation +import Combine +import Testing +@testable import AetherEngine + +@Suite(.serialized, .timeLimit(.minutes(2))) +@MainActor +struct SoftwareRepositionResultTests { + actor Gate { + private var waits: [Int: CheckedContinuation] = [:] + private(set) var count = 0 + func next() async -> Demuxer.RepositionOutcome { + let id = count; count += 1 + return await withCheckedContinuation { waits[id] = $0 } + } + func finish(_ id: Int, _ result: Demuxer.RepositionOutcome) { + waits.removeValue(forKey: id)?.resume(returning: result) + } + } + + private func demuxer() throws -> Demuxer { + let dem = Demuxer() + try dem.open(reader: DataIOReader(data: ProbeTestFixtures.hdr10Plus()), formatHint: "mp4") + return dem + } + + private func loadedHost(_ reposition: @escaping SoftwarePlaybackHost.Reposition) async throws -> SoftwarePlaybackHost { + let host = SoftwarePlaybackHost(reposition: reposition) + do { + try await host.load(demuxer: demuxer(), startPosition: nil, audioSourceStreamIndex: nil) + return host + } catch { host.stop(); throw error } + } + + @Test("Failed resume never publishes readiness, target clock or first-frame readiness") + func failedResume() async throws { + let host = SoftwarePlaybackHost(reposition: { _, _, _, _, _ in .stalled }) + defer { host.stop() } + do { + try await host.load(demuxer: demuxer(), startPosition: 42, audioSourceStreamIndex: nil) + Issue.record("failed resume returned successfully") + } catch SoftwarePlaybackHost.HostError.repositionFailed {} catch { throw error } + #expect(!host.isReady) + #expect(!host.isPlaying) + #expect(!host.isVideoReadyForDisplay) + #expect(host.currentTime == 0) + #expect(host.sourceClockSeconds == 0) + } + + @Test("Successful resume retains its anchor without claiming a displayed first frame") + func successfulResume() async throws { + let host = SoftwarePlaybackHost(reposition: { _, _, _, _, _ in .landed }) + defer { host.stop() } + try await host.load(demuxer: demuxer(), startPosition: 42, audioSourceStreamIndex: nil) + #expect(host.isReady) + #expect(host.currentTime == 42) + #expect(!host.isVideoReadyForDisplay) + #expect(!host.isPlaying) + } + + @Test("Stop during resume cannot publish readiness or resurrect the target clock") + func stoppedResume() async throws { + let gate = Gate() + let host = SoftwarePlaybackHost(reposition: { _, _, _, _, _ in await gate.next() }) + let dem = try demuxer() + let load = Task { try await host.load(demuxer: dem, startPosition: 42, audioSourceStreamIndex: nil) } + try await waitFor { await gate.count >= 1 } + host.stop() + await gate.finish(0, .landed) + do { try await load.value; Issue.record("stopped resume returned successfully") } + catch is CancellationError {} catch { throw error } + #expect(!host.isReady) + #expect(host.currentTime == 0) + #expect(!host.isPlaying) + } + + @Test("Caller cancellation during resume cannot publish a successful load") + func cancelledResume() async throws { + let gate = Gate() + let host = SoftwarePlaybackHost(reposition: { _, _, _, _, _ in await gate.next() }) + defer { host.stop() } + let dem = try demuxer() + let load = Task { try await host.load(demuxer: dem, startPosition: 42, audioSourceStreamIndex: nil) } + try await waitFor { await gate.count >= 1 } + load.cancel() + await gate.finish(0, .landed) + do { try await load.value; Issue.record("cancelled resume returned successfully") } + catch is CancellationError {} catch { throw error } + #expect(!host.isReady) + #expect(!host.isPlaying) + #expect(host.currentTime == 0) + } + + @Test("Failed software seek stays failed through engine finalization and keeps its request identity", arguments: [false, true]) + func failedSeek(wireFailure: Bool) async throws { + let host = try await loadedHost { _, _, _, _, _ in .stalled } + let engine = try AetherEngine() + engine.softwareHost = host + engine.duration = 100 + engine.state = .paused + defer { engine.stop() } + let failureSink = wireFailure ? host.$failure.compactMap { $0 }.sink { engine.publishError($0) } : nil + defer { failureSink?.cancel() } + var events: [SeekEvent] = [] + let sub = engine.seekEvents.sink { events.append($0) } + defer { sub.cancel() } + await engine.seek(to: 42) + #expect(host.failure?.kind == .softwarePipelineFailed) + #expect(!host.isPlaying) + #expect(host.currentTime != 42) + #expect(engine.currentTime != 42) + #expect(!engine.isSeeking) + guard case .error = engine.state else { Issue.record("failed seek was finalized as playable"); return } + #expect(events.map(\.outcome) == [.began, .stalled]) + #expect(Set(events.map(\.id)).count == 1) + host.play() + #expect(!host.isPlaying) + } + + @Test("A late failed seek cannot poison a newer successful paused seek") + func supersededFailure() async throws { + let gate = Gate() + let host = try await loadedHost { _, _, _, _, _ in await gate.next() } + let engine = try AetherEngine() + engine.softwareHost = host + engine.duration = 100 + engine.state = .paused + defer { engine.stop() } + var events: [SeekEvent] = [] + let sub = engine.seekEvents.sink { events.append($0) } + defer { sub.cancel() } + let first = Task { await engine.seek(to: 12) } + try await waitFor { await gate.count >= 1 } + let second = Task { await engine.seek(to: 24) } + try await waitFor { await gate.count >= 2 } + await gate.finish(0, .stalled) + await first.value + #expect(host.failure == nil) + await gate.finish(1, .landed) + await second.value + #expect(host.currentTime == 24) + #expect(engine.currentTime == 24) + #expect(engine.state == .paused) + #expect(!engine.isSeeking) + #expect(host.failure == nil) + #expect(events.filter { $0.target == 12 }.map(\.outcome) == [.began, .superseded]) + #expect(events.filter { $0.target == 24 }.count == 2) + #expect(events.last?.target == 24) + if case .landed = events.last?.outcome {} else { Issue.record("newest seek did not land") } + } + +} diff --git a/Tests/AetherEngineTests/SoftwareResidentRangesTests.swift b/Tests/AetherEngineTests/SoftwareResidentRangesTests.swift index 8cd47c854..85cafd566 100644 --- a/Tests/AetherEngineTests/SoftwareResidentRangesTests.swift +++ b/Tests/AetherEngineTests/SoftwareResidentRangesTests.swift @@ -141,6 +141,124 @@ struct SoftwareResidentRangesTests { #expect(value.videoFloor == -.infinity) } + private func consume(_ source: SoftwarePacketReadAhead, + expecting packets: [SoftwareStoredPacket], fails: Bool) async throws { + try await withTaskCancellationHandler { + try await withCheckedThrowingContinuation { (continuation: CheckedContinuation) in + Thread.detachNewThread { + do { + for expected in packets { #expect(try source.read() == expected) } + if fails { #expect(throws: SourceFailure.self) { _ = try source.read() } } + else { #expect(try source.read() == nil) } + continuation.resume() + } catch { continuation.resume(throwing: error) } + } + } + } onCancel: { source.close() } + } + + private enum SourceFailure: Error { case unavailable } + private final class FailingInput: @unchecked Sendable { + private let lock = NSLock() + private var packets: [SoftwareStoredPacket] + private var index = 0 + private var failAtEnd = true + let reachedEnd = DispatchSemaphore(value: 0) + let releaseEnd = DispatchSemaphore(value: 0) + init(_ packets: [SoftwareStoredPacket]) { self.packets = packets } + func next() throws -> SoftwareStoredPacket? { + lock.lock() + if index < packets.count { + let packet = packets[index]; index += 1 + lock.unlock(); return packet + } + let fail = failAtEnd + lock.unlock() + if fail { + reachedEnd.signal(); releaseEnd.wait() + throw SourceFailure.unavailable + } + return nil + } + func replace(_ packets: [SoftwareStoredPacket]) { + lock.lock(); defer { lock.unlock() } + self.packets = packets; index = 0; failAtEnd = false + } + } + + @Test("A source failure preserves committed coverage and cached seeks, but never becomes EOF", + arguments: [false, true]) + func retainedAfterSourceFailure(reordered: Bool) async throws { + let packets = (0..<100).map { packet(Int64($0) * 40, key: $0 == 0) } + let input = FailingInput(packets) + let fifo = try SoftwarePacketDiskFIFO(chunkTargetBytes: 1024, retainConsumed: true) + let source = SoftwarePacketReadAhead(video: .init(index: 0, numerator: 1, denominator: 1000), + audio: nil, byteBudget: 1_000_000, forwardSeconds: 3600, initialSourceClock: 0, + fifo: fifo, videoReorderDepth: reordered ? 4 : nil) { _ in try input.next() } + defer { input.releaseEnd.signal(); source.close() } + source.start() + try await waitFor { input.reachedEnd.wait(timeout: .now()) == .success } + let before = source.residentRanges + #expect(!before.isEmpty) + let epoch = source.snapshot.sourceEpoch + input.releaseEnd.signal() + try await waitFor { source.producerParkedForTesting } + #expect(source.producerParkedForTesting) + #expect(!source.snapshot.sourceEnded) + #expect(source.residentRanges == before) + // A failure cannot flush unresolved successor PTS as though it proved a normal EOF. + if reordered { #expect((before.last?.upperBound ?? 4) < 3.96) } + try await consume(source, expecting: packets, fails: true) + for target in [0.5, 2.0, 1.0] { + let token = source.beginSeek(to: target) + #expect(try source.prepareSeek(token, to: target)) + source.endSeek(token, sourceClock: target) + #expect(source.snapshot.sourceEpoch == epoch) + #expect(source.residentRanges == before) + try await consume(source, expecting: packets, fails: true) + #expect(!source.snapshot.sourceEnded) + } + // A miss still resets the failed source lifetime and accepts only the new epoch's data. + let replacement = (0..<100).map { packet(10_000 + Int64($0) * 40, key: $0 == 0) } + let miss = source.beginSeek(to: 10) + #expect(try !source.prepareSeek(miss, to: 10)) + #expect(source.residentRanges.isEmpty) + input.replace(replacement) + source.endSeek(miss, sourceClock: 10) + try await waitFor { source.snapshot.sourceEnded } + #expect(source.snapshot.sourceEpoch == epoch + 1) + #expect(source.residentRanges.first?.lowerBound == 10) + try await consume(source, expecting: replacement, fails: false) + } + + @Test("A storage append failure still invalidates retained coverage and cached seeks") + func storageFailureRemainsClosed() async throws { + let packets = (0..<10).map { packet(Int64($0) * 40) } + let input = Input(packets) + let fifo = try SoftwarePacketDiskFIFO(chunkTargetBytes: 1024, retainConsumed: true) + let reached = DispatchSemaphore(value: 0), release = DispatchSemaphore(value: 0) + let source = SoftwarePacketReadAhead(video: .init(index: 0, numerator: 1, denominator: 1000), + audio: nil, byteBudget: 1_000_000, forwardSeconds: 3600, initialSourceClock: 0, + fifo: fifo) { _ in + if let packet = input.next() { return packet } + reached.signal(); release.wait() + return packets[0] + } + defer { release.signal(); source.close() } + source.start() + try await waitFor { reached.wait(timeout: .now()) == .success } + #expect(source.residentRanges == [0...0.4]) + try fifo.close() + release.signal() + try await waitFor { source.producerParkedForTesting } + #expect(source.producerParkedForTesting) + #expect(!source.snapshot.sourceEnded) + #expect(source.residentRanges.isEmpty) + let miss = source.beginSeek(to: 0.1) + #expect(try !source.prepareSeek(miss, to: 0.1)) + #expect(source.snapshot.cacheSeekHits == 0) + } + @Test("Source origin mapping preserves gaps, clips negatives and rejects unknown origins") func displayMapping() { let spans = [90.0...105, 110...120] diff --git a/docs/api.md b/docs/api.md index 7ecfb3ae4..1d7cd87a2 100644 --- a/docs/api.md +++ b/docs/api.md @@ -590,7 +590,7 @@ Time lives on `player.clock`, a separate `ObservableObject`, so ~10 Hz ticks nev | `$duration` | seconds; a `LoadOptions.declaredDurationSeconds` outranks the container's | For software VOD, `residentRanges` intersects video presentation coverage with the selected audio -stream, even if muted; a video-only source does not require audio. H.264/HEVC use verified +stream, even if muted; a video-only source does not require audio. H.264/HEVC and VC-1 use verified successor-PTS intervals, so the final picture with no known successor does not acquire an invented end. Ranges are clipped against retained keyframes and discarded packet timestamps; unknown or invalid coverage stays empty. Metadata limits can conservatively omit old spans rather than claim @@ -604,6 +604,25 @@ cannot establish a backward cache start. Live playback continues to expose its s window rather than VOD resident ranges. +A source read failure preserves already committed software packets and their verified ranges. +Cached seeks may replay them, but draining those packets still throws the original read failure; +the error does not become EOF and does not flush unresolved successor timestamps. Storage or +packet-envelope failures invalidate usability instead. A cold seek starts a new source lifetime. + +Software VOD resume and cache-miss seek positioning run off the main actor with a **30 s positioning +budget**, measured after acquiring demux access. Waiting for an earlier read can add to the total +call time. A deadline-bounded cold byte probe uses the existing streaming reconnect immediately; +already resident detour blocks remain usable, and ordinary unbounded parsing keeps its detour policy. +This allows multi-probe keyframe searches on slow remote storage to complete while retaining a finite +I/O deadline. An unavailable source can consequently take longer to report failure. + +Failed software resume throws before readiness or a requested clock anchor is published. Failed +software seek reports `.stalled` and `.softwarePipelineFailed`, clears the in-flight state, and does +not resume decoding or finalize the target as landed. Unlike an AVPlayer seek that remains pending +for a late landing, this software failure is terminal for the host. Stop, cancellation and newer +seeks still prevent stale completions from changing the active session. + + **`sourceTime` on `nativeRemoteHLS` (AE#616).** The engine sees no segment on that route, so the clock is AVPlayer's item time. An origin whose playlist places a segment at its slot while the segment starts at the keyframe before it (a Jellyfin transcode restarted by `-ss -noaccurate_seek -copyts`) makes