diff --git a/CHANGELOG.md b/CHANGELOG.md index 633aa195..f81122eb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,7 +10,20 @@ the public-API contract. ## [Unreleased] -_Nothing yet._ +### Fixed + +- A timestamp leap that escapes the timeline rebase no longer turns a VOD session into a + long-lived zombie (#369). Three containment gaps, one field trace: the look-behind sample + duration is now capped at the discontinuity threshold instead of handing movenc the wrap + itself as a duration (device: 8226410192 ticks, rejected as invalid, packet silently lost — + the write rc is now logged on first failure too); discontinuity-scale fold runs now reach + the fold counters instead of being discarded above 64 indices, so the #358 recovery arms + actually arm for exactly the folds most certain to trigger them; and the advance-path + backpressure park skips a release target beyond the sequential playlist's advertisable + frontier, which only this pump's own finalize reports can move — parking on it was waiting + for oneself. Deliberately unchanged: `OutputTimestampSanitizer` keeps latching, because + movenc latches monotonicity on its own once a wrapped packet is accepted, and a sanitizer + reset would only convert garbage timestamps into rejected writes. ## [6.25.1] - 2026-08-13 diff --git a/Sources/AetherEngine/Video/HLSSegmentProducer.swift b/Sources/AetherEngine/Video/HLSSegmentProducer.swift index 0c3e74b9..abec5c75 100644 --- a/Sources/AetherEngine/Video/HLSSegmentProducer.swift +++ b/Sources/AetherEngine/Video/HLSSegmentProducer.swift @@ -205,6 +205,15 @@ final class HLSSegmentProducer: @unchecked Sendable { private var seqNextReportIndex: Int? = nil private var seqReadyReports: [Int: Double] = [:] + /// #369: highest sequential index the append playlist can currently advertise (only entries + /// with a real duration get a URI). Pump-thread only; read by the advance park to detect a + /// release target beyond the advertisable frontier. + private var seqHighestAdvertisedIndex = Int.min + + /// #369: one-shot latches for the containment logs. Pump-thread only. + private var loggedVideoWriteFailure = false + private var loggedSequentialParkSkip = false + /// Order-preserving funnel for sequential finalize reports. private func emitSequentialReport(index: Int, duration: Double) { let base = seqNextReportIndex ?? index @@ -213,11 +222,21 @@ final class HLSSegmentProducer: @unchecked Sendable { var next = base while let d = seqReadyReports.removeValue(forKey: next) { onSequentialSegmentFinalized?(next, d) + if d > 0 { seqHighestAdvertisedIndex = next } // #369: zero-duration holes get no URI next += 1 } seqNextReportIndex = next } + /// #369: the advance park releases on a consumer fetch of `target`, but a sequential append + /// playlist can only advertise up to the frontier this pump's OWN finalize reports have fed + /// it — parking on an index beyond that is waiting for oneself (field case: a fold-to-tail + /// parked at target=364 while the playlist ended at seg61). A negative target releases + /// instantly, so it is no deadlock. Pure for the unit test. + static func sequentialParkWouldSelfDeadlock(target: Int, highestAdvertised: Int) -> Bool { + target >= 0 && target > highestAdvertised + } + /// Forward discontinuity threshold. Distinct from NOPTS-dts repair (+1 tick scale); only fires on genuine multi-second leaps. static let discontinuityThresholdSeconds: Double = 10.0 @@ -1680,6 +1699,14 @@ final class HLSSegmentProducer: @unchecked Sendable { if !isLive, newIdx > currentMuxerSegmentIndex + 1 { let folded = (currentMuxerSegmentIndex + 1).. SegmentCache.maxFoldRunLength { + // #369: a leap this wide is not a long GOP — a timeline jump escaped the rebase. + EngineLog.emit( + "[HLSSegmentProducer] #369 cut leap of \(folded.count) plan indices " + + "(discontinuity-scale; a timeline jump escaped the rebase)", + category: .session + ) + } EngineLog.emit( "[HLSSegmentProducer] #358 plan indices \(folded.lowerBound)...\(folded.upperBound - 1) " + "folded into seg-\(newIdx) (no IRAP reached their boundary)", @@ -1697,7 +1724,24 @@ final class HLSSegmentProducer: @unchecked Sendable { if !awaitLiveWindowHeadroom(head: newIdx) { return nil } } else { let backpressureTarget = newIdx - bufferAheadSegments - if !awaitBackpressureRelease(target: backpressureTarget, head: newIdx, context: "advance") { return nil } + if onSequentialSegmentFinalized != nil, + Self.sequentialParkWouldSelfDeadlock(target: backpressureTarget, + highestAdvertised: seqHighestAdvertisedIndex) { + // #369: skip the self-deadlocking park; the disk budget below stays the resource + // bound, and normal parking resumes as soon as the frontier catches back up. + if !loggedSequentialParkSkip { + loggedSequentialParkSkip = true + EngineLog.emit( + "[HLSSegmentProducer] #369 backpressure park skipped: " + + "target=\(backpressureTarget) is beyond the advertisable " + + "frontier=\(seqHighestAdvertisedIndex); the frontier only advances " + + "while this pump runs", + category: .session + ) + } + } else if !awaitBackpressureRelease(target: backpressureTarget, head: newIdx, context: "advance") { + return nil + } if !awaitPrefetchDiskBudgetRelease(head: newIdx, context: "advance") { return nil } } if checkShouldStop() { return nil } @@ -3282,19 +3326,33 @@ final class HLSSegmentProducer: @unchecked Sendable { /// (`Packet duration: -N ... out of range` -> DTS clamp + `pts has no value` -> wrong trun timing, /// the #92 transient blocky glitch). Falls back to the source packet's own positive duration, then /// `fallback`, only when no usable forward delta exists (EOF tail, NOPTS, or a non-increasing next). + /// #369: `capTicks` bounds the inferred delta — a sample longer than a discontinuity is + /// definitionally invalid. Across a 33-bit PTS wrap the look-behind delta IS the wrap + /// (device: 8226410192 ticks ≈ 91404 s; movenc rejects it as "Application provided duration + /// ... is invalid" and the packet is lost), so an over-cap delta falls back like a + /// non-forward one. static func resolveVideoSampleDuration( existingDuration: Int64, dts: Int64, nextDts: Int64?, - fallback: Int64 + fallback: Int64, + capTicks: Int64 ) -> Int64 { if let next = nextDts, dts != Int64.min, next != Int64.min { let inferred = next - dts - if inferred > 0 { return inferred } + if inferred > 0, inferred <= capTicks { return inferred } } return existingDuration > 0 ? existingDuration : fallback } + /// #369: the sample-duration cap in source-video ticks (`discontinuityThresholdSeconds`, so + /// live rebases and the duration cap share one definition of "discontinuity"). + private var videoSampleDurationCapTicks: Int64 { + sourceVideoTbSeconds > 0 + ? Int64(Self.discontinuityThresholdSeconds / sourceVideoTbSeconds) + : Int64.max + } + private func finalizeAndWriteVideo( _ packet: UnsafeMutablePointer, nextDts: Int64?, @@ -3304,7 +3362,8 @@ final class HLSSegmentProducer: @unchecked Sendable { existingDuration: packet.pointee.duration, dts: packet.pointee.dts, nextDts: nextDts, - fallback: videoFallbackDurationPts + fallback: videoFallbackDurationPts, + capTicks: videoSampleDurationCapTicks ) packet.pointee.stream_index = muxer.videoOutputStreamIndex @@ -3355,7 +3414,19 @@ final class HLSSegmentProducer: @unchecked Sendable { let frameSegmentIndex = currentMuxerSegmentIndex av_packet_rescale_ts(packet, sourceVideoTimeBase, muxer.muxerVideoTimeBase) - let written = muxer.writePacket(packet).written + let write = muxer.writePacket(packet) + if write.rc < 0, !loggedVideoWriteFailure { + // #369: this rc used to be dropped on the floor; the field failure (movenc rejecting a + // wrap-scale sample duration) was only findable through libav's own stderr line. + loggedVideoWriteFailure = true + EngineLog.emit( + "[HLSSegmentProducer] #369 video packet write failed rc=\(write.rc) " + + "dts=\(packet.pointee.dts) duration=\(packet.pointee.duration) (muxer TB; " + + "first occurrence only)", + category: .session + ) + } + let written = write.written if let frameObserver, let source = Self.cmTime(ticks: frameSourcePts, timeBase: sourceVideoTimeBase), diff --git a/Sources/AetherEngine/Video/SegmentCache.swift b/Sources/AetherEngine/Video/SegmentCache.swift index 7afdbefe..4d9cc4ea 100644 --- a/Sources/AetherEngine/Video/SegmentCache.swift +++ b/Sources/AetherEngine/Video/SegmentCache.swift @@ -55,8 +55,11 @@ final class SegmentCache: @unchecked Sendable { /// Plan index -> how many pumps passed it without opening a segment (#358). Survives producer /// restarts on purpose: the repeat across a restart is the signal. private var foldCounts: [Int: Int] = [:] - /// Above this, a jump is a reposition rather than a fold. - private static let maxFoldRunLength = 64 + /// #369: log-classification threshold — a run wider than this is a discontinuity-scale cut + /// leap, not a long GOP. (It used to DROP such runs from the counters on the assumption they + /// were repositions; the field case was a 2^33 wrap folding 312 indices, and dropping it left + /// every fold counter at 0, which is exactly what disarms the #358 recovery arms.) + static let maxFoldRunLength = 64 /// (10, 20)=30 entries, ~300 MB at 4K HDR HEVC ~10 MB/seg. init(forwardWindow: Int = 10, backwardWindow: Int = 20, retentionBudgetBytes: Int = 0) { @@ -372,13 +375,13 @@ final class SegmentCache: @unchecked Sendable { /// Record plan indices a cut jumped over. VOD only: a live playlist is built from what was /// finalized, so it never offers an index the pump skipped. + /// #369: runs wider than `maxFoldRunLength` count too — the #358 arms exist precisely for a + /// consumer that requests a folded index, and the widest folds are the ones most certain to + /// produce such a request. Memory is one Int per folded index, bounded by the plan size. func noteFolded(_ indices: Range) { guard !indices.isEmpty else { return } condition.lock() defer { condition.unlock() } - // A jump this wide is a restart or a seek, not a fold; counting it would grow the table - // without describing anything. - guard indices.count <= Self.maxFoldRunLength else { return } for index in indices where entries[index] == nil { foldCounts[index, default: 0] += 1 } diff --git a/Tests/AetherEngineTests/SegmentCacheFoldRunTests.swift b/Tests/AetherEngineTests/SegmentCacheFoldRunTests.swift new file mode 100644 index 00000000..035edff1 --- /dev/null +++ b/Tests/AetherEngineTests/SegmentCacheFoldRunTests.swift @@ -0,0 +1,53 @@ +// #369: discontinuity-scale fold runs must reach the fold counters. The old guard dropped runs +// wider than maxFoldRunLength on the assumption they were repositions; the field case was a 2^33 +// wrap folding 312 indices in one cut, and dropping it left every counter at 0 — which disarmed +// both #358 recovery arms (the consumer-side reanchor and the engine's unrecoverable-gap handler) +// for exactly the folds most certain to trigger them. +import Foundation +import Testing +@testable import AetherEngine + +@Suite("Fold-run counting (#369)") +struct SegmentCacheFoldRunTests { + + @Test("A discontinuity-scale run counts every folded index") + func wideRunCounts() { + let cache = SegmentCache() + defer { cache.close() } + cache.noteFolded(2..<314) // the field fold: indices 2...313 into seg-314 + #expect(cache.foldCount(2) == 1) + #expect(cache.foldCount(160) == 1) + #expect(cache.foldCount(313) == 1) + #expect(cache.foldCount(314) == 0) // the tail segment itself was opened, not folded + } + + @Test("A repeat across a producer restart is the #358 signal and increments") + func repeatAcrossPumpsIncrements() { + let cache = SegmentCache() + defer { cache.close() } + cache.noteFolded(2..<314) + cache.noteFolded(2..<314) + #expect(cache.foldCount(2) == 2) + #expect(cache.foldCount(313) == 2) + } + + @Test("An index that produced a segment is never counted as folded") + func storedIndexIsNotCounted() { + let cache = SegmentCache() + defer { cache.close() } + cache.store(index: 5, data: Data([0x00])) + cache.noteFolded(2..<10) + #expect(cache.foldCount(5) == 0) + #expect(cache.foldCount(6) == 1) + } + + @Test("Narrow runs behave as before") + func narrowRunUnchanged() { + let cache = SegmentCache() + defer { cache.close() } + cache.noteFolded(7..<9) + #expect(cache.foldCount(7) == 1) + #expect(cache.foldCount(8) == 1) + #expect(cache.foldCount(9) == 0) + } +} diff --git a/Tests/AetherEngineTests/SequentialParkFrontierTests.swift b/Tests/AetherEngineTests/SequentialParkFrontierTests.swift new file mode 100644 index 00000000..3a0698b5 --- /dev/null +++ b/Tests/AetherEngineTests/SequentialParkFrontierTests.swift @@ -0,0 +1,28 @@ +// #369: the advance park's release is a consumer fetch of `target`, but a sequential append +// playlist advertises only what the pump's own finalize reports have fed it. A target beyond +// that frontier is a self-deadlock (field: fold-to-tail parked at target=364 while the playlist +// ended at seg61, freezing it for good); a negative target releases instantly and is none. +import Testing +@testable import AetherEngine + +@Suite("Sequential park frontier (#369)") +struct SequentialParkFrontierTests { + + @Test("A target beyond the advertisable frontier is a self-deadlock") + func beyondFrontierDeadlocks() { + #expect(HLSSegmentProducer.sequentialParkWouldSelfDeadlock(target: 364, highestAdvertised: 61)) + #expect(HLSSegmentProducer.sequentialParkWouldSelfDeadlock(target: 0, highestAdvertised: Int.min)) + } + + @Test("A target the playlist already advertises parks normally") + func atOrBelowFrontierParks() { + #expect(!HLSSegmentProducer.sequentialParkWouldSelfDeadlock(target: 52, highestAdvertised: 60)) + #expect(!HLSSegmentProducer.sequentialParkWouldSelfDeadlock(target: 60, highestAdvertised: 60)) + } + + @Test("A negative target releases instantly and never counts as a deadlock") + func negativeTargetIsNoDeadlock() { + // Session start: head < window makes the target negative while the frontier is still empty. + #expect(!HLSSegmentProducer.sequentialParkWouldSelfDeadlock(target: -8, highestAdvertised: Int.min)) + } +} diff --git a/Tests/AetherEngineTests/VideoSampleDurationTests.swift b/Tests/AetherEngineTests/VideoSampleDurationTests.swift index 27a459fb..7884137c 100644 --- a/Tests/AetherEngineTests/VideoSampleDurationTests.swift +++ b/Tests/AetherEngineTests/VideoSampleDurationTests.swift @@ -30,7 +30,8 @@ struct VideoSampleDurationTests { existingDuration: constantDefaultDuration, dts: dts[i], nextDts: dts[i + 1], - fallback: fallback + fallback: fallback, + capTicks: 10_000 // #369: 10 s in the ms grid; irrelevant to these deltas ) // Each sample duration is the true decode-order delta, not the constant 42. #expect(resolved == dts[i + 1] - dts[i]) @@ -45,27 +46,46 @@ struct VideoSampleDurationTests { func fallsBackWithoutForwardDelta() { // EOF tail (nextDts == nil): keep the source packet's own positive duration. #expect(HLSSegmentProducer.resolveVideoSampleDuration( - existingDuration: 40, dts: 1_000, nextDts: nil, fallback: 42) == 40) + existingDuration: 40, dts: 1_000, nextDts: nil, fallback: 42, capTicks: 10_000) == 40) // Source duration missing too: use the configured fallback. #expect(HLSSegmentProducer.resolveVideoSampleDuration( - existingDuration: 0, dts: 1_000, nextDts: nil, fallback: 42) == 42) + existingDuration: 0, dts: 1_000, nextDts: nil, fallback: 42, capTicks: 10_000) == 42) } @Test("A non-forward next DTS never yields a zero or negative sample duration") func nonForwardNextDtsStaysPositive() { // Equal DTS (delta 0): fall back to the positive source duration. #expect(HLSSegmentProducer.resolveVideoSampleDuration( - existingDuration: 42, dts: 1_000, nextDts: 1_000, fallback: 33) == 42) + existingDuration: 42, dts: 1_000, nextDts: 1_000, fallback: 33, capTicks: 10_000) == 42) // Backward DTS with no usable source duration: fall back. #expect(HLSSegmentProducer.resolveVideoSampleDuration( - existingDuration: 0, dts: 1_000, nextDts: 990, fallback: 33) == 33) + existingDuration: 0, dts: 1_000, nextDts: 990, fallback: 33, capTicks: 10_000) == 33) + } + + @Test("A wrap-scale inferred delta is rejected by the discontinuity cap (#369)") + func wrapScaleDeltaIsCapped() { + // Device trace: seam-preceding dts 363524400, seam packet wrap-corrected to exactly 2^33. + // The look-behind delta IS the wrap (8226410192 ticks ≈ 91404 s @ 90 kHz); movenc rejected + // it ("Application provided duration ... is invalid") and the packet was silently lost. + // Cap = 10 s of 90 kHz ticks, the shared discontinuity threshold. + #expect(HLSSegmentProducer.resolveVideoSampleDuration( + existingDuration: 1800, dts: 363_524_400, nextDts: 8_589_934_592, + fallback: 3600, capTicks: 900_000) == 1800) + // No usable source duration either: the configured fallback. + #expect(HLSSegmentProducer.resolveVideoSampleDuration( + existingDuration: 0, dts: 363_524_400, nextDts: 8_589_934_592, + fallback: 3600, capTicks: 900_000) == 3600) + // A delta exactly at the cap still telescopes. + #expect(HLSSegmentProducer.resolveVideoSampleDuration( + existingDuration: 1800, dts: 0, nextDts: 900_000, + fallback: 3600, capTicks: 900_000) == 900_000) } @Test("A NOPTS dts or next-dts cannot produce an overflowing delta") func nopts() { #expect(HLSSegmentProducer.resolveVideoSampleDuration( - existingDuration: 42, dts: Int64.min, nextDts: 1_000, fallback: 33) == 42) + existingDuration: 42, dts: Int64.min, nextDts: 1_000, fallback: 33, capTicks: 10_000) == 42) #expect(HLSSegmentProducer.resolveVideoSampleDuration( - existingDuration: 0, dts: 1_000, nextDts: Int64.min, fallback: 33) == 33) + existingDuration: 0, dts: 1_000, nextDts: Int64.min, fallback: 33, capTicks: 10_000) == 33) } }