diff --git a/CHANGELOG.md b/CHANGELOG.md index 8f7a9fb8f..3c0eb0889 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -168,6 +168,13 @@ the public-API contract. the bound port while `stop()` reset it under the lock. Host-only redirects now take the same lock, and the port stays private to the server. Explicit `--redirect-port` targets are unchanged. +- **A software playback clock stays stopped when first-frame arming overlaps a pause.** + The feeder's post-arm rate correction could restart the synchronizer after `pause()` stopped it + but before the host published its paused intent, leaving the clock running behind parked decode + loops and causing video to catch up rapidly on resume. The correction now runs on the main actor + with transport changes, ignores retired outputs and seek generations, and yields to newer clock + commands, including rebuffer pauses. Pending pauses and end-of-media tail playback are preserved. + - **Source failures keep their cause and request ownership.** Initial live probes preserve HTTP, URL and TLS failures without changing the live-start `liveSourceUnavailable` kind (HTTP and certificate classifications stay specific). A transport failure at open (a reset or a timeout from diff --git a/Sources/AetherEngine/Audio/AudioOutput.swift b/Sources/AetherEngine/Audio/AudioOutput.swift index cb88d9182..31c70b28d 100644 --- a/Sources/AetherEngine/Audio/AudioOutput.swift +++ b/Sources/AetherEngine/Audio/AudioOutput.swift @@ -19,6 +19,16 @@ final class AudioOutput: @unchecked Sendable { @MainActor private var attachedVideoLayer: AVSampleBufferDisplayLayer? @MainActor private var videoLayerRemovalTask: Task? + /// External clock commands retire queued arm corrections. Compared under the same lock as + /// the rate change so a later feeder rebuffer pause cannot be overwritten by a stale correction. + /// Preroll completion belongs to its initiating command and does not advance this revision. + private var _clockCommandRevision: UInt64 = 0 + var clockCommandRevision: UInt64 { + lock.lock() + defer { lock.unlock() } + return _clockCommandRevision + } + /// AE#464: the host's audio presentation offset, applied to every buffer on its way into the /// renderer. Guarded because the host writes it from the main actor while the demux thread reads /// it in `enqueue`. @@ -197,41 +207,57 @@ final class AudioOutput: @unchecked Sendable { } /// Set playback speed (0.5-2.0). Hosts own rate state (lastRate/pausedByHost); this object is stateless about it. - func setRate(_ rate: Float) { + /// A queued correction supplies `ifRevision`; normal transport commands omit it. + func setRate(_ rate: Float, ifRevision revision: UInt64? = nil) { // A host resumes right after it arms the clock; while a preroll waits, that rate is the one // the preroll starts at, not a start of its own. - if rate > 0, updatePrerollRate(rate) { return } + if rate > 0, updatePrerollRate(rate, ifRevision: revision) { return } // A clock that stands still starts through the preroll on a session that has one: a seek // landed while paused anchors at rate 0 and never asks the gate, so the play that follows // would otherwise start it with nothing queued. if rate > 0, synchronizer.rate == 0, videoPrerollSatisfied?(currentTime.seconds) != nil { - seekClock(to: currentTime, rate: rate) + seekClock(to: currentTime, rate: rate, ifRevision: revision) return } - cancelPreroll() + lock.lock() + if let revision, revision != _clockCommandRevision { + lock.unlock() + return + } + _clockCommandRevision &+= 1 + endPreroll() let at = synchronizer.currentTime() - EngineLog.emit("[AudioOutput] setRate \(rate) at t=\(String(format: "%.3f", timeline.sourceSeconds(forRenderer: at.seconds)))", category: .swPlayback) + let sourceSeconds = timeline.sourceSeconds(forRenderer: at.seconds) synchronizer.setRate(rate, time: at) + lock.unlock() + EngineLog.emit("[AudioOutput] setRate \(rate) at t=\(String(format: "%.3f", sourceSeconds))", category: .swPlayback) } /// Pause audio (and the master clock). Hosts resume via setRate (pausedByHost pattern); deliberately no resume() here. func pause() { - cancelPreroll() + lock.lock() + _clockCommandRevision &+= 1 + endPreroll() let at = synchronizer.currentTime() - EngineLog.emit("[AudioOutput] pause at t=\(String(format: "%.3f", timeline.sourceSeconds(forRenderer: at.seconds)))", category: .swPlayback) + let sourceSeconds = timeline.sourceSeconds(forRenderer: at.seconds) synchronizer.setRate(0.0, time: at) + lock.unlock() + EngineLog.emit("[AudioOutput] pause at t=\(String(format: "%.3f", sourceSeconds))", category: .swPlayback) } /// Pause the master clock, at `latest` if it has already run past it. A park deferred to the last /// sample runs whenever its task is scheduled, and on a starved main actor that is after the clock /// has walked on (#694: 1.149 s on a 1.0 s source on a CI runner). func pause(notAfter latest: Double) { - cancelPreroll() + lock.lock() + _clockCommandRevision &+= 1 + endPreroll() let now = timeline.sourceTime(forRenderer: synchronizer.currentTime()) let seconds = SoftwareEndOfMediaClock.parkSeconds(clockSeconds: CMTimeGetSeconds(now), notAfter: latest) let at = seconds.map { CMTime(seconds: $0, preferredTimescale: 90000) } ?? now - EngineLog.emit("[AudioOutput] pause at t=\(String(format: "%.3f", at.seconds))", category: .swPlayback) synchronizer.setRate(0.0, time: timeline.rendererTime(forSource: at)) + lock.unlock() + EngineLog.emit("[AudioOutput] pause at t=\(String(format: "%.3f", at.seconds))", category: .swPlayback) } /// AE#464: set the audio presentation offset. Positive presents audio later than video, which on @@ -318,6 +344,23 @@ final class AudioOutput: @unchecked Sendable { /// Test-only: runs inside `enqueue`'s locked section, right after the renderer took the buffer, which /// is where a renderer that posts its notifications from inside `enqueue` would post them. var afterRendererEnqueueForTesting: (@Sendable () -> Void)? + + /// Holds a pending start without its timeout, so transport tests exercise cancellation + /// independently of how long the main actor takes to run on a loaded test runner. + func holdPrerollForTesting(rate: Float) { + lock.lock() + defer { lock.unlock() } + _clockCommandRevision &+= 1 + endPreroll() + prerollRate = rate + } + + var prerollRateForTesting: Float? { + lock.lock() + defer { lock.unlock() } + return prerollRate + } + private var _loggedFirstEnqueue = false #endif private var _loggedRendererError = false @@ -400,6 +443,7 @@ final class AudioOutput: @unchecked Sendable { func flush() { lock.lock() defer { lock.unlock() } + _clockCommandRevision &+= 1 _epoch &+= 1 renderer.flush() } @@ -407,6 +451,7 @@ final class AudioOutput: @unchecked Sendable { func stop() { lock.lock() defer { lock.unlock() } + _clockCommandRevision &+= 1 _epoch &+= 1 endPreroll() synchronizer.setRate(0.0, time: .zero) @@ -416,12 +461,14 @@ final class AudioOutput: @unchecked Sendable { /// Atomically jump the master clock to a time and resume at a rate. The ONLY way the clock is (re)anchored: /// demux loops call it once on the first decoded packet, seek paths call it directly. Avoids the /// falling-through-time races that pause -> flush -> setRate would expose. - func seekClock(to time: CMTime, rate: Float) { + func seekClock(to time: CMTime, rate: Float, ifRevision revision: UInt64? = nil) { // Asked before taking `lock`: the gate reads the video renderer's state under its own lock. let preroll = rate > 0 ? videoPrerollSatisfied?(time.seconds) : nil let progress = preroll == false ? videoPrerollProgress?() : nil lock.lock() defer { lock.unlock() } + if let revision, revision != _clockCommandRevision { return } + _clockCommandRevision &+= 1 EngineLog.emit("[AudioOutput] seekClock to=\(String(format: "%.3f", time.seconds)) rate=\(rate)", category: .swPlayback) endPreroll() switch preroll { @@ -569,12 +616,6 @@ final class AudioOutput: @unchecked Sendable { synchronizer.setRate(rate, time: timeline.rendererTime(forSource: time), atHostTime: host) } - private func cancelPreroll() { - lock.lock() - endPreroll() - lock.unlock() - } - /// Retires the current preroll chain, and reports a hold that was still waiting as cancelled. /// Called with `lock` held. private func endPreroll() { @@ -585,10 +626,12 @@ final class AudioOutput: @unchecked Sendable { prerollRate = nil } - private func updatePrerollRate(_ rate: Float) -> Bool { + private func updatePrerollRate(_ rate: Float, ifRevision revision: UInt64?) -> Bool { lock.lock() defer { lock.unlock() } + if let revision, revision != _clockCommandRevision { return false } guard prerollRate != nil else { return false } + _clockCommandRevision &+= 1 prerollRate = rate return true } diff --git a/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift b/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift index 0cfbfd7e7..42dc0961e 100644 --- a/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift +++ b/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift @@ -1681,8 +1681,46 @@ final class SoftwarePlaybackHost { } var clockRateForTesting: Float? { audioOutput?.rate } + + func installClockOutputForTesting(_ output: AudioOutput, armed: Bool = false) { + audioOutput = output + clockArmed = armed + } + + func finishClockOutputForTesting() { + didReachEnd = true + isPlaying = false + } #endif + /// The latch is synchronous for the feeder; transport reconciliation shares play/pause's actor. + func makeClockArmCallback(audioOutput aOut: AudioOutput?) -> @Sendable () -> Void { + { [weak self] in + guard let self, let aOut else { return } + let generation = self.seekGeneration + let revision = aOut.clockCommandRevision + self.clockArmed = true + // A transport change before arming still needs correction, but the feeder must not + // restart the clock between pause() stopping it and publishing isPlaying=false. + DispatchQueue.main.async { [weak self] in + guard let self, self.audioOutput === aOut, + self.admitsRead(generation: generation) else { return } + // EOF may drain its queued audio tail, but an explicit pre-arm pause still wins. + guard self.pausedByHost || (!self.didParkClockAtEnd && !self.didReachEnd) else { return } + if !self.isPlaying || self.pausedBeforeFirstFrame { + // A pending preroll can own a future start even while the clock reads zero. + aOut.setRate(0, ifRevision: revision) + } else if let rate = PausedFirstFrame.rateCorrectionAtArm( + transportPlaying: true, pausedBeforeFirstFrame: false, + synchronizerRate: aOut.synchronizer.rate, lastRate: self.lastRate) { + // A later rebuffer pause, seek or rate command owns the output now. The + // comparison and correction are atomic with those commands inside AudioOutput. + aOut.setRate(rate, ifRevision: revision) + } + } + } + } + /// AE#549: re-anchor a master clock that stopped without a pause of ours, where it stands. /// /// The one re-anchor this path can make: the demuxer is an audio lead ahead of the clock by now @@ -2431,16 +2469,7 @@ final class SoftwarePlaybackHost { let getClockArmed: @Sendable () -> Bool = { [weak self] in self?.clockArmed ?? true } - let setClockArmed: @Sendable () -> Void = { [weak self] in - guard let self else { return } - self.clockArmed = true - // Sodalite#104 round 4: a pause() or play() between this arm reading its rate and the line - // above found no clock to act on. - guard let aOut, let rate = PausedFirstFrame.rateCorrectionAtArm( - transportPlaying: self.isPlaying, pausedBeforeFirstFrame: self.pausedBeforeFirstFrame, - synchronizerRate: aOut.synchronizer.rate, lastRate: self.lastRate) else { return } - if rate == 0 { aOut.pause() } else { aOut.setRate(rate) } - } + let setClockArmed = makeClockArmCallback(audioOutput: aOut) let getSeekGeneration: @Sendable () -> UInt64 = { [weak self] in self?.seekGeneration ?? 0 } diff --git a/Tests/AetherEngineTests/Sodalite104PausedFirstFrameTests.swift b/Tests/AetherEngineTests/Sodalite104PausedFirstFrameTests.swift index 89aa6d3a3..6139c263c 100644 --- a/Tests/AetherEngineTests/Sodalite104PausedFirstFrameTests.swift +++ b/Tests/AetherEngineTests/Sodalite104PausedFirstFrameTests.swift @@ -1,3 +1,6 @@ +import AVFoundation +import Combine +import Foundation import Testing @testable import AetherEngine @@ -54,4 +57,242 @@ struct Sodalite104PausedFirstFrameTests { #expect(PausedFirstFrame.presentationAnchor(framePTS: 10, clockArmed: true, clockSeconds: .nan) == nil) #expect(PausedFirstFrame.presentationAnchor(framePTS: .nan, clockArmed: true, clockSeconds: 0) == nil) } + + @MainActor + @Test("clock-arm reconciliation cannot restart the clock halfway through pause()", .timeLimit(.minutes(1))) + func clockArmDuringPause() async throws { + let host = SoftwarePlaybackHost() + let output = AudioOutput() + defer { output.stop() } + host.installClockOutputForTesting(output, armed: true) + host.isPlaying = true + output.seekClock(to: .zero, rate: 1) + let arm = host.makeClockArmCallback(audioOutput: output) + let revisionBeforePause = output.clockCommandRevision + + var correctionEntered = false + // @Published emits synchronously: pause() has issued its stop command here, but has not yet + // set isPlaying=false. Re-entering the real arm callback pins the feeder's bad ordering + // without relying on two threads happening to hit that window on a loaded runner. + let observation = host.$rate.dropFirst().sink { rate in + guard rate == 0 else { return } + #expect(host.isPlaying) + let pauseRevision = output.clockCommandRevision + #expect(pauseRevision == revisionBeforePause &+ 1) + correctionEntered = true + arm() + #expect(output.clockCommandRevision == pauseRevision, + "arming must not issue a transport command while pause intent is being published") + } + defer { observation.cancel() } + host.pause() + + #expect(correctionEntered) + #expect(!host.isPlaying) + #expect(output.rate == 0, "the feeder must not restart a clock pause() just stopped") + try await drainClockArmCallbacks() + #expect(output.rate == 0, "the queued correction sees the completed pause") + } + + @MainActor + @Test("an arm reconciles a pause that preceded the armed latch", .timeLimit(.minutes(1))) + func pauseBeforeClockArm() async throws { + let host = SoftwarePlaybackHost() + let output = AudioOutput() + defer { output.stop() } + host.installClockOutputForTesting(output) + host.isPlaying = true + host.pause() + // The feeder had already read the old playback intent before pause() latched it. + output.seekClock(to: .zero, rate: 1) + host.makeClockArmCallback(audioOutput: output)() + try await drainClockArmCallbacks() + #expect(output.rate == 0) + #expect(!host.isPlaying) + } + + @MainActor + @Test("a queued arm uses the current playback intent and speed", .timeLimit(.minutes(1))) + func clockArmUsesLatestPlaybackIntent() async throws { + let host = SoftwarePlaybackHost() + let output = AudioOutput() + defer { output.stop() } + host.installClockOutputForTesting(output) + host.setRate(1.5) + host.makeClockArmCallback(audioOutput: output)() + host.isPlaying = true + try await drainClockArmCallbacks() + #expect(output.rate == 1.5, "a play before reconciliation must not be lost") + } + + @MainActor + @Test("a queued arm cannot restart an output after stop was requested", .timeLimit(.minutes(1))) + func clockArmAfterStopRequest() async throws { + let host = SoftwarePlaybackHost() + let output = AudioOutput() + defer { output.stop() } + host.installClockOutputForTesting(output) + host.isPlaying = true + host.makeClockArmCallback(audioOutput: output)() + host.stopRequested = true + try await drainClockArmCallbacks() + #expect(output.rate == 0) + } + + @MainActor + @Test("a queued arm cannot change an output replaced by another session", .timeLimit(.minutes(1))) + func clockArmAfterOutputReplacement() async throws { + let host = SoftwarePlaybackHost() + let oldOutput = AudioOutput() + let newOutput = AudioOutput() + defer { oldOutput.stop(); newOutput.stop() } + host.installClockOutputForTesting(oldOutput) + host.isPlaying = true + host.makeClockArmCallback(audioOutput: oldOutput)() + host.installClockOutputForTesting(newOutput) + try await drainClockArmCallbacks() + #expect(oldOutput.rate == 0) + #expect(newOutput.rate == 0) + } + + @MainActor + @Test("an arm under pause cancels a pending preroll start even while the clock reads zero", + .timeLimit(.minutes(1))) + func pausedClockArmCancelsPreroll() async throws { + let host = SoftwarePlaybackHost() + let output = AudioOutput() + defer { output.stop() } + host.installClockOutputForTesting(output) + output.holdPrerollForTesting(rate: 1) + #expect(output.rate == 0) + #expect(output.prerollPending) + host.makeClockArmCallback(audioOutput: output)() + try await drainClockArmCallbacks() + #expect(!output.prerollPending, "rate zero must not hide a pending positive start") + #expect(output.rate == 0) + } + + @MainActor + @Test("a queued arm leaves the running audio tail to the end-of-media park", .timeLimit(.minutes(1))) + func clockArmAfterEndOfMedia() async throws { + let host = SoftwarePlaybackHost() + let output = AudioOutput() + defer { output.stop() } + host.installClockOutputForTesting(output) + host.isPlaying = true + output.seekClock(to: .zero, rate: 1) + host.makeClockArmCallback(audioOutput: output)() + // EOF clears playback intent before the queued audio tail's deferred clock park. + host.finishClockOutputForTesting() + try await drainClockArmCallbacks() + #expect(output.rate == 1, "reconciliation must not truncate the remaining audio tail") + } + + @MainActor + @Test("EOF cannot discard an explicit pause that preceded the armed latch", .timeLimit(.minutes(1))) + func pausedClockArmAfterEndOfMedia() async throws { + let host = SoftwarePlaybackHost() + let output = AudioOutput() + defer { output.stop() } + host.installClockOutputForTesting(output) + host.isPlaying = true + host.pause() + // The feeder had read the previous intent before pause() found an unarmed clock. + output.seekClock(to: .zero, rate: 1) + host.makeClockArmCallback(audioOutput: output)() + host.finishClockOutputForTesting() + try await drainClockArmCallbacks() + #expect(output.rate == 0, "EOF must preserve the user's explicit pause") + } + + @MainActor + @Test("a queued arm cannot undo a feeder's rebuffer pause", .timeLimit(.minutes(1))) + func clockArmAfterFeederPause() async throws { + let host = SoftwarePlaybackHost() + let output = AudioOutput() + defer { output.stop() } + host.installClockOutputForTesting(output) + host.isPlaying = true + output.seekClock(to: .zero, rate: 1) + host.makeClockArmCallback(audioOutput: output)() + // The DVR feeder owns this hold locally; the host still reports playing and its + // combined-loop rebuffer diagnostics do not change. + output.pause() + try await drainClockArmCallbacks() + #expect(output.rate == 0) + } + + @MainActor + @Test("a queued arm cannot overwrite a newer seek", .timeLimit(.minutes(1))) + func clockArmAfterNewerSeek() async throws { + let host = SoftwarePlaybackHost() + let output = AudioOutput() + defer { output.stop() } + host.installClockOutputForTesting(output) + host.isPlaying = true + output.seekClock(to: .zero, rate: 1) + host.makeClockArmCallback(audioOutput: output)() + output.seekClock(to: CMTime(seconds: 42, preferredTimescale: 90000), rate: 0) + try await drainClockArmCallbacks() + #expect(output.rate == 0) + #expect(output.currentTimeSeconds == 42) + } + + @MainActor + @Test("a queued arm cannot overwrite a newer speed", .timeLimit(.minutes(1))) + func clockArmAfterNewerRate() async throws { + let host = SoftwarePlaybackHost() + let output = AudioOutput() + defer { output.stop() } + host.installClockOutputForTesting(output) + host.isPlaying = true + host.makeClockArmCallback(audioOutput: output)() + output.setRate(1.5) + try await drainClockArmCallbacks() + #expect(output.rate == 1.5) + } + + @MainActor + @Test("a queued arm cannot overwrite a newer pending preroll speed", .timeLimit(.minutes(1))) + func clockArmAfterNewerPrerollRate() async throws { + let host = SoftwarePlaybackHost() + let output = AudioOutput() + defer { output.stop() } + host.installClockOutputForTesting(output) + host.isPlaying = true + output.holdPrerollForTesting(rate: 1) + host.makeClockArmCallback(audioOutput: output)() + output.setRate(1.5) + try await drainClockArmCallbacks() + #expect(output.prerollRateForTesting == 1.5) + #expect(output.rate == 0) + } + + @Test("a correction rechecks its revision after the preroll preflight", + .timeLimit(.minutes(1)), arguments: [false, true]) + func clockCorrectionAfterPreflightPause(usesPreroll: Bool) { + let output = AudioOutput() + defer { output.videoPrerollSatisfied = nil; output.stop() } + let revision = output.clockCommandRevision + // This callback runs outside the clock lock. A pause can win while readiness is + // being checked; both the direct-rate and seek/preroll branches must reject the + // earlier correction when they acquire the lock to apply it. + output.videoPrerollSatisfied = { _ in + output.pause() + return usesPreroll ? false : nil + } + output.setRate(1, ifRevision: revision) + #expect(output.clockCommandRevision != revision) + #expect(output.rate == 0) + #expect(!output.prerollPending) + } + + @MainActor + private func drainClockArmCallbacks() async throws { + // The marker follows every correction queued by this test on the same serial queue. + // waitFor keeps the wait cancellable and leaves the main actor free to execute them. + var drained = false + DispatchQueue.main.async { drained = true } + try await waitFor { drained } + } } diff --git a/docs/architecture.md b/docs/architecture.md index 994601894..f3d163664 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -114,6 +114,16 @@ Source URL ──► Demuxer ──┬─► SoftwareVideoDecoder (dav1d) ── AVR / speakers ``` +Clock arming publishes its latch synchronously on the feeder, so `play()` and `pause()` can see +that a clock exists. Its post-arm rate correction runs on the main actor alongside those transport +operations: it cannot restart the synchronizer halfway through a pause. Before applying a correction, +it rechecks the output identity and seek generation, ignores a stopped or seeking session, and leaves +end-of-media tail playback to its owner unless the host explicitly paused. It also carries the output's clock-command revision: +`AudioOutput` compares that revision and applies the rate under its existing lock, so a later +rebuffer pause, seek or rate command supersedes the queued correction. Preroll completion belongs +to its original command and keeps the revision. Paused intent cancels a pending preroll start, +even while the synchronizer still reports rate zero. + A seek holds the last frame on screen rather than blanking it. `SampleBufferRenderer.flush()` takes a `removingDisplayedImage` flag and hands it straight to `AVSampleBufferVideoRenderer.flush(removingDisplayedImage:)`: stop/teardown clears the visible frame (the default), but `SoftwarePlaybackHost.seek()` passes `false`, so the previous frame stays up until the post-seek keyframe decodes instead of flashing black on slow sources like MPEG-2. This matches the native/AVPlayer path, which holds the frame through a seek (#90). **The channel layout on that same format description has to name the order the resampler wrote** (#401). `AudioDecoder` resamples into `makeResamplerOutputLayout`'s layout and stamps `audioChannelLayoutTag`'s tag, and if the two disagree the renderer places audio where the decoder never put it: measured per channel, the earlier pairing moved every channel of a 7.1 track and placed the LFE hard left at full gain, put a 4.0 centre hard left, and mixed a 2.1 LFE into both channels instead of dropping it. Only 5.0 and 5.1 lined up, which is why it went unseen. The two functions therefore live side by side and a test holds them against each other, once structurally and once by rendering each channel through a real downmix. 6.1 is the single count CoreAudio has no matching tag for, so the resampler is pointed at 6.1(back) rather than leaving the well-trodden tags for a `UseChannelDescriptions` layout.