fix(video): contain a timestamp leap that escapes the timeline rebase (#369)

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
 

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)..<newIdx
             cache.noteFolded(folded)
+            if folded.count > 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<AVPacket>,
         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),

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<Int>) {
         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
         }

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)
+    }
+}

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))
+    }
+}

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)
     }
 }