diff --git a/CHANGELOG.md b/CHANGELOG.md index b4a2f5fd..54afac36 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,7 +10,43 @@ 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. +- **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, + 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/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..8cd3943f 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. @@ -210,7 +220,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 @@ -285,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 @@ -300,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 @@ -392,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 @@ -493,6 +519,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 +532,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 +545,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 @@ -1275,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() @@ -1422,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 @@ -1446,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 { @@ -1470,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. @@ -1518,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 } @@ -1566,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 { @@ -1609,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. @@ -1670,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 @@ -1685,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 { @@ -1901,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 @@ -1971,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) @@ -2099,7 +2142,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 +2186,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 @@ -2189,7 +2232,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/LiveWindowBackpressureTests.swift b/Tests/AetherEngineTests/LiveWindowBackpressureTests.swift new file mode 100644 index 00000000..8a8230a5 --- /dev/null +++ b/Tests/AetherEngineTests/LiveWindowBackpressureTests.swift @@ -0,0 +1,155 @@ +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") + #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)") + } +} 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)) } }