From d3c4527d657961d3e14aa746a905613361c9f3bf Mon Sep 17 00:00:00 2001 From: Andreas Busslinger Date: Fri, 7 Aug 2026 17:46:16 +0200 Subject: [PATCH 1/3] fix(io): a live window absorbs the join burst once instead of cycling end-drain-reconnect MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 16 MB high-water end (#310) had no live branch, and live connections are open-ended by design, so ending at high water was the only thing that ever terminated a healthy live connection. Each end drained ~8 MB to low water and re-requested "at the frontier" — a byte offset that means nothing to a live origin — so everything broadcast during the drain was lost and the demuxer rejoined on a corrupt TS packet. And it never happened once: IPTV panels serve their ring buffer as a join burst at line rate on every (re)connect, so the burst refilled the window immediately and each reconnect caused the next one, forever. A field trace against an Xtream panel cycled every ~9.5 MB with a mpegts "Packet corrupt" and an h264 decode error per cycle; the loopback repro accepts 17 MB of a 24 MB burst, parks at 16.9 MB and holds no connection. - Live readers run a 64 MB high water (matching streamHighWater, the bound already accepted for the other reader that cannot bound by range request): the join burst is absorbed once and steady state plateaus at burst size — arrival rate equals media rate once the burst is over — with the connection never voluntarily ended. - The end-and-refill mechanism is unchanged and survives as the memory backstop for a "live" source that sustainedly outruns realtime (a misdeclared VOD). No suspend, no parked task: the #310 contract (actively delivering flow or no flow) holds verbatim. - The high water is an init parameter for the same reason connStallTimeout is one: a process-wide test hook would leak into concurrently running suites. Test plan: swift test on macOS 26 (1564 tests green, including a new live-window suite against the loopback origin: a 24 MB serve-then-silent burst must be accepted on one open-ended connection with parked=false, and a shrunken backstop must still end, stay bounded, and refill at the frontier on drain). Both new cases verified failing on the unfixed base. Field shape from an Apple TV 4K (tvOS 26) playing 1080i H.264 live TS from an Xtream aggregator with max_connections=1 and 302->http redirects. Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 18 ++- .../AetherEngine+Diagnostics.swift | 8 +- Sources/AetherEngine/Demuxer/AVIOReader.swift | 32 ++++- .../LiveWindowBackpressureTests.swift | 113 ++++++++++++++++++ 4 files changed, 163 insertions(+), 8 deletions(-) create mode 100644 Tests/AetherEngineTests/LiveWindowBackpressureTests.swift diff --git a/CHANGELOG.md b/CHANGELOG.md index b4a2f5fd..d69f9a5b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,7 +10,23 @@ the public-API contract. ## [Unreleased] -_Nothing yet._ +### Fixed + +- **A live source is no longer stuttered on a steady cycle by the reader's own backpressure.** + The 16 MB high-water end (#310) had no live branch, and live connections are open-ended by + design, so ending at high water was the only thing that ever terminated a healthy live + connection. Each end drained ~8 MB to low water and re-requested "at the frontier" — a byte + offset that means nothing to a live origin — so everything broadcast during the drain was lost + and the demuxer rejoined on a corrupt TS packet. And it never happened once: IPTV panels serve + their ring buffer as a join burst at line rate on every (re)connect, so the burst refilled the + window immediately and each reconnect caused the next one, forever (a field trace against an + Xtream panel cycled every ~9.5 MB with a `Packet corrupt` and an h264 decode error per cycle; + the loopback repro accepts 17 MB of a 24 MB burst, parks at 16.9 MB and holds no connection). + Live readers now run a 64 MB high water (matching `streamHighWater`, the bound the engine + already accepts for the other reader that cannot bound by range request): the join burst is + absorbed once, steady state plateaus at burst size with the connection never voluntarily + ended, and the end-and-refill survives unchanged as the memory backstop for a "live" source + that sustainedly outruns realtime. ## [6.15.1] - 2026-08-08 diff --git a/Sources/AetherEngine/AetherEngine+Diagnostics.swift b/Sources/AetherEngine/AetherEngine+Diagnostics.swift index baeedd76..5778e5ad 100644 --- a/Sources/AetherEngine/AetherEngine+Diagnostics.swift +++ b/Sources/AetherEngine/AetherEngine+Diagnostics.swift @@ -127,8 +127,10 @@ extension AetherEngine { } // #220: the two readers of a subtitled VOD session, separately attributable. - // `ahead` far above winHighWater (16 MB) with `parked=0` is backpressure that - // never engaged; the pump's own window is the control. + // `ahead` far above winHighWater (16 MB VOD / 64 MB live) with `parked=0` is + // backpressure that never engaged; the pump's own window is the control. A live + // reader plateauing between the two marks is healthy: that is the join burst, + // absorbed once and held. // // Both paths, not just software. On a direct-play source the native path runs // the HLS loopback, so `HLSVideoEngine` demuxes from the origin through an @@ -268,7 +270,7 @@ extension AetherEngine { /// #220: one memprobe fragment per live `AVIOReader` window. `win` is the whole buffer, /// `ahead` the undrained forward extent that `appendPersistentData` gates the backpressure - /// end on. `ahead` far above winHighWater (16 MB) while `parked=0` means the backpressure + /// end on. `ahead` far above winHighWater (16 MB VOD / 64 MB live) while `parked=0` means the backpressure /// never engaged, which is a different defect from a transport overshoot past an end that /// fired (#310: the end replaced the suspend, so the overshoot is bounded by one /// delivery's in-flight amount rather than by whatever a suspended task lets through). diff --git a/Sources/AetherEngine/Demuxer/AVIOReader.swift b/Sources/AetherEngine/Demuxer/AVIOReader.swift index 0883bc52..0aaff20d 100644 --- a/Sources/AetherEngine/Demuxer/AVIOReader.swift +++ b/Sources/AetherEngine/Demuxer/AVIOReader.swift @@ -210,7 +210,23 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { // window peak is now bounded by construction at highWater plus one delivery's in-flight // overshoot, which subsumes the former winHardCap escape hatch and the realloc-doubling // peak it had to be sized against. - private static let winHighWater = 16 * 1024 * 1024 + // + // Live raises the high water instead of changing the mechanism. Live connections are + // open-ended by design (no ranges to bound them), so the high-water end is the ONLY + // thing that ever terminates a healthy live connection — and "re-request at the + // frontier" is a lie to a live origin: the bytes broadcast during the drain are gone, + // so every cycle rejoined the stream on a corrupt TS packet. Worse, IPTV panels serve + // their ring buffer as a join burst at line rate on EVERY (re)connect, so a 16 MB cap + // made each reconnect the cause of the next one: burst to high water, end, drain ~8 MB + // losing that much realtime, reconnect, absorb the next burst. The live threshold is + // sized to absorb the burst ONCE; steady state then plateaus at burst size (arrival + // rate == media rate once the burst is over) with the connection never voluntarily + // ended. The end-and-refill stays, unchanged, as the memory backstop for a "live" + // source that sustainedly outruns realtime (a misdeclared VOD). 64 MB matches + // streamHighWater, the forward bound the engine already accepts for the other reader + // that cannot bound by range request. + private static let winHighWaterDefault = 16 * 1024 * 1024 + private static let liveWinHighWaterDefault = 64 * 1024 * 1024 private static let winLowWater = 8 * 1024 * 1024 // #220: how much the persistent reader asks for at a time. Bounds a single request's // exposure by construction (an origin cannot serve more than it was asked for, whatever @@ -493,6 +509,12 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { /// load. It is not a `LoadOptions` field either: #272 measured that a shorter threshold is worse /// under CPU starvation, and that conclusion is unchanged. private let connStallTimeout: TimeInterval + /// High-water mark this reader ends the connection at. Mode-dependent (live absorbs a + /// join burst the VOD value was never sized for — see the backpressure doc block) and + /// an init parameter for the same reason `connStallTimeout` is one: a process-wide + /// hook would leak into whatever suite runs concurrently. The shipped values are the + /// two statics above. + private let winHighWater: Int private var throttleVClockNs: UInt64 = 0 private let throttleLock = NSLock() @@ -500,7 +522,7 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { /// origin at once and the line used to name none of them. private let label: String - init(url: URL, extraHeaders: [String: String] = [:], label: String = "source", chunkSize: Int = 4 * 1024 * 1024, prefetchEnabled: Bool = true, isLive: Bool = false, chunkRequestTimeout: TimeInterval = 35, chunkMaxRetries: Int = 3, boundedInitialFetch: Int64? = nil, connStallTimeout: TimeInterval = AVIOReader.connStallTimeoutDefault) { + init(url: URL, extraHeaders: [String: String] = [:], label: String = "source", chunkSize: Int = 4 * 1024 * 1024, prefetchEnabled: Bool = true, isLive: Bool = false, chunkRequestTimeout: TimeInterval = 35, chunkMaxRetries: Int = 3, boundedInitialFetch: Int64? = nil, connStallTimeout: TimeInterval = AVIOReader.connStallTimeoutDefault, windowHighWater: Int? = nil) { self.url = url self.label = label self.extraHeaders = extraHeaders @@ -513,6 +535,8 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { self.throttleKbps = AetherEngine.sourceThrottleKbpsForTesting self.backoffScale = AetherEngine.reconnectBackoffScaleForTesting self.connStallTimeout = max(0.05, connStallTimeout) + self.winHighWater = max(1, windowHighWater + ?? (isLive ? Self.liveWinHighWaterDefault : Self.winHighWaterDefault)) } /// Slow-CDN simulation: hold delivered bytes to `throttleKbps` by sleeping the demux thread before the @@ -2099,7 +2123,7 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { var overshootToLog: Int64? = nil if connEndedByBackpressure { postEndDeliveryBytes += Int64(count) - if postEndDeliveryBytes > Int64(Self.winHighWater), !postEndOvershootLogged { + if postEndDeliveryBytes > Int64(winHighWater), !postEndOvershootLogged { postEndOvershootLogged = true overshootToLog = postEndDeliveryBytes } @@ -2143,7 +2167,7 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { // the frontier once the consumer drains below low water. var toCancel: URLSessionDataTask? let ahead = window.count - max(0, Int(position - winStart)) - if ahead > Self.winHighWater, !connEnded, !isClosed, activeTask != nil { + if ahead > winHighWater, !connEnded, !isClosed, activeTask != nil { connEndedByBackpressure = true connEnded = true toCancel = activeTask diff --git a/Tests/AetherEngineTests/LiveWindowBackpressureTests.swift b/Tests/AetherEngineTests/LiveWindowBackpressureTests.swift new file mode 100644 index 00000000..1ce44916 --- /dev/null +++ b/Tests/AetherEngineTests/LiveWindowBackpressureTests.swift @@ -0,0 +1,113 @@ +import Testing +import Foundation +@testable import AetherEngine + +/// Live sources are open-ended by design, so the high-water end is the ONLY thing that +/// ever terminates a healthy live connection — and "re-request at the frontier" is a lie +/// to a live origin: the bytes broadcast during the drain are gone, and the demuxer +/// rejoins on a corrupt TS packet. IPTV panels also serve their ring buffer as a join +/// burst at line rate on every (re)connect, so the 16 MB VOD cap turned each reconnect +/// into the cause of the next one, forever. The live threshold absorbs the burst ONCE; +/// the end-and-refill stays as the memory backstop for a "live" source that sustainedly +/// outruns realtime (a misdeclared VOD). These tests pin both halves. +@Suite("AVIOReader live window backpressure") +struct LiveWindowBackpressureTests { + + @Test("a live join burst past the VOD high water keeps its one connection", + .timeLimit(.minutes(2))) + func liveJoinBurstKeepsConnection() async throws { + // The join-burst shape: 24 MB delivered at line rate into a stalled consumer, + // then silence with the connection held open — no FIN, no error, exactly a live + // origin that has caught up to realtime. + let burst: Int64 = 24 * 1024 * 1024 + let serverMaybe = ThrottledOriginServer( + totalSize: 512 * 1024 * 1024, + respond: { _, _, _ in .serveThenGoSilent(afterBytes: burst) } + ) + let server = try #require(serverMaybe) + defer { server.stop() } + let reader = AVIOReader(url: URL(string: "http://127.0.0.1:\(server.port)/live.ts")!, + isLive: true, + connStallTimeout: 600) + defer { reader.markClosed(); reader.close() } + try reader.open() + + // The origin must be able to FINISH its burst: on the 16 MB threshold the reader + // cancelled the task mid-burst and the remaining bytes were never accepted. + let deadline = Date().addingTimeInterval(30) + while server.bytesWritten < burst && Date() < deadline { + try await Task.sleep(for: .milliseconds(50)) + } + #expect(server.bytesWritten >= burst, + "origin only placed \(server.bytesWritten / (1024 * 1024)) MB of its \(burst / (1024 * 1024)) MB burst; the reader ended the connection") + + // Let the in-flight tail land in the window, then pin the live contract: the + // burst is absorbed, the connection is NOT voluntarily ended, and no re-request + // was ever issued. + let settle = Date().addingTimeInterval(10) + while reader.windowDiagnostics.aheadBytes < Int(burst) && Date() < settle { + try await Task.sleep(for: .milliseconds(50)) + } + let diag = reader.windowDiagnostics + #expect(diag.aheadBytes >= Int(burst), + "window holds \(diag.aheadBytes / (1024 * 1024)) MB of the burst") + #expect(!diag.parked, "a live burst inside the live threshold must not trip the end") + #expect(reader.hasLiveConnectionForTesting, + "the live connection must survive the burst") + #expect(server.rangeRequestCount == 1, + "a live reader re-requested mid-stream: \(server.requestedRanges)") + #expect(server.requestedRanges.first?.end == nil, + "the live request must be open-ended") + } + + @Test("a live source past the live backstop is ended, stays bounded, and refills at the frontier", + .timeLimit(.minutes(2))) + func liveBackstopEndsAndRefills() async throws { + let server = try #require(ThrottledOriginServer(totalSize: 64 * 1024 * 1024)) + defer { server.stop() } + // A shrunken backstop keeps the test inside seconds; the shipped live value only + // changes WHERE the end fires, not what it does. (4 MB sits under winLowWater, + // which just means the refill gate is already open on the first read.) + let reader = AVIOReader(url: URL(string: "http://127.0.0.1:\(server.port)/live.ts")!, + isLive: true, + connStallTimeout: 600, + windowHighWater: 4 * 1024 * 1024) + defer { reader.markClosed(); reader.close() } + try reader.open() + + // A "live" origin outrunning realtime into a stalled consumer: the backstop must + // end the connection (bounded memory, the #310 contract, unchanged for live). + let deadline = Date().addingTimeInterval(15) + while reader.hasLiveConnectionForTesting && Date() < deadline { + try await Task.sleep(for: .milliseconds(50)) + } + #expect(!reader.hasLiveConnectionForTesting, + "the live backstop must still end a connection that outruns the consumer") + #expect(reader.windowDiagnostics.parked, + "the backstop end must be recorded as backpressure so the refill owns it") + #expect(server.rangeRequestCount == 1, + "the refill must not fire while nothing drains: \(server.requestedRanges)") + #expect(server.bytesWritten < 24 * 1024 * 1024, + "origin served \(server.bytesWritten / (1024 * 1024)) MB past a 4 MB backstop") + + // Draining must trigger the frontier refill and keep delivering fresh bytes. + let sliceCap = 256 * 1024 + let target = 12 * 1024 * 1024 + let buf = UnsafeMutablePointer.allocate(capacity: sliceCap) + defer { buf.deallocate() } + var got = 0 + let readDeadline = Date().addingTimeInterval(30) + while got < target && Date() < readDeadline { + let n = reader.read(into: buf, size: Int32(sliceCap)) + if n <= 0 { break } + got += Int(n) + } + #expect(got >= target, "only \(got / (1024 * 1024)) MB delivered after the backstop") + #expect(server.rangeRequestCount >= 2, "the frontier refill never fired") + let starts = server.requestedRanges.map(\.start) + #expect(starts == starts.sorted(), + "a live refill went backwards past the frontier: \(server.requestedRanges)") + #expect(server.requestedRanges.allSatisfy { $0.end == nil }, + "every live request must be open-ended: \(server.requestedRanges)") + } +} From 8a8e00b0b7252180007da9e5ed7fa6e0a0723102 Mon Sep 17 00:00:00 2001 From: Andreas Busslinger Date: Fri, 7 Aug 2026 17:46:41 +0200 Subject: [PATCH 2/3] fix(io): 509 is a metered origin - keep the pin, pay the rate-limit ladder (#307 follow-up) 509 "Bandwidth Limit Exceeded" is what a connection-capped IPTV panel answers while the slot the reader is replacing has not been torn down server-side yet. It classified as a hard 5xx, so every attempt dropped the pinned post-redirect URL and re-resolved through the portal: latency per attempt, plus the second request against the very origin that has no room for it, which is the 519ae26e reasoning left incomplete. And the hammering never paced itself: ~8 MB of progress per cycle reset the unproductive streak every time, so a permanent 509 ground through 13 attempts with 12 portal re-resolves at zero backoff on the loopback repro before the unproductive cap ended it. - One classifier, isRateLimitStatus (429/503/509), used by the pin-drop exemption, both reconnect ladders, the detour fetch, and the Retry-After parse - honoured on 509 when a panel sends it. - A lingering-slot 509 now keeps the pin and retries in place; a permanent 509 pays the rate-limit streak and its bounded give-up (7 attempts, paced) instead of the hard-5xx grind. Test plan: swift test on macOS 26 (1564 tests green; new cases in the resolved-URL suite: a two-attempt 509 keeps the pin end to end, a permanent 509 gives up at the rate-limit cap without ever re-resolving through the source, and the classifier assertions cover 509 against 500/502/504). Both verified failing on the unfixed base. Field shape from an Apple TV 4K (tvOS 26) against an Xtream aggregator with max_connections=1 whose external portal answers 509 while the replaced connection's slot lingers. Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 10 ++ Sources/AetherEngine/Demuxer/AVIOReader.swift | 54 +++++---- .../ResolvedURLInvalidationTests.swift | 111 ++++++++++++++++++ 3 files changed, 153 insertions(+), 22 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d69f9a5b..b6ad12ff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -27,6 +27,16 @@ the public-API contract. absorbed once, steady state plateaus at burst size with the connection never voluntarily ended, and the end-and-refill survives unchanged as the memory backstop for a "live" source that sustainedly outruns realtime. +- **HTTP 509 from a pinned redirect target is treated as metering, not as a dead pin.** + 509 "Bandwidth Limit Exceeded" is what a connection-capped IPTV panel answers while the slot + the reader is replacing has not been torn down server-side yet. It classified as a hard 5xx, + so every attempt dropped the pinned post-redirect URL and re-resolved through the portal — + latency per attempt, plus the second request against the very origin that has no room for it, + which is the 519ae26e reasoning left incomplete (a permanent 509 ground through 13 attempts + with 12 portal re-resolves at zero backoff, because ~8 MB of progress per cycle reset the + unproductive streak every time). 509 now keeps the pin and pays the rate-limit streak and + backoff alongside 429/503, honouring Retry-After when sent, with the same bounded give-up + (#307 follow-up). ## [6.15.1] - 2026-08-08 diff --git a/Sources/AetherEngine/Demuxer/AVIOReader.swift b/Sources/AetherEngine/Demuxer/AVIOReader.swift index 0aaff20d..d98dba83 100644 --- a/Sources/AetherEngine/Demuxer/AVIOReader.swift +++ b/Sources/AetherEngine/Demuxer/AVIOReader.swift @@ -5,7 +5,7 @@ import Libavutil /// Custom AVIO context feeding FFmpeg via URLSession. Three modes: /// - **Persistent** (known size + prefetch=true, playback path): single long-lived -/// `Range: bytes=-` GET into a sliding window; reconnects on drop/429/503. +/// `Range: bytes=-` GET into a sliding window; reconnects on drop/429/503/509. /// Fix for AetherEngine#25 (CDN stutter collapsing playback). See `readPersistent`. /// - **Seekable chunked** (known size + prefetch=false, still/frame-extraction): /// discrete Range chunks for random access. See `readSeekable`. @@ -108,12 +108,22 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { return status == 401 || status == 403 || status == 404 || status == 410 } + /// Rate-limit-shaped statuses: the origin is metering us, not failing. 429/503 carry + /// Retry-After (#71); 509 "Bandwidth Limit Exceeded" (nonstandard) is what a + /// connection-capped IPTV panel answers while its slot is still occupied by the + /// connection being replaced — the slot frees in seconds, the pinned redirect target + /// is fine, and re-resolving through the portal spends the one request there is no + /// room for (519ae26e, #307 follow-up). + static func isRateLimitStatus(_ status: Int) -> Bool { + return status == 429 || status == 503 || status == 509 + } + /// Hard server errors answered by a pinned post-redirect URL: the redirect target - /// may be dead or expired while the source URL would mint a fresh one. 503 is - /// excluded — it is rate limiting (#71), carries Retry-After, and the pin is not - /// the problem there. + /// may be dead or expired while the source URL would mint a fresh one. Rate-limit + /// statuses are excluded — the origin is metering us, and the pin is not the + /// problem there. static func isResolvedHardServerError(_ status: Int) -> Bool { - return status >= 500 && status != 503 + return status >= 500 && !isRateLimitStatus(status) } // Cumulative bytes fetched since open; memory probe compares against RSS growth. @@ -301,7 +311,7 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { min(detourFetchBudgetSeconds, chunkRequestTimeout) } - // Cap on CONSECUTIVE rate-limited (429/503) network attempts before giving up cleanly. + // Cap on CONSECUTIVE rate-limited (429/503/509) network attempts before giving up cleanly. // Distinct axis from unproductiveReconnects: NOT reset by seekReconnect, so parse-driven // seeks cannot mask a throttled origin into an infinite reconnect loop (AetherEngine#71). private static let rateLimitMaxStreak = 6 @@ -316,7 +326,7 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { // Connection state. private var connEnded = false private var connStatus = 0 - // Retry-After seconds from 429/503, honoured before reconnect. + // Retry-After seconds from a rate-limit status, honoured before reconnect. private var connRetryAfter: TimeInterval = 0 // Bumped on every (re)connect; stale delegate callbacks are ignored. private var connGeneration = 0 @@ -408,7 +418,7 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { // Consecutive unproductive reconnects (demux-thread-only). private var unproductiveReconnects = 0 private var bytesAtLastReconnect: Int64 = 0 - // Consecutive 429/503 attempts; survives seekReconnect, resets on real read progress (#71). + // Consecutive rate-limited attempts; survives seekReconnect, resets on real read progress (#71). private var rateLimitStreak = 0 /// Detour LRU block cache (its own leaf lock, never held across `fetchChunk`/network or @@ -1299,7 +1309,7 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { // RETRY the detour fetch; do NOT open a fresh connection (that re-enters // the 429 churn the cache exists to remove). Give up cleanly at the cap. if recordRateLimitAndShouldGiveUp() { - EngineLog.emit("[AVIOReader] Detour rate-limit gave up at offset \(curPosition) (\(rateLimitStreak) consecutive 429/503)", category: .demux) + EngineLog.emit("[AVIOReader] Detour rate-limit gave up at offset \(curPosition) (\(rateLimitStreak) consecutive rate-limited)", category: .demux) return totalRead > 0 ? Int32(totalRead) : -1 } let backoffStart = DispatchTime.now() @@ -1446,7 +1456,7 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { continue } - // Connection ended before EOF; reconnect at frontier. Honour Retry-After for 429/503. + // Connection ended before EOF; reconnect at frontier. Honour Retry-After when rate-limited. winCond.unlock() // #220/#310: we ended it ourselves at high water and the consumer has now emptied // the window (the low-water refill normally fires first; this is the backstop for @@ -1470,14 +1480,14 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { timedReconnect(seek: false, at: frontier) continue } - // A 429/503 is rate limiting, not a dead source: drive give-up + backoff off the + // A 429/503/509 is rate limiting, not a dead source: drive give-up + backoff off the // rate-limit streak, which (unlike unproductiveReconnects) survives the seekReconnect // that parse seeks fire, so a throttled origin fails cleanly instead of looping (#71). - let isRateLimited = (status == 429 || status == 503) + let isRateLimited = Self.isRateLimitStatus(status) let giveUp = isRateLimited ? recordRateLimitAndShouldGiveUp() : recordReconnectAndShouldGiveUp(status: status) if giveUp { - let streakDesc = isRateLimited ? "\(rateLimitStreak) consecutive 429/503" : "\(unproductiveReconnects) unproductive" + let streakDesc = isRateLimited ? "\(rateLimitStreak) consecutive rate-limited" : "\(unproductiveReconnects) unproductive" EngineLog.emit("[AVIOReader] \(label) reconnect exhausted at offset \(frontier) status=\(status) (\(streakDesc))\(isLive ? " [live source lost]" : "")", category: .demux) emitNetworkPhase(.flowing) // reader is exiting; let state carry the terminal outcome (#85) if isLive { @@ -1494,7 +1504,7 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { // not a transient: drop it so the retry re-resolves through the source URL // for a fresh redirect. No-op when nothing is pinned. // - // A rate-limit streak is deliberately NOT a reason to drop it: 429/503 says the + // A rate-limit streak is deliberately NOT a reason to drop it: 429/503/509 says the // origin is metering us, not that the target is dead (#71), and re-resolving // spends a second request on the very origin that is refusing them. On the // connection-capped panel behind #307 that is the request that cannot be spared. @@ -1542,10 +1552,10 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { unproductiveReconnects += 1 } bytesAtLastReconnect = now - // Hard 4xx/5xx (not 429/503 which carry Retry-After) on a source that has - // never delivered a byte = server-side failure (e.g. Jellyfin 500 after + // Hard 4xx/5xx (not the rate-limit statuses, which are metering) on a source that + // has never delivered a byte = server-side failure (e.g. Jellyfin 500 after // transcode-failure latency ~15-20s/attempt). One retry, then out. - let isHardError = status >= 400 && status != 429 && status != 503 + let isHardError = status >= 400 && !Self.isRateLimitStatus(status) if now == 0 && isHardError { return unproductiveReconnects > 1 } @@ -1590,7 +1600,7 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { /// when playback genuinely starves. Demux-thread-only. private func chargeFaultedRunwayRefill(at frontier: Int64, ahead: Int, status: Int, retryAfter: TimeInterval) -> Bool { - let isRateLimited = (status == 429 || status == 503) + let isRateLimited = Self.isRateLimitStatus(status) let giveUp = isRateLimited ? recordRateLimitAndShouldGiveUp() : recordReconnectAndShouldGiveUp(status: status) if giveUp { @@ -1633,7 +1643,7 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { } } - /// Increments the consecutive 429/503 streak; returns true once the bounded cap is hit. + /// Increments the consecutive rate-limited streak; returns true once the bounded cap is hit. /// Demux-thread-only. Deliberately NOT reset by `seekReconnect` (parse seeks must not mask a /// throttled origin into an endless reconnect loop, #71); only real read progress clears it. /// Internal (not private) so the bounded give-up is unit-tested without a live origin. @@ -1694,7 +1704,7 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { return .served(n) } - /// Single Range fetch for a detour block over the pooled chunkSession. Surfaces 429/503 with + /// Single Range fetch for a detour block over the pooled chunkSession. Surfaces rate limiting with /// its Retry-After so the caller can back off in place rather than churn the connection (#71). private func detourFetchBlock(from offset: Int64, size: Int) -> DetourFetch { let rangeEnd = offset + Int64(size) - 1 @@ -1709,7 +1719,7 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { let (data, response) = try syncRequest(request, budget: budget) if let http = response as? HTTPURLResponse { let status = http.statusCode - if status == 429 || status == 503 { + if Self.isRateLimitStatus(status) { return .rateLimited(Self.parseRetryAfter(http)) } if status != 200 && status != 206 { @@ -2213,7 +2223,7 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { let status = http.statusCode var isOK = status == 200 || status == 206 var retryAfter: TimeInterval = 0 - if status == 429 || status == 503 { + if Self.isRateLimitStatus(status) { retryAfter = Self.parseRetryAfter(http) } var headerMs: Double? = nil diff --git a/Tests/AetherEngineTests/ResolvedURLInvalidationTests.swift b/Tests/AetherEngineTests/ResolvedURLInvalidationTests.swift index 5383976d..1170ad05 100644 --- a/Tests/AetherEngineTests/ResolvedURLInvalidationTests.swift +++ b/Tests/AetherEngineTests/ResolvedURLInvalidationTests.swift @@ -128,6 +128,111 @@ struct ResolvedURLInvalidationTests { "a metered refill re-resolved through the source: \(source.requestLog)") } + /// 509 "Bandwidth Limit Exceeded" is what a connection-capped panel answers while the + /// slot the reader is REPLACING has not been torn down server-side yet. The slot frees + /// in seconds and the pinned target is fine; dropping the pin sent every retry back + /// through the portal — latency plus the one request there is no room for. + @Test("a 509 refill keeps the pin instead of re-resolving through the source", + .timeLimit(.minutes(2))) + func connectionCappedRefillKeepsThePin() async throws { + AetherEngine.reconnectBackoffScaleForTesting = 0.02 + defer { AetherEngine.reconnectBackoffScaleForTesting = 1.0 } + + let firstRange: Int64 = 256 * 1024 + let attempts = AttemptCounter() + // CDN: the lingering-slot shape — 509 for the first two attempts at the boundary + // refill, then the slot has freed and it serves. + let cdnMaybe = ThrottledOriginServer( + totalSize: 64 * 1024 * 1024, + respond: { _, offset, _ in + offset == firstRange && attempts.next(for: offset) <= 2 + ? .status(509) : .serve206 + } + ) + let cdn = try #require(cdnMaybe) + defer { cdn.stop() } + let cdnPort = cdn.port + let sourceMaybe = ThrottledOriginServer( + totalSize: 64 * 1024 * 1024, + respond: { _, _, _ in .redirect(to: "http://127.0.0.1:\(cdnPort)/cdn/movie.bin") } + ) + let source = try #require(sourceMaybe) + defer { source.stop() } + + let reader = AVIOReader(url: URL(string: "http://127.0.0.1:\(source.port)/movie.bin")!, + boundedInitialFetch: firstRange) + defer { reader.markClosed(); reader.close() } + try reader.open() + + let sliceCap = 128 * 1024 + let buf = UnsafeMutablePointer.allocate(capacity: sliceCap) + defer { buf.deallocate() } + let target = Int(firstRange) + 256 * 1024 + var got = 0 + while got < target { + let n = reader.read(into: buf, size: Int32(min(sliceCap, target - got))) + if n <= 0 { break } + got += Int(n) + } + #expect(got == target, "read stopped at \(got) of \(target); the 509s were terminal") + + let sourceHitsAtBoundary = source.requestLog.filter { $0.start == firstRange } + #expect(sourceHitsAtBoundary.isEmpty, + "a 509 refill re-resolved through the source: \(source.requestLog)") + } + + /// An origin that answers 509 forever must get the PACED rate-limit ladder and its + /// bounded give-up — not the hard-5xx treatment (pin dropped every attempt, retries + /// through the portal at zero backoff until the unproductive cap). + @Test("a permanent 509 pays the rate-limit ladder and gives up cleanly", + .timeLimit(.minutes(2))) + func permanent509TakesTheRateLimitLadder() async throws { + AetherEngine.reconnectBackoffScaleForTesting = 0.02 + defer { AetherEngine.reconnectBackoffScaleForTesting = 1.0 } + + let firstRange: Int64 = 256 * 1024 + let cdnMaybe = ThrottledOriginServer( + totalSize: 64 * 1024 * 1024, + respond: { _, offset, _ in + offset == firstRange ? .status(509) : .serve206 + } + ) + let cdn = try #require(cdnMaybe) + defer { cdn.stop() } + let cdnPort = cdn.port + let sourceMaybe = ThrottledOriginServer( + totalSize: 64 * 1024 * 1024, + respond: { _, _, _ in .redirect(to: "http://127.0.0.1:\(cdnPort)/cdn/movie.bin") } + ) + let source = try #require(sourceMaybe) + defer { source.stop() } + + let reader = AVIOReader(url: URL(string: "http://127.0.0.1:\(source.port)/movie.bin")!, + boundedInitialFetch: firstRange) + defer { reader.markClosed(); reader.close() } + try reader.open() + + let sliceCap = 128 * 1024 + let buf = UnsafeMutablePointer.allocate(capacity: sliceCap) + defer { buf.deallocate() } + var got = 0 + while true { + let n = reader.read(into: buf, size: Int32(sliceCap)) + if n <= 0 { break } + got += Int(n) + } + #expect(got == Int(firstRange), + "everything before the metered boundary must still be served (got \(got))") + + let boundaryAttempts = cdn.requestLog.filter { $0.start == firstRange }.count + #expect(boundaryAttempts >= 2, "the ladder must retry a metered origin") + #expect(boundaryAttempts <= 7, + "509 must give up at the bounded rate-limit cap, not grind: \(boundaryAttempts) attempts") + let sourceHitsAtBoundary = source.requestLog.filter { $0.start == firstRange } + #expect(sourceHitsAtBoundary.isEmpty, + "a metered origin must keep the pin throughout: \(source.requestLog)") + } + @Test("hard-server-error classification excludes rate limiting") func classifierExcludesRateLimiting() { #expect(AVIOReader.isResolvedHardServerError(500)) @@ -135,7 +240,13 @@ struct ResolvedURLInvalidationTests { #expect(AVIOReader.isResolvedHardServerError(504)) #expect(!AVIOReader.isResolvedHardServerError(503), "503 is rate limiting (#71)") #expect(!AVIOReader.isResolvedHardServerError(429)) + #expect(!AVIOReader.isResolvedHardServerError(509), + "509 is a connection-capped panel metering us (#307 follow-up)") #expect(!AVIOReader.isResolvedHardServerError(404), "auth expiry is its own class") #expect(!AVIOReader.isResolvedHardServerError(200)) + #expect(AVIOReader.isRateLimitStatus(429)) + #expect(AVIOReader.isRateLimitStatus(503)) + #expect(AVIOReader.isRateLimitStatus(509)) + #expect(!AVIOReader.isRateLimitStatus(500)) } } From ef350fcde76053b7de995ab2b91ae0c6c1a0b069 Mon Sep 17 00:00:00 2001 From: Andreas Busslinger Date: Fri, 7 Aug 2026 18:15:35 +0200 Subject: [PATCH 3/3] fix(io): a live reconnect asks for the stream like a join, not at a byte frontier The reconnect request carried Range: bytes=-, but the frontier is reader bookkeeping - the window position delivered bytes are appended at - not a server-side byte address, because a live origin has none. Panels that ignore the offset and serve "from now" masked this; a panel that answers 416 to every offset it cannot satisfy turned each reconnect into an unrecoverable rejection loop. Field trace (Apple TV 4K, tvOS 26, Xtream panel): the origin cleanly completes every response after its ~14 MB ring burst, then 416'd the same frontier offset 35 generations in a row at ~1/s - the read-progress streak reset kept the ladder at streak=1 with 1 s backoff while the runway drained from 8 MB to zero, the re-resolve at streak=2 changed nothing (the offset was the problem, not the pin), and the session starved into an AVPlayer -12888. - Live persistent connections now always request `bytes=0-` - the one shape every origin serves, and the shape the join already uses. The append anchors delivered bytes at the window frontier exactly as it always has (the persistent path appends sequentially and never re-anchors on Content-Range), so the logical bookkeeping is unchanged. - VOD requests are untouched: bounded ranges and frontier offsets remain correct where byte addresses are real. Test plan: swift test on macOS 26 (1565 tests green). New case in the live-window suite: an origin that completes every response after a 4 MB burst and rejects any nonzero offset with 416 must keep delivering across reconnect cycles with every request at bytes=0-. On the unfixed code the case reproduces the field failure verbatim: exactly one burst delivered, then the rejection loop starves the read. Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 10 ++++ Sources/AetherEngine/Demuxer/AVIOReader.swift | 15 ++++-- .../LiveWindowBackpressureTests.swift | 48 +++++++++++++++++-- 3 files changed, 67 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b6ad12ff..54afac36 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -27,6 +27,16 @@ the public-API contract. absorbed once, steady state plateaus at burst size with the connection never voluntarily ended, and the end-and-refill survives unchanged as the memory backstop for a "live" source that sustainedly outruns realtime. +- **A live reconnect asks for the stream the way a join does, instead of at a byte frontier.** + The reconnect request carried `Range: bytes=-`, but the frontier is reader + bookkeeping — the window position delivered bytes are appended at — not a server-side byte + address, because a live origin has none. Panels that ignore the offset and serve "from now" + masked this; a panel that answers 416 to every offset it cannot satisfy turned each reconnect + into an unrecoverable rejection loop (field trace: a panel that cleanly completes every + response after its ~14 MB ring burst then 416'd the same frontier 35 generations in a row, + ~1/s, while the runway drained from 8 MB to zero and the session starved). Live requests are + now always `bytes=0-` — the one shape every origin serves, and the shape the join already + uses — and the append anchors the bytes at the frontier exactly as it always has. - **HTTP 509 from a pinned redirect target is treated as metering, not as a dead pin.** 509 "Bandwidth Limit Exceeded" is what a connection-capped IPTV panel answers while the slot the reader is replacing has not been torn down server-side yet. It classified as a hard 5xx, diff --git a/Sources/AetherEngine/Demuxer/AVIOReader.swift b/Sources/AetherEngine/Demuxer/AVIOReader.swift index d98dba83..8cd3943f 100644 --- a/Sources/AetherEngine/Demuxer/AVIOReader.swift +++ b/Sources/AetherEngine/Demuxer/AVIOReader.swift @@ -1935,8 +1935,9 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { // MARK: - Persistent Connection (lifecycle + delegate callbacks) - /// Open a fresh Range: bytes=- connection. Bumps generation so - /// late callbacks from the old connection are ignored. + /// Open a fresh Range: bytes=- connection (live: always `bytes=0-`, see the + /// request construction below). Bumps generation so late callbacks from the old + /// connection are ignored. private func startPersistentConnection(at offset: Int64, boundedTo: Int64? = nil) { winCond.lock() connGeneration &+= 1 @@ -2005,7 +2006,15 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { if let resolvedBound { request.setValue("bytes=\(offset)-\(offset + resolvedBound - 1)", forHTTPHeaderField: "Range") } else { - request.setValue("bytes=\(offset)-", forHTTPHeaderField: "Range") + // Live: `offset` is reader bookkeeping (the window frontier the delivered bytes + // are appended at), not a server-side position — a live origin has no byte + // addresses, and panels disagree on what a nonzero offset means: some ignore it + // and serve from now (which is why the frontier request ever worked), others + // answer 416 to every offset they cannot satisfy, turning each reconnect into an + // unrecoverable rejection loop. Ask for the stream the way a join does + // (`bytes=0-`, the one shape every origin serves) and let the append anchor the + // bytes at the frontier, exactly as it already does. + request.setValue("bytes=\(isLive ? 0 : offset)-", forHTTPHeaderField: "Range") } request.timeoutInterval = 0 // long-lived; stalls handled by the reader applyExtraHeaders(&request) diff --git a/Tests/AetherEngineTests/LiveWindowBackpressureTests.swift b/Tests/AetherEngineTests/LiveWindowBackpressureTests.swift index 1ce44916..8a8230a5 100644 --- a/Tests/AetherEngineTests/LiveWindowBackpressureTests.swift +++ b/Tests/AetherEngineTests/LiveWindowBackpressureTests.swift @@ -104,10 +104,52 @@ struct LiveWindowBackpressureTests { } #expect(got >= target, "only \(got / (1024 * 1024)) MB delivered after the backstop") #expect(server.rangeRequestCount >= 2, "the frontier refill never fired") - let starts = server.requestedRanges.map(\.start) - #expect(starts == starts.sorted(), - "a live refill went backwards past the frontier: \(server.requestedRanges)") + #expect(server.requestedRanges.allSatisfy { $0.start == 0 }, + "a live request carried a byte frontier the origin never promised to honour: \(server.requestedRanges)") #expect(server.requestedRanges.allSatisfy { $0.end == nil }, "every live request must be open-ended: \(server.requestedRanges)") } + + /// The field shape behind the fix's third half: a panel that CLEANLY ends every + /// connection after serving its ring-buffer burst, and answers 416 to any request + /// with a nonzero byte offset (a live stream has no byte addresses). Reconnecting + /// "at the frontier" against such an origin is an unrecoverable rejection loop: + /// every retry asks the same unsatisfiable offset until the runway drains and the + /// session starves. A live reconnect must ask for the stream the way a join does. + @Test("a live reconnect after a completed burst asks like a join, not at a frontier", + .timeLimit(.minutes(2))) + func liveReconnectAsksLikeAJoin() async throws { + // 4 MB per connection: the origin serves its "ring buffer" and completes the + // response; anything with offset > 0 is rejected the way the field panel does. + let serverMaybe = ThrottledOriginServer( + totalSize: 4 * 1024 * 1024, + respond: { _, offset, _ in offset > 0 ? .status(416) : .serve206 } + ) + let server = try #require(serverMaybe) + defer { server.stop() } + let reader = AVIOReader(url: URL(string: "http://127.0.0.1:\(server.port)/live.ts")!, + isLive: true, + connStallTimeout: 600) + defer { reader.markClosed(); reader.close() } + try reader.open() + + // Read through several burst-reconnect cycles: 10 MB needs at least three + // connections against a 4 MB-per-connection origin. + let sliceCap = 256 * 1024 + let target = 10 * 1024 * 1024 + let buf = UnsafeMutablePointer.allocate(capacity: sliceCap) + defer { buf.deallocate() } + var got = 0 + let deadline = Date().addingTimeInterval(30) + while got < target && Date() < deadline { + let n = reader.read(into: buf, size: Int32(sliceCap)) + if n <= 0 { break } + got += Int(n) + } + #expect(got >= target, + "only \(got / (1024 * 1024)) MB delivered; the reconnect starved on a rejected frontier") + #expect(server.rangeRequestCount >= 3, "expected one connection per burst cycle") + #expect(server.requestedRanges.allSatisfy { $0.start == 0 }, + "a live reconnect carried a frontier offset: \(server.requestedRanges)") + } }