feat(software): a live join starts near the newest media once the connect burst has landed

CHANGELOG.md

@@ -10,6 +10,19 @@ the public-API contract.
 
 ## [Unreleased]
 
+### Changed
+
+- **A live stream on the software path starts near the newest media.** A continuous live stream over
+  HTTP opens with a burst of what the origin holds of the channel (8 to 18 s on one IPTV panel, at
+  link speed), and the software host armed its clock on that burst's first packet. Nothing on the
+  software path catches up, so the picture then ran the whole burst behind the broadcast for as long
+  as the channel played: measured 17 s behind on a channel whose native-path session ran 6 s behind.
+  The live feeder now waits for the burst to land (the ring growing no faster than real time over a
+  quarter second, from 0.3 s on, or 1.5 s at most) and starts 4 s behind the newest media, on a
+  keyframe; a source that delivers in real time, or a ring too shallow to be worth a later keyframe,
+  still starts from its first packet. The media behind the start stays in the rewind ring.
+  `SoftwareLiveJoinPolicy`, `SoftwareLiveJoinPlacementTests`.
+
 ### Fixed
 
 - **Profile 7 FEL chroma residual is sampled where the base chroma sits.** With a quarter-resolution

Sources/AetherEngine/Native/SoftwareLiveJoinPolicy.swift

@@ -0,0 +1,78 @@
+import Foundation
+
+/// Where a live session on the software host starts playing, once what the source sends at connect
+/// has landed.
+///
+/// A continuous live stream over HTTP (an IPTV panel's MPEG-TS form of a channel) opens with a burst:
+/// the panel sends what it holds of the channel, 8 to 18 s of media measured on one panel, at link
+/// speed, and real time after that. The feeder used to arm the clock on the first packet of that
+/// burst, and since nothing on the software path ever catches up, the picture then ran the whole
+/// burst behind the broadcast for as long as the channel played: 17 s on a channel whose native-path
+/// session, which AVPlayer places a holdback behind the newest segment, ran 6 s behind. For live
+/// sport that is the difference other players are judged by.
+///
+/// So a live join on the software host waits for the burst to land, then starts `leadSeconds` behind
+/// the newest media, on the keyframe at or before that moment. The burst is landed when the ring has
+/// grown no faster than real time over `calmWindowSeconds` (a source that delivers in real time is
+/// calm from its first packets and starts after `minimumSettleSeconds`), or after
+/// `maximumSettleSeconds` on a link that is still catching up, where the start takes what has
+/// arrived. A ring holding less than `leadSeconds + minimumSkipSeconds` is played from its start:
+/// there is nothing worth skipping. The media behind the start stays in the rewind ring.
+///
+/// Pure, so the rule is testable without a session; `SoftwarePlaybackHost` feeds it the ring's
+/// first and newest PTS from the feeder thread until it decides.
+struct SoftwareLiveJoinPolicy: Equatable {
+    /// How far behind the newest media the picture starts. The audio pump keeps this much decoded
+    /// audio ahead of the clock (`AudioLookaheadPolicy.targetLeadSeconds`), and a panel that hands
+    /// over about a second of media at a time leaves the lead swinging within it.
+    static let leadSeconds: Double = 4
+    /// The burst is given at least this long to begin landing before the ring's growth is judged.
+    static let minimumSettleSeconds: TimeInterval = 0.3
+    /// After this long the start takes what has arrived, calm or not.
+    static let maximumSettleSeconds: TimeInterval = 1.5
+    /// Growth is judged over windows this long.
+    static let calmWindowSeconds: TimeInterval = 0.25
+    /// A ring that grew by more than this multiple of the wall clock over a window is still landing
+    /// the burst.
+    static let calmGrowthRatio: Double = 1.5
+    /// A start is moved only when it skips at least this much; less is not worth a later keyframe.
+    static let minimumSkipSeconds: Double = 2
+
+    enum Decision: Equatable {
+        /// Play from the ring's first packet.
+        case fromStart
+        /// Start on the keyframe at or before this source PTS.
+        case place(targetPts: Double)
+    }
+
+    private var firstObservedAt: TimeInterval?
+    private var windowStartPts: Double?
+    private var windowStartAt: TimeInterval?
+
+    /// Notes the ring's first and newest PTS at `now` (seconds on any monotonic clock); nil while the
+    /// burst is still landing, then the decision, once.
+    mutating func observe(firstPts: Double, newestPts: Double, at now: TimeInterval) -> Decision? {
+        guard let firstObservedAt, let windowStartPts, let windowStartAt else {
+            self.firstObservedAt = now
+            self.windowStartPts = newestPts
+            self.windowStartAt = now
+            return nil
+        }
+        let elapsed = now - firstObservedAt
+        var settled = elapsed >= Self.maximumSettleSeconds
+        if !settled, elapsed >= Self.minimumSettleSeconds, now - windowStartAt >= Self.calmWindowSeconds {
+            let grew = newestPts - windowStartPts
+            let wall = now - windowStartAt
+            if grew <= wall * Self.calmGrowthRatio {
+                settled = true
+            } else {
+                self.windowStartPts = newestPts
+                self.windowStartAt = now
+            }
+        }
+        guard settled else { return nil }
+        let span = newestPts - firstPts
+        guard span.isFinite, span >= Self.leadSeconds + Self.minimumSkipSeconds else { return .fromStart }
+        return .place(targetPts: newestPts - Self.leadSeconds)
+    }
+}

Sources/AetherEngine/Native/SoftwarePlaybackHost.swift

@@ -371,6 +371,9 @@ final class SoftwarePlaybackHost {
     nonisolated(unsafe) private var sessionStartPts: Double = .nan
 
     nonisolated(unsafe) private var newestSourcePts: Double = .nan
+    /// Where a live session's picture starts once the source's connect burst has landed; read and
+    /// advanced by the live feeder under `liveEdgeLock`, reset with the session.
+    nonisolated(unsafe) private var liveJoinPolicy = SoftwareLiveJoinPolicy()
 
     /// Guards `sessionStartPts` / `newestSourcePts` against the demux
     /// thread writing while the main-actor time tick reads them.
@@ -2057,6 +2060,7 @@ final class SoftwarePlaybackHost {
         retentionClaims.removeAll()
         liveEdgeLock.lock()
         sessionStartPts = .nan
+        liveJoinPolicy = SoftwareLiveJoinPolicy()
         newestSourcePts = .nan
         liveEdgeLock.unlock()
 
@@ -2219,6 +2223,20 @@ final class SoftwarePlaybackHost {
             let recoverCursor: @Sendable (Int) -> Void = { [weak self] old in
                 self?.recoverEvictedFeedCursor(from: old, ring: ring)
             }
+            // Where the live join starts, once the connect burst has landed (see the policy). Nil
+            // while it is still landing; the feeder asks again a moment later.
+            let liveJoinPlacement: @Sendable () -> SoftwareLiveJoinPolicy.Decision? = { [weak self] in
+                guard let self else { return .fromStart }
+                self.liveEdgeLock.lock()
+                defer { self.liveEdgeLock.unlock() }
+                guard self.sessionStartPts.isFinite, self.newestSourcePts.isFinite else { return nil }
+                return self.liveJoinPolicy.observe(firstPts: self.sessionStartPts,
+                                                   newestPts: self.newestSourcePts,
+                                                   at: ProcessInfo.processInfo.systemUptime)
+            }
+            let placeCursor: @Sendable (Int) -> Void = { [weak self] seq in
+                self?.setFeedCursor(seq)
+            }
             let setSourceEnded: @Sendable () -> Void = { [weak self] in
                 self?.sourceEnded = true
             }
@@ -2267,6 +2285,8 @@ final class SoftwarePlaybackHost {
                     readCursor: readCursor,
                     advanceCursor: advanceCursor,
                     recoverCursor: recoverCursor,
+                    liveJoinPlacement: liveJoinPlacement,
+                    placeCursor: placeCursor,
                     seekGeneration: getSeekGeneration,
                     currentRate: currentRate,
                     isPlaying: getFeederPlaying,
@@ -2541,6 +2561,8 @@ final class SoftwarePlaybackHost {
         readCursor: @Sendable () -> Int,
         advanceCursor: @Sendable (Int) -> Void,
         recoverCursor: @Sendable (Int) -> Void,
+        liveJoinPlacement: @Sendable () -> SoftwareLiveJoinPolicy.Decision? = { .fromStart },
+        placeCursor: @Sendable (Int) -> Void = { _ in },
         seekGeneration: @Sendable () -> UInt64,
         currentRate: @Sendable () -> Float,
         isPlaying: @Sendable () -> Bool,
@@ -2552,6 +2574,9 @@ final class SoftwarePlaybackHost {
         audioTapSink: @Sendable () -> ((@Sendable (CMSampleBuffer) -> Void)?),
         noteDecodeGeneration: @Sendable () -> Void
     ) {
+        // The join's start is placed once, before the first packet is fed; a seek before then is
+        // the viewer's own choice of position and retires it.
+        var liveJoinPending = true
         // Audio look-ahead pump (#107 audio chopping): feed audio packets ahead of the
         // combined cursor so the audio renderer holds AudioLookaheadPolicy.targetLeadSeconds
         // of decoded audio regardless of video decode pace. Without it audio lead is capped
@@ -2689,6 +2714,40 @@ final class SoftwarePlaybackHost {
                 preArmPacketsFed = 0
                 hadLead = false
                 rebuffering = false
+                liveJoinPending = false
+            }
+            if liveJoinPending {
+                switch liveJoinPlacement() {
+                case nil:
+                    if sourceEnded() {
+                        // Nothing more will land; play what there is from its start.
+                        liveJoinPending = false
+                    } else {
+                        condition.lock()
+                        autoreleasepool {
+                            _ = condition.wait(until: Date(timeIntervalSinceNow: 0.05))
+                        }
+                        condition.unlock()
+                        return true
+                    }
+                case .fromStart?:
+                    liveJoinPending = false
+                case .place(let targetPts)?:
+                    liveJoinPending = false
+                    let cursor = readCursor()
+                    if let seq = ring.seq(forKeyframeAtOrBefore: targetPts), seq > cursor,
+                       let keyframe = ring.packet(atSeq: seq) {
+                        let skipped = keyframe.pts - (ring.packet(atSeq: cursor)?.pts ?? keyframe.pts)
+                        EngineLog.emit(
+                            "[SWHost] live join placed \(String(format: "%.1f", SoftwareLiveJoinPolicy.leadSeconds))s "
+                            + "behind the newest media: skipped \(String(format: "%.1f", skipped))s of the "
+                            + "connect burst (cursor seq=\(cursor) -> \(seq), keyframe pts=\(String(format: "%.2f", keyframe.pts)))",
+                            category: .swPlayback
+                        )
+                        placeCursor(seq)
+                        return true
+                    }
+                }
             }
             let cursor = readCursor()
             let bounds = ring.seqBounds

Tests/AetherEngineTests/DocumentedConstantsTests.swift

@@ -107,6 +107,15 @@ final class DocumentedConstantsTests: XCTestCase {
         return text
     }
 
+    /// docs/formats.md quotes the lead a software-path live join keeps behind the newest media.
+    func testSoftwareLiveJoinLeadMatchesDocumentation() throws {
+        XCTAssertEqual(SoftwareLiveJoinPolicy.leadSeconds, 4)
+        XCTAssertEqual(SoftwareLiveJoinPolicy.minimumSettleSeconds, 0.3)
+        XCTAssertEqual(SoftwareLiveJoinPolicy.maximumSettleSeconds, 1.5)
+        XCTAssertEqual(SoftwareLiveJoinPolicy.calmWindowSeconds, 0.25)
+        assertDocumented("`SoftwareLiveJoinPolicy.leadSeconds` (4 s) behind the newest media", try documentation())
+    }
+
     // MARK: - Documentation corpus
 
     private static var repoRoot: URL {

Tests/AetherEngineTests/SoftwareLiveJoinPlacementTests.swift

@@ -0,0 +1,94 @@
+import Testing
+import Foundation
+@testable import AetherEngine
+
+/// A live session on the software host starts near the newest media once the source's connect burst
+/// has landed, instead of at the burst's first packet, where it would run the whole burst behind the
+/// broadcast for the rest of the channel.
+@Suite("Software live join placement")
+struct SoftwareLiveJoinPlacementTests {
+
+    /// A panel that sends 12 s of media in the first half second, then real time: the join waits
+    /// for the burst to calm and starts `leadSeconds` behind the newest media.
+    @Test("a connect burst is landed before the start is placed behind the newest media")
+    func burstIsLandedThenPlaced() throws {
+        var policy = SoftwareLiveJoinPolicy()
+        let first = 1000.0
+        var decision: SoftwareLiveJoinPolicy.Decision?
+        var decidedAt: TimeInterval?
+        var now: TimeInterval = 0
+        while now <= 2, decision == nil {
+            // 12 s of media over the first 0.5 s, then one second of media per second.
+            let newest = now < 0.5 ? first + 24 * now : first + 12 + (now - 0.5)
+            decision = policy.observe(firstPts: first, newestPts: newest, at: now)
+            if decision != nil { decidedAt = now }
+            now += 0.05
+        }
+        let at = try #require(decidedAt)
+        // Calm is judged from 0.3 s on, over a quarter-second window.
+        #expect(at >= 0.3 && at <= 1.0, "decided at \(at)s")
+        let newestThen = first + 12 + (at - 0.5)
+        #expect(decision == .place(targetPts: newestThen - SoftwareLiveJoinPolicy.leadSeconds))
+    }
+
+    @Test("a source that delivers in real time starts from its first packet, soon")
+    func realTimeSourceStartsFromTheFirstPacket() throws {
+        var policy = SoftwareLiveJoinPolicy()
+        let first = 50.0
+        var decision: SoftwareLiveJoinPolicy.Decision?
+        var decidedAt: TimeInterval?
+        var now: TimeInterval = 0
+        while now <= 2, decision == nil {
+            decision = policy.observe(firstPts: first, newestPts: first + now, at: now)
+            if decision != nil { decidedAt = now }
+            now += 0.05
+        }
+        #expect(decision == .fromStart)
+        let at = try #require(decidedAt)
+        #expect(at >= SoftwareLiveJoinPolicy.minimumSettleSeconds && at <= 0.7, "decided at \(at)s")
+    }
+
+    @Test("a burst still landing at the deadline is placed on what has arrived")
+    func slowLinkPlacesAtTheDeadline() throws {
+        var policy = SoftwareLiveJoinPolicy()
+        let first = 0.0
+        var decision: SoftwareLiveJoinPolicy.Decision?
+        var decidedAt: TimeInterval?
+        var now: TimeInterval = 0
+        while now <= 3, decision == nil {
+            // Six seconds of media per second for as long as it takes: never calm.
+            decision = policy.observe(firstPts: first, newestPts: first + 6 * now, at: now)
+            if decision != nil { decidedAt = now }
+            now += 0.05
+        }
+        let at = try #require(decidedAt)
+        #expect(abs(at - SoftwareLiveJoinPolicy.maximumSettleSeconds) < 0.06, "decided at \(at)s")
+        #expect(decision == .place(targetPts: 6 * at - SoftwareLiveJoinPolicy.leadSeconds))
+    }
+
+    @Test("a short burst is not worth a later start")
+    func shortBurstPlaysFromTheStart() {
+        var policy = SoftwareLiveJoinPolicy()
+        let first = 10.0
+        var decision: SoftwareLiveJoinPolicy.Decision?
+        var now: TimeInterval = 0
+        while now <= 2, decision == nil {
+            // Five seconds of media at once, then real time: a lead of four leaves one to skip.
+            decision = policy.observe(firstPts: first, newestPts: first + 5 + now, at: now)
+            now += 0.05
+        }
+        #expect(decision == .fromStart)
+    }
+
+    @Test("media whose timestamps run backwards is played from the start")
+    func backwardsTimestampsPlayFromTheStart() {
+        var policy = SoftwareLiveJoinPolicy()
+        var decision: SoftwareLiveJoinPolicy.Decision?
+        var now: TimeInterval = 0
+        while now <= 2, decision == nil {
+            decision = policy.observe(firstPts: 5000, newestPts: 20 + now, at: now)
+            now += 0.05
+        }
+        #expect(decision == .fromStart)
+    }
+}

docs/formats.md

@@ -607,7 +607,7 @@ Server-side ad insertion (SSAI) plays through the direct path instead of bouncin
 
 The live path's sliding-window eviction (which bounds resident memory) and DVR rewind are confirmed on Apple TV against a real broadcast feed: `behindLiveSeconds` holds at real-time pacing and the resident footprint stays bounded within the tvOS jetsam budget. The same behavior is exercised off-device through the `aetherctl live` / `hlsfixture` harnesses (sliding-window retention, real-time pacing, mid-stream reconnect, program-boundary discontinuities, DVR timeshift).
 
-Raw live MPEG-TS over HTTP (a tuner or tuner proxy serving the transport stream directly, no HLS) plays on the software path; the forward-only reader routes it there automatically. Load it with `isLive: true` plus a `dvrWindowSeconds` for live semantics and the fastest open (an explicitly live load skips the open-time size probes entirely); `SourceProbe.isLive` flags no-duration network streams so hosts can detect and reload. Mid-stream-joined sources, which deliver their first samples at arbitrary PTS hours past zero, anchor the clock at the first decoded sample on every session shape (live, live+DVR, and a plain VOD open of the same URL or of a mid-broadcast capture file), publishing session-relative positions while `sourceTime` stays on the source axis for subtitles (#107).
+Raw live MPEG-TS over HTTP (a tuner, a tuner proxy, or an IPTV panel's continuous form of a channel) is routed like any live source: the native loopback path unless the codec, declared interlace or a host override sends it to the software path. Such a stream opens with whatever the origin holds of the channel, sent at link speed (8 to 18 s on one IPTV panel), and a software session no longer starts on the first packet of that burst, where it would run the whole burst behind the broadcast for as long as it played: once the burst has landed the live feeder places the start `SoftwareLiveJoinPolicy.leadSeconds` (4 s) behind the newest media, on a keyframe, and the burst stays in the rewind ring (the native path's AVPlayer already sits a holdback behind the newest segment). Load it with `isLive: true` plus a `dvrWindowSeconds` for live semantics and the fastest open (an explicitly live load skips the open-time size probes entirely); `SourceProbe.isLive` flags no-duration network streams so hosts can detect and reload. Mid-stream-joined sources, which deliver their first samples at arbitrary PTS hours past zero, anchor the clock at the first decoded sample on every session shape (live, live+DVR, and a plain VOD open of the same URL or of a mid-broadcast capture file), publishing session-relative positions while `sourceTime` stays on the source axis for subtitles (#107).
 
 ## Known limitations