From c02ed6616ea13e3f103f3f971d6d8ab281ecc72c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Santos?= Date: Fri, 16 Oct 2026 00:51:43 +0200 Subject: [PATCH] feat(software): a live join starts near the newest media once the connect burst has landed 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 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. Measured on the same channel: first picture 1.1 s after the open, 4 to 6 s behind the newest media. SoftwareLiveJoinPolicy, SoftwareLiveJoinPlacementTests. --- CHANGELOG.md | 13 +++ .../Native/SoftwareLiveJoinPolicy.swift | 78 +++++++++++++++ .../Native/SoftwarePlaybackHost.swift | 59 ++++++++++++ .../DocumentedConstantsTests.swift | 9 ++ .../SoftwareLiveJoinPlacementTests.swift | 94 +++++++++++++++++++ docs/formats.md | 2 +- 6 files changed, 254 insertions(+), 1 deletion(-) create mode 100644 Sources/AetherEngine/Native/SoftwareLiveJoinPolicy.swift create mode 100644 Tests/AetherEngineTests/SoftwareLiveJoinPlacementTests.swift diff --git a/CHANGELOG.md b/CHANGELOG.md index 8bca37f58..6ac74bc5c 100644 --- a/CHANGELOG.md +++ b/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`. + ## [7.37.0] - 2026-10-15 ### Added diff --git a/Sources/AetherEngine/Native/SoftwareLiveJoinPolicy.swift b/Sources/AetherEngine/Native/SoftwareLiveJoinPolicy.swift new file mode 100644 index 000000000..fd9222e61 --- /dev/null +++ b/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) + } +} diff --git a/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift b/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift index 33b97a6ef..aeea19bd9 100644 --- a/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift +++ b/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 diff --git a/Tests/AetherEngineTests/DocumentedConstantsTests.swift b/Tests/AetherEngineTests/DocumentedConstantsTests.swift index c7688b15a..4305539a6 100644 --- a/Tests/AetherEngineTests/DocumentedConstantsTests.swift +++ b/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 { diff --git a/Tests/AetherEngineTests/SoftwareLiveJoinPlacementTests.swift b/Tests/AetherEngineTests/SoftwareLiveJoinPlacementTests.swift new file mode 100644 index 000000000..c163e7c98 --- /dev/null +++ b/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) + } +} diff --git a/docs/formats.md b/docs/formats.md index 142e9dcc3..c69172161 100644 --- a/docs/formats.md +++ b/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