fix(audio): keep the Atmos confirmation pass off one-request origins

CHANGELOG.md

@@ -10,7 +10,16 @@ the public-API contract.
 
 ## [Unreleased]
 
-_Nothing yet._
+### Fixed
+
+- **`confirmAtmos` no longer starves the video pump on an origin limited to one request.** Each
+  E-AC-3 track's pass is a second handle on the source, and at `maxConcurrentSourceRequests = 1`
+  every read it made took the pump's only slot: an integrator measured stalls every ten seconds on a
+  4K remux with 17 E-AC-3 tracks behind an Xtream origin that answers a second connection with 509,
+  and the pass never finished. The pass is now declined there, and for `sequentialOrigin` and
+  `heldSourceConnection`, the same set the I-frame rendition declines; the budget is re-checked
+  before every track, so a limit learned from refusals mid-session stops it too. The
+  `[AtmosConfirmation]` line now also appears for a stopped pass, with how many tracks it reached.
 
 ## [7.40.1] - 2026-10-18
 

Sources/AetherEngine/Audio/AtmosConfirmationWorker.swift

@@ -48,9 +48,20 @@ extension AetherEngine {
     /// Skipped for live sources: a second connection to a live origin costs a tuner on some backends, the
     /// pass would start at the live edge rather than where playback is, and a live Atmos badge does not
     /// justify either. Skipped for forward-only custom readers, which cannot hand out a second cursor.
+    /// Skipped where the origin rations requests (`atmosConfirmationDeclineReason`).
     func startAtmosConfirmation() {
         cancelAtmosConfirmation()
         guard loadedOptions.confirmAtmos, !isLive, let url = loadedURL else { return }
+        if let reason = Self.atmosConfirmationDeclineReason(
+            originSerial: !isCustomSource && OriginRequestBudget.shared.requiresSerialRequests(url),
+            sequentialOrigin: loadedOptions.sequentialOrigin,
+            heldSourceConnection: loadedOptions.heldSourceConnection
+        ) {
+            EngineLog.emit(
+                "[AtmosConfirmation] skipped: \(reason); isAtmos stays the pre-decode guess",
+                category: .engine)
+            return
+        }
         // A session-preserving reload of the same item keeps what the previous pass already learned and
         // only chases what is still open; a new item starts from scratch.
         if confirmedAtmosSource != url {
@@ -66,14 +77,33 @@ extension AetherEngine {
         let formatHint = customFormatHint
         let probesize = loadedOptions.probesize
         let maxAnalyzeDuration = loadedOptions.maxAnalyzeDuration
+        let budgetURL = isCustomSource ? nil : url
         atmosConfirmationTask = BlockingWork.detached(priority: .utility) { [weak self] in
             await self?.runAtmosConfirmation(
-                url: url, formatHint: formatHint, headers: headers,
+                url: url, budgetURL: budgetURL, formatHint: formatHint, headers: headers,
                 callerProbesize: probesize, callerMaxAnalyzeDuration: maxAnalyzeDuration,
                 candidates: candidates)
         }
     }
 
+    /// Why the pass must not run on this origin, or nil when it may.
+    ///
+    /// Each track's pass is a second handle on the source (head, Matroska Cues at the tail, then paced
+    /// cluster reads), and on an origin limited to one request every one of those reads takes the slot
+    /// the video pump needs. Measured at an Xtream origin that refuses a second connection with 509: a
+    /// 4K remux with 17 E-AC-3 tracks stalled every ten seconds for as long as the pass ran, and the pass
+    /// never finished. `sequentialOrigin` and `heldSourceConnection` are declarations of the same kind of
+    /// origin, one that punishes a second request rather than merely slowing it down. The I-frame
+    /// rendition declines the same three for the same reason.
+    nonisolated static func atmosConfirmationDeclineReason(
+        originSerial: Bool, sequentialOrigin: Bool, heldSourceConnection: Bool
+    ) -> String? {
+        if originSerial { return "the origin is limited to one request at a time (#377)" }
+        if sequentialOrigin { return "sequentialOrigin" }
+        if heldSourceConnection { return "heldSourceConnection" }
+        return nil
+    }
+
     /// Whether the host's reader can hand out the independent cursor each pass needs. The probe builds a
     /// real reader, which the `IOReader` contract leaves to the engine to close (audit DEC-109); a host
     /// whose fork opens a connection would otherwise hold one per load for nothing.
@@ -95,14 +125,23 @@ extension AetherEngine {
     /// One bounded pass per candidate, sequentially. Each opens its own demuxer and closes it before the
     /// next starts, so at most one extra handle on the source exists at a time. Every failure is silent
     /// and terminal for that track: the badge simply does not appear.
+    ///
+    /// `budgetURL` is re-checked before every track: an uncapped origin that answers 429/503/509 is lowered
+    /// to one request by the budget mid-session, and the remaining passes would then compete with the pump.
     nonisolated private func runAtmosConfirmation(
-        url: URL, formatHint: String?, headers: [String: String],
+        url: URL, budgetURL: URL?, formatHint: String?, headers: [String: String],
         callerProbesize: Int64?, callerMaxAnalyzeDuration: Int64?,
         candidates: [Int32]
     ) async {
         var confirmed: [Int32] = []
+        var scanned = 0
+        var stopReason: String?
         for index in candidates {
-            guard !Task.isCancelled else { break }
+            if Task.isCancelled { stopReason = "cancelled"; break }
+            if let budgetURL, OriginRequestBudget.shared.requiresSerialRequests(budgetURL) {
+                stopReason = "the origin dropped to one request at a time"
+                break
+            }
             if await confirmAtmosForOneTrack(
                 url: url, formatHint: formatHint, headers: headers,
                 callerProbesize: callerProbesize, callerMaxAnalyzeDuration: callerMaxAnalyzeDuration,
@@ -115,15 +154,29 @@ extension AetherEngine {
                     self.applyConfirmedTrackFlags()
                 }
             }
+            scanned += 1
         }
-        guard !Task.isCancelled else { return }
+        if stopReason == nil, Task.isCancelled { stopReason = "cancelled" }
         EngineLog.emit(
-            "[AtmosConfirmation] scanned \(candidates.count) E-AC-3 track(s), confirmed JOC on "
-            + (confirmed.isEmpty ? "none" : confirmed.map(String.init).joined(separator: ", ")),
+            Self.atmosConfirmationSummary(
+                scanned: scanned, of: candidates.count, confirmed: confirmed, stopReason: stopReason),
             category: .engine
         )
     }
 
+    /// The pass's one log line. Emitted on every exit, a stopped pass included, so a report shows how far
+    /// it got instead of showing nothing at all.
+    nonisolated static func atmosConfirmationSummary(
+        scanned: Int, of total: Int, confirmed: [Int32], stopReason: String?
+    ) -> String {
+        let joc = confirmed.isEmpty ? "none" : confirmed.map(String.init).joined(separator: ", ")
+        guard let stopReason else {
+            return "[AtmosConfirmation] scanned \(total) E-AC-3 track(s), confirmed JOC on \(joc)"
+        }
+        return "[AtmosConfirmation] stopped (\(stopReason)) after \(scanned) of \(total) E-AC-3 track(s), "
+            + "confirmed JOC on \(joc)"
+    }
+
     /// Open, detect, close. Returns whether this track is authoritatively JOC.
     nonisolated private func confirmAtmosForOneTrack(
         url: URL, formatHint: String?, headers: [String: String],

Sources/AetherEngine/PlayerState.swift

@@ -743,7 +743,7 @@ public struct LoadOptions: Sendable, Equatable {
     /// this runs the same bounded decode pass as `AetherEngine.probeDetectingAtmos` (see `AtmosDetectionOptions`
     /// for the caps) on a second handle to the source, once per E-AC-3 track, and republishes `audioTracks` as
     /// tracks confirm. It starts only after the session is up and runs at utility priority, so it never delays
-    /// the first frame. Skipped for live sources and for forward-only custom readers, which cannot be re-read.
+    /// the first frame. Skipped for live sources, for forward-only custom readers, which cannot be re-read, and where a second request would cost the video path its link: an origin limited to one request (`maxConcurrentSourceRequests == 1`, or a limit the engine learned from 429/503/509 refusals, re-checked before every track), `sequentialOrigin` and `heldSourceConnection`. One `[AtmosConfirmation]` log line names the reason, or how far a stopped pass got.
     /// Default `false` (#214 follow-up).
     public var confirmAtmos: Bool = false
 

Tests/AetherEngineTests/AtmosConfirmationTests.swift

@@ -164,3 +164,38 @@ struct AtmosConfirmationTests {
             callerProbesize: nil, callerMaxAnalyzeDuration: nil) == false)
     }
 }
+
+/// The pass is a second handle on the source per E-AC-3 track. On an origin that rations requests every
+/// read it makes takes the video pump's slot, so it must not run there at all.
+@Suite("Atmos confirmation: origin gate and summary line")
+struct AtmosConfirmationOriginGateTests {
+
+    @Test("an unrestricted origin runs the pass")
+    func unrestrictedOriginRuns() {
+        #expect(AetherEngine.atmosConfirmationDeclineReason(
+            originSerial: false, sequentialOrigin: false, heldSourceConnection: false) == nil)
+    }
+
+    @Test("a one-request origin, a sequential origin and a held connection each decline the pass")
+    func rationedOriginsDecline() {
+        #expect(AetherEngine.atmosConfirmationDeclineReason(
+            originSerial: true, sequentialOrigin: false, heldSourceConnection: false) != nil)
+        #expect(AetherEngine.atmosConfirmationDeclineReason(
+            originSerial: false, sequentialOrigin: true, heldSourceConnection: false) != nil)
+        #expect(AetherEngine.atmosConfirmationDeclineReason(
+            originSerial: false, sequentialOrigin: false, heldSourceConnection: true) != nil)
+    }
+
+    @Test("a completed pass keeps the summary format reports already grep for")
+    func completedSummary() {
+        #expect(AetherEngine.atmosConfirmationSummary(scanned: 3, of: 3, confirmed: [2, 5], stopReason: nil)
+            == "[AtmosConfirmation] scanned 3 E-AC-3 track(s), confirmed JOC on 2, 5")
+    }
+
+    @Test("a stopped pass still logs how far it got and what it found")
+    func stoppedSummary() {
+        let line = AetherEngine.atmosConfirmationSummary(
+            scanned: 2, of: 17, confirmed: [], stopReason: "cancelled")
+        #expect(line == "[AtmosConfirmation] stopped (cancelled) after 2 of 17 E-AC-3 track(s), confirmed JOC on none")
+    }
+}

docs/architecture.md

@@ -406,7 +406,7 @@ Sources/AetherEngine/
 ├── SubtitleSessionCarryover.swift           Subtitle session state carried across an engine-initiated `reloadAtCurrentPosition` (AirPlay LAN swap, background-return reopen), which otherwise rebuilds the session from `LoadOptions` alone (#170)
 ├── BackgroundTeardownSelection.swift        The session selection a teardown hands to the reload that follows it, for the one case where teardown and reload are not the same call: the paused-background teardown (#127, #357)
 ├── Audio/
-│   ├── AtmosConfirmationWorker.swift        LoadOptions.confirmAtmos: runs the bounded JOC pass on a side demuxer after the session is up, one E-AC-3 track at a time, and republishes audioTracks via a source-keyed ledger (#214)
+│   ├── AtmosConfirmationWorker.swift        LoadOptions.confirmAtmos: runs the bounded JOC pass on a side demuxer after the session is up, one E-AC-3 track at a time, and republishes audioTracks via a source-keyed ledger (#214); declined on an origin limited to one request, sequentialOrigin and heldSourceConnection, re-checked per track
 │   ├── AtmosDetectionProbe.swift            Opt-in bounded E-AC-3 JOC decode pass behind probeDetectingAtmos: AtmosDetectionOptions caps, one decoder, no playback session (#214)
 │   ├── AudioAVPlayerHost.swift              Audio-only path: bare AVPlayer host for whitelisted codecs, owns the persistent per-player MPNowPlayingSession (tvOS / iOS); publishes the rebuffer axis (`waitingToPlayAtSpecifiedRate` / `AVPlayerItemPlaybackStalled` after first play) that the engine folds into `isBuffering`
 │   ├── AudioBridge.swift                    Native path: decode + re-encode per `AudioBridgeMode` (EAC3 5.1 default or lossless FLAC opt-in) for source codecs that can't stream-copy into fMP4