Zero-delay stream switching: prepare(next:) and seamless load

Sources/LumeEngine/LumePlayer.swift

@@ -49,6 +49,17 @@ public final class LumePlayer {
     private var tickTask: Task<Void, Never>?
     private let configuration: PlayerConfiguration
 
+    /// Standby session pre-opened by `prepare(next:)`, held through
+    /// first-frame-decoded until `load(url:)` consumes it.
+    private var preparedNext: (url: String, session: PlayerSession, info: MediaInfo)?
+    /// Supersession guards: a `load`/`prepare` that lost a race against a
+    /// newer call (or `stop`) discards its session instead of installing it.
+    private var loadGeneration: UInt64 = 0
+    private var prepareGeneration: UInt64 = 0
+    /// Bound on the first-frame wait during a seamless swap; past it the swap
+    /// proceeds anyway (losing only the no-gap guarantee).
+    private static let firstFrameTimeout: Double = 10
+
     public init(configuration: PlayerConfiguration = PlayerConfiguration()) {
         self.configuration = configuration
     }
@@ -57,37 +68,209 @@ public final class LumePlayer {
 
     /// Opens `url`, replacing any previous session (each open is a fresh
     /// engine session — PLAN.md §3.1).
+    ///
+    /// Zero-delay switching (PLAN.md §6), governed by
+    /// `PlayerConfiguration.switchPolicy`: when a session matching `url` was
+    /// staged by `prepare(next:)`, the swap is immediate under any policy.
+    /// Otherwise `.overlapped` keeps the current session rendering while the
+    /// replacement opens through its first decoded frame (two connections
+    /// briefly), and `.sequential` closes the current session *first* —
+    /// freeing its source connection — while its last decoded frame stays
+    /// frozen on screen until the replacement is ready. A failed overlapped
+    /// open (e.g. a provider capped at one concurrent connection) falls back
+    /// to the sequential path before the error surfaces.
     public func load(url: String) async throws -> MediaInfo {
+        loadGeneration &+= 1
+        let generation = loadGeneration
+
+        // A prepared session for this exact URL swaps in with no open cost
+        // (consuming it opens nothing new), so it is honored under any policy.
+        if let prepared = preparedNext, prepared.url == url {
+            preparedNext = nil
+            if await prepared.session.state != .failed {
+                guard generation == loadGeneration else {
+                    Task { await prepared.session.shutdown() }
+                    throw EngineError(code: .invalidState, message: "load superseded by a newer load")
+                }
+                adopt(session: prepared.session, info: prepared.info)
+                return prepared.info
+            }
+            // The standby died while waiting (network drop) — fall through.
+            let dead = prepared.session
+            Task { await dead.shutdown() }
+        }
+        discardPreparedSession()
+
+        let canSwitch = session != nil && state != .failed
+        if canSwitch, configuration.switchPolicy == .overlapped {
+            let next = PlayerSession(configuration: configuration)
+            do {
+                let info = try await next.open(url: url)
+                await next.waitForFirstFrame(timeout: Self.firstFrameTimeout)
+                guard generation == loadGeneration else {
+                    Task { await next.shutdown() }
+                    throw EngineError(code: .invalidState, message: "load superseded by a newer load")
+                }
+                adopt(session: next, info: info)
+                return info
+            } catch {
+                Task { await next.shutdown() }
+                guard generation == loadGeneration else { throw error }
+                // Fall through to the sequential retry: a provider capped at
+                // one concurrent connection refuses the overlapped open but
+                // accepts the same URL once the current stream is closed.
+            }
+        }
+        if canSwitch, configuration.switchPolicy != .none {
+            return try await sequentialLoad(url: url, generation: generation)
+        }
+
         await teardownSession()
+        guard generation == loadGeneration else {
+            throw EngineError(code: .invalidState, message: "load superseded by a newer load")
+        }
 
         let session = PlayerSession(configuration: configuration)
         self.session = session
         state = .opening
+        startObservation(of: session)
 
-        eventTask = Task { [events = session.events] in
-            for await event in events {
-                self.handle(event: event)
+        do {
+            let info = try await session.open(url: url)
+            guard generation == loadGeneration else {
+                throw EngineError(code: .invalidState, message: "load superseded by a newer load")
             }
+            mediaInfo = info
+            duration = info.duration.map(MediaTime.seconds)
+            return info
+        } catch {
+            // A superseded load must not clobber the newer load's state.
+            guard generation == loadGeneration else { throw error }
+            lastError = error as? EngineError
+            state = .failed
+            throw error
         }
-        tickTask = Task {
-            while !Task.isCancelled {
-                try? await Task.sleep(for: .milliseconds(100))
-                await self.tick()
-            }
+    }
+
+    /// Single-connection switch: the current session shuts down first (its
+    /// display layer keeps the last decoded frame — `SystemRenderer.shutdown`
+    /// flushes without removing the displayed image, so the screen never
+    /// blanks), then the replacement opens and the layers swap at its first
+    /// frame. At most one source connection exists at any moment. Playback
+    /// pauses for the duration of the open — the price of the connection cap.
+    private func sequentialLoad(url: String, generation: UInt64) async throws -> MediaInfo {
+        eventTask?.cancel()
+        tickTask?.cancel()
+        if let old = session {
+            // Bounded join: the connection must actually be closed before the
+            // replacement opens, or a one-connection provider refuses it.
+            await old.shutdown()
+        }
+        state = .opening
+        guard generation == loadGeneration else {
+            throw EngineError(code: .invalidState, message: "load superseded by a newer load")
         }
 
+        let next = PlayerSession(configuration: configuration)
         do {
-            let info = try await session.open(url: url)
-            mediaInfo = info
-            duration = info.duration.map(MediaTime.seconds)
+            let info = try await next.open(url: url)
+            await next.waitForFirstFrame(timeout: Self.firstFrameTimeout)
+            guard generation == loadGeneration else {
+                Task { await next.shutdown() }
+                throw EngineError(code: .invalidState, message: "load superseded by a newer load")
+            }
+            adopt(session: next, info: info)
             return info
         } catch {
+            Task { await next.shutdown() }
+            guard generation == loadGeneration else { throw error }
+            // The dead session stays attached: its frozen frame keeps the
+            // surface alive behind whatever failure UI the app raises.
             lastError = error as? EngineError
             state = .failed
             throw error
         }
     }
 
+    /// Stages `url` in a standby session, opened through first-frame-decoded
+    /// but never played (PLAN.md §6). A following `load(url:)` for the same
+    /// URL swaps it in with zero delay — this powers next-episode
+    /// auto-advance and channel zapping. Current playback is untouched.
+    ///
+    /// Only one URL is staged at a time; a newer `prepare` replaces the
+    /// previous standby. The standby holds its **own source connection** and
+    /// read-ahead buffer until consumed or discarded — only call this when
+    /// the source allows a stream beside the playing one (the same budget as
+    /// `SwitchPolicy.overlapped`); on one-connection providers rely on the
+    /// sequential switch instead.
+    @discardableResult
+    public func prepare(next url: String) async throws -> MediaInfo {
+        discardPreparedSession()
+        prepareGeneration &+= 1
+        let generation = prepareGeneration
+
+        let session = PlayerSession(configuration: configuration)
+        do {
+            let info = try await session.open(url: url)
+            await session.waitForFirstFrame(timeout: Self.firstFrameTimeout)
+            guard generation == prepareGeneration else {
+                Task { await session.shutdown() }
+                throw EngineError(code: .invalidState, message: "prepare superseded by a newer prepare")
+            }
+            preparedNext = (url, session, info)
+            return info
+        } catch {
+            Task { await session.shutdown() }
+            throw error
+        }
+    }
+
+    /// Discards the standby session staged by `prepare(next:)`, if any.
+    public func discardPreparedSession() {
+        prepareGeneration &+= 1
+        guard let prepared = preparedNext else { return }
+        preparedNext = nil
+        Task { await prepared.session.shutdown() }
+    }
+
+    /// Atomic swap of `prepare`d/seamlessly opened media: the new session
+    /// becomes the active one (its display layer replaces the old one via the
+    /// `displayLayer` observation) and the old session is torn down
+    /// asynchronously — the swap never waits for thread joins.
+    private func adopt(session next: PlayerSession, info: MediaInfo) {
+        eventTask?.cancel()
+        tickTask?.cancel()
+        if let old = session {
+            Task { await old.shutdown() }
+        }
+        session = next
+        state = .ready
+        mediaInfo = info
+        duration = info.duration.map(MediaTime.seconds)
+        lastError = nil
+        subtitleText = nil
+        startObservation(of: next)
+        if rate != 1.0 {
+            let rate = rate
+            Task { await next.setRate(rate) }
+        }
+    }
+
+    /// Event pump + 10 Hz position tick for the active session.
+    private func startObservation(of session: PlayerSession) {
+        eventTask = Task { [events = session.events] in
+            for await event in events {
+                self.handle(event: event)
+            }
+        }
+        tickTask = Task {
+            while !Task.isCancelled {
+                try? await Task.sleep(for: .milliseconds(100))
+                await self.tick()
+            }
+        }
+    }
+
     public func play() {
         let session = session
         Task { await session?.play() }
@@ -127,6 +310,8 @@ public final class LumePlayer {
     }
 
     public func stop() async {
+        loadGeneration &+= 1
+        discardPreparedSession()
         await teardownSession()
         state = .idle
         position = 0

Sources/LumeEngineCore/Session/PlayerSession.swift

@@ -46,6 +46,28 @@ public struct PlayerConfiguration: Sendable {
     /// Seconds without clock progress (while expected) before `.stalled` fires.
     public var stallThreshold: Double = 8
 
+    /// How `LumePlayer.load(url:)` replaces an already-open source
+    /// (zero-delay switching, PLAN.md §6).
+    public enum SwitchPolicy: Sendable, Equatable {
+        /// Zero-delay: the replacement opens behind the still-playing session
+        /// and swaps in once it holds its first decoded frame. Briefly holds
+        /// **two** source connections — use only when the source allows a
+        /// second concurrent stream (a refused overlapped open falls back to
+        /// `.sequential` automatically).
+        case overlapped
+        /// Single-connection zero-black: the current session closes *first*
+        /// (releasing its connection), its last decoded frame stays frozen on
+        /// screen while the replacement opens, and the layers swap at first
+        /// frame. Playback pauses during the switch but the screen never
+        /// blanks. The safe choice for providers capped at one connection.
+        case sequential
+        /// Teardown-first with a blank surface (no switching aid).
+        case none
+    }
+
+    /// Switch behavior for `LumePlayer.load(url:)`; see `SwitchPolicy`.
+    public var switchPolicy: SwitchPolicy = .overlapped
+
     public init() {}
 }
 
@@ -78,7 +100,7 @@ public actor PlayerSession {
     /// `activeCues(at: renderer.currentTime)` from any thread.
     public nonisolated let subtitles = SubtitleStore()
 
-    private(set) var state: State = .idle {
+    public private(set) var state: State = .idle {
         didSet {
             if state != oldValue { eventSink.yield(.stateChanged(state)) }
         }
@@ -372,6 +394,35 @@ public actor PlayerSession {
         )
     }
 
+    // MARK: Zero-delay switching support (PLAN.md §6)
+
+    /// Suspends until the renderer holds the first decoded frame of the
+    /// presentation lane — video when a video track is active, else audio.
+    /// The renderers accept media while the clock is paused, so a freshly
+    /// opened session reaches this point without ever playing: this is the
+    /// "prepared through first-frame-decoded" gate for zero-delay switching
+    /// (a standby session whose display layer already shows a picture can be
+    /// swapped in with no black gap).
+    ///
+    /// Returns `false` if the session fails, is torn down, or the timeout
+    /// elapses first — callers may still swap then; they just lose the
+    /// no-gap guarantee.
+    @discardableResult
+    public func waitForFirstFrame(timeout: Double = 10) async -> Bool {
+        let deadline = Date(timeIntervalSinceNow: timeout)
+        while Date() < deadline {
+            if state == .failed || state == .idle && info != nil { return false }
+            let lowWater = renderer.enqueuedLowWaterMark
+            if videoDecoder != nil {
+                if MediaTime.isValid(lowWater.video) { return true }
+            } else if audioDecoder != nil {
+                if MediaTime.isValid(lowWater.audio) { return true }
+            }
+            try? await Task.sleep(for: .milliseconds(50))
+        }
+        return false
+    }
+
     // MARK: Seek
 
     /// Seeks to `position` seconds (media-relative). Flush order matters: