From ea1642a62ed4f1a135b2e0d383aa552f377163bf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Santos?= Date: Fri, 16 Oct 2026 00:51:43 +0200 Subject: [PATCH 1/2] fix(live): a live stream read off its URL rides out an IPTV panel's refusals A connection-limited panel answers a live TS stream's reconnects with 509 while another device on the account holds its last slot, for seconds to tens of seconds. The reader climbed its 0.5, 1, 2, 4, 8 s ladder, gave up after six refusals, and each refusal armed the origin's request pacer, which the reload after it waited out (up to 10 s a request). A live stream's 429/503/509 is now asked again every second, or after the origin's Retry-After up to 5 s, for up to 60 s, the way the live HLS ingest asks for its playlist; its refusals arm no pacer and lower no budget, and the books still say the origin refused so the revive arm can tell metered from lost. A live open the origin answers with a status fails typed with that status at once, where the read loop used to spend the open budget on the same request and hand back an untyped failure. LiveStreamRefusalRetryTests. --- CHANGELOG.md | 16 ++ Sources/AetherEngine/Demuxer/AVIOReader.swift | 76 +++++- .../LiveStreamRefusalRetryTests.swift | 235 ++++++++++++++++++ 3 files changed, 322 insertions(+), 5 deletions(-) create mode 100644 Tests/AetherEngineTests/LiveStreamRefusalRetryTests.swift diff --git a/CHANGELOG.md b/CHANGELOG.md index 8bca37f5..9e6c8556 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,22 @@ the public-API contract. ## [Unreleased] +### Fixed + +- **A live stream read off its URL and refused for a while keeps its session.** The continuous + MPEG-TS form of an IPTV channel is one connection for as long as it plays, and a connection-limited + panel answers its reconnects with 509 while another device on the account holds the last slot, for + seconds to tens of seconds. The reader climbed its 0.5, 1, 2, 4, 8 s ladder, gave up after six + refusals, and each refusal armed the origin's request pacer (a quiet of up to 15 s, with each + request waiting up to 10 s on it), which the reload after it then waited out. A live stream's + 429/503/509 is now asked again every second, or after the origin's Retry-After up to 5 s, for up to + 60 s, the way the live HLS ingest asks for its playlist (#778); its refusals arm no pacer and lower + no budget, and the books still say the origin refused, so the revive arm tells metered from lost. A + live open the origin answers with a status fails typed with that status at once + (`AVIOReaderError.httpStatus`, `sourceRateLimited` or `sourceRefused` at the host), where the read + loop used to spend the open budget on the same request and hand back an untyped read failure. + `LiveStreamRefusalRetryTests`. + ## [7.37.0] - 2026-10-15 ### Added diff --git a/Sources/AetherEngine/Demuxer/AVIOReader.swift b/Sources/AetherEngine/Demuxer/AVIOReader.swift index d2635a77..6016f98d 100644 --- a/Sources/AetherEngine/Demuxer/AVIOReader.swift +++ b/Sources/AetherEngine/Demuxer/AVIOReader.swift @@ -438,6 +438,18 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { private func noteOriginRefusal(status: Int, retryAfter: TimeInterval? = nil, respondedBy: URL? = nil) { let refusing = respondedBy ?? requestURL() + if isLive { + // A live stream's refusal is retried on its own short cadence (`liveRefusalRetryInterval`), + // which is the pacing: arming the pacer as well would hold the retry, and the reload or + // reopen after it, for the quiet period (up to 15 s, and up to 10 s of waiting on each + // request) over a refusal that clears in seconds. The books still carry the refusal, so + // the engine's revive arm can tell a metered source from a lost one. + OriginRequestBudget.shared.noteRefusalWitnessed(for: refusing) + if OriginRequestBudget.originKey(for: refusing) != OriginRequestBudget.originKey(for: url) { + OriginRequestBudget.shared.noteRefusalWitnessed(for: url) + } + return + } OriginRequestBudget.shared.noteRefusal(for: refusing, status: status, retryAfter: retryAfter) // The refusal usually comes back from the post-redirect CDN, while the engine's revive arm // only knows the URL the host loaded. Where those differ (a proxy that 302s to a signed CDN @@ -749,6 +761,20 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { // 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 + /// A LIVE stream refused mid-session is asked for again on a steady, short cadence rather than + /// up the exponential ladder, and for a bounded while rather than a bounded count. The refusal + /// this answers is a connection-limited IPTV panel's 509 while another device on the account + /// holds its last slot: it lasts seconds to tens of seconds and clears the moment that slot is + /// free, and every second of waiting past that is a second the picture stays frozen with the + /// origin ready to serve. One small request a second is what the panel's own clients do, and + /// what the live HLS ingest does for the same refusal (`HLSLiveIngestReader`). The origin's own + /// Retry-After stretches the wait, up to `liveRefusalMaximumRetryWait`. + static let liveRefusalRetryInterval: TimeInterval = 1 + static let liveRefusalMaximumRetryWait: TimeInterval = 5 + /// How long a live stream keeps asking before the refusal is the session's answer. + static let liveRefusalRetryBudget: TimeInterval = 60 + /// Test seam: a shorter budget, so the give-up is exercised without a minute of wall clock. + nonisolated(unsafe) static var liveRefusalRetryBudgetForTesting: TimeInterval? // Rate-limited attempts that keep the pinned redirect target before one attempt through the // source URL is spent on a fresh redirect. Three paced attempts (~7 s of ladder) ride out the // lingering-slot 509 of a connection-capped panel (#307 follow-up: the slot frees in seconds); @@ -973,6 +999,8 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { // streak, resets on real read progress and whenever the header store's revision moves. private var credentialRefusalStreak = 0 private var credentialStreakRevision: UInt64? + /// When the current rate-limited streak began, for the live budget. Cleared with the streak. + private var rateLimitStreakStartedAt: DispatchTime? /// Detour LRU block cache (its own leaf lock, never held across `fetchChunk`/network or /// `winCond`). Stores only full-size blocks; short bodies are served once but never cached @@ -1283,6 +1311,18 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { winCond.unlock() startPersistentConnection(at: 0, boundedTo: boundedInitialFetch) var opened = awaitFirstPersistentData() + // A live open the origin answered with a status is reported as that answer, the + // way the live HLS ingest reports a refused join: the host retries a 509 on its own + // cadence with the right words for it, where the read loop's ladder would have spent + // the open budget on the same request and handed back an untyped read failure. + if !opened, isLive, let refusal = refusalOfLiveOpenConnection() { + EngineLog.emit( + "[AVIOReader] \(label) live open refused status=\(refusal); failing the open typed", + category: .demux) + markClosed() + close() + throw AVIOReaderError.httpStatus(refusal) + } // AE#693: one refusal is the origin's answer to that request, not yet to the // resource. The #378 path below takes a 401/403/404/410 here as final and settles // the session forward-only, which costs it every seek and the resume position. An @@ -1548,6 +1588,15 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { return (connEnded && Self.isResolvedExpiryStatus(connStatus)) ? connStatus : nil } + /// The status a live open's first connection ended with before any data, when it was an answer + /// (any 4xx or 5xx, the rate-limit statuses included); nil for a connection still open, one that + /// delivered, or one that ended with no status (a transport failure, which the read loop rides). + private func refusalOfLiveOpenConnection() -> Int? { + winCond.lock() + defer { winCond.unlock() } + return (connEnded && window.isEmpty && connStatus >= 400) ? connStatus : nil + } + /// The streaming GET was answered with a status instead of a body, or the ranged open was /// already refused with one and the unranged GET then delivered nothing either. Either way the /// demuxer would be handed an empty stream (or an error page) and report it as invalid data; @@ -2379,6 +2428,7 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { if fetched { unproductiveReconnects = 0 rateLimitStreak = 0 + rateLimitStreakStartedAt = nil emitNetworkPhase(.flowing) // the fetch crossed the network: the origin is serving (#85/#410) } detourTrackSequential(at: curPosition, length: n) @@ -2443,6 +2493,7 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { unproductiveReconnects = 0 // real progress rateLimitStreak = 0 // real progress clears the 429 give-up streak (#71) credentialRefusalStreak = 0 // and the credential one (#792) + rateLimitStreakStartedAt = nil emitNetworkPhase(.flowing) // recovered: source delivering again (#85/#410) } // No flow installed and the consumer has drawn down to low water: request at the @@ -2610,7 +2661,8 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { // nothing, so the exponential pacing charged against the target that did refuse is not // its debt. A server-sent Retry-After still applies: that is the origin's own ask, and // the source belongs to the same origin. - backoffBeforeReconnect(streak: repinned ? 0 : backoffStreak, retryAfter: retryAfter) + backoffBeforeReconnect(streak: repinned ? 0 : backoffStreak, retryAfter: retryAfter, + isRateLimited: isRateLimited) diag.recordBackoff(ms: msSince(backoffStart)) timedReconnect(seek: false, at: frontier) } @@ -2725,7 +2777,10 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { /// Exponential backoff (0.5s..8s) growing with streak; immediate on streak=0. How long the /// ladder waits before its next attempt. Shared by the blocking backoff below and the /// non-blocking one in `chargeFaultedRunwayRefill`, so both pace an origin identically. - private func backoffDelay(streak: Int, retryAfter: TimeInterval) -> TimeInterval { + private func backoffDelay(streak: Int, retryAfter: TimeInterval, isRateLimited: Bool = false) -> TimeInterval { + if isLive, isRateLimited { + return min(max(Self.liveRefusalRetryInterval, retryAfter), Self.liveRefusalMaximumRetryWait) * backoffScale + } let expo = streak <= 0 ? 0.0 : min(Double(1 << min(streak, 4)) * 0.5, 8.0) return min(max(expo, retryAfter), 15.0) * backoffScale } @@ -2771,7 +2826,8 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { : isCredentialRefusal ? credentialRefusalStreak : unproductiveReconnects let repinned = isCredentialRefusal ? false : dropPinIfTheRefusalCallsForIt(isRateLimited: isRateLimited) lastUnplannedReconnectAt = Date() - let delay = backoffDelay(streak: repinned ? 0 : streak, retryAfter: retryAfter) + let delay = backoffDelay(streak: repinned ? 0 : streak, retryAfter: retryAfter, + isRateLimited: isRateLimited) winCond.lock() nextFaultedRefillAt = Date().addingTimeInterval(delay) winCond.unlock() @@ -2784,8 +2840,8 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { } /// Blocking form of the wait above: sleeps in 0.1s slices so a close is honoured promptly. - private func backoffBeforeReconnect(streak: Int, retryAfter: TimeInterval) { - let total = backoffDelay(streak: streak, retryAfter: retryAfter) + private func backoffBeforeReconnect(streak: Int, retryAfter: TimeInterval, isRateLimited: Bool = false) { + let total = backoffDelay(streak: streak, retryAfter: retryAfter, isRateLimited: isRateLimited) if total <= 0 { return } var slept = 0.0 while slept < total { @@ -2802,6 +2858,16 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { /// Internal (not private) so the bounded give-up is unit-tested without a live origin. func recordRateLimitAndShouldGiveUp() -> Bool { rateLimitStreak += 1 + if isLive { + // A live stream is given a while, not a count: at one attempt a second the count would + // be spent in six seconds, inside the refusal this cadence exists to ride out. + let now = DispatchTime.now() + let started = rateLimitStreakStartedAt ?? now + rateLimitStreakStartedAt = started + let budget = Self.liveRefusalRetryBudgetForTesting ?? Self.liveRefusalRetryBudget + let elapsed = Double(now.uptimeNanoseconds &- started.uptimeNanoseconds) / 1_000_000_000 + return elapsed > budget + } return rateLimitStreak > Self.rateLimitMaxStreak } diff --git a/Tests/AetherEngineTests/LiveStreamRefusalRetryTests.swift b/Tests/AetherEngineTests/LiveStreamRefusalRetryTests.swift new file mode 100644 index 00000000..bd27a48b --- /dev/null +++ b/Tests/AetherEngineTests/LiveStreamRefusalRetryTests.swift @@ -0,0 +1,235 @@ +import Testing +import Foundation +@testable import AetherEngine + +/// A live stream read straight off its URL (an IPTV panel's continuous MPEG-TS form of a channel) +/// and refused for a while, as a connection-limited panel refuses an account's requests with 509 +/// while another device holds its last slot, keeps its session: the reader asks again every second +/// and carries on the moment it is served, the refusals arm no pacer, and a refused open is the +/// origin's answer, typed, rather than a minute of the same request. +@Suite("Live stream reader rides out refusals", .offCooperativePool, .timeLimit(.minutes(2))) +struct LiveStreamRefusalRetryTests { + + // MARK: - Loopback origin + + /// One HTTP/1.1 response per connection (Connection: close), scripted per request. + private final class ScriptedOrigin: @unchecked Sendable { + struct Response { + var status: Int + var headers: [String: String] = [:] + var body = Data() + } + + let port: UInt16 + private let listener: LoopbackListener + private let respond: @Sendable (_ index: Int) -> Response + private let lock = NSLock() + private var times: [Date] = [] + + /// `respond` gets how many requests came before this one. + init?(respond: @escaping @Sendable (_ index: Int) -> Response) { + self.respond = respond + guard let listener = LoopbackListener(backlog: 16) else { return nil } + self.listener = listener + port = listener.port + listener.start { [weak self] conn in + guard let self else { close(conn); return false } + Thread.detachNewThread { [weak self] in + self?.serve(conn) + } + return true + } + } + + func stop() { + listener.stop() + } + + var requestTimes: [Date] { + lock.withLock { times } + } + + var url: URL { + URL(string: "http://127.0.0.1:\(port)/live/user/pass/1.ts")! + } + + private func serve(_ conn: Int32) { + defer { close(conn) } + var request = Data() + var buf = [UInt8](repeating: 0, count: 4096) + while request.range(of: Data("\r\n\r\n".utf8)) == nil { + let n = read(conn, &buf, buf.count) + guard n > 0 else { return } + request.append(contentsOf: buf[0.. 64 * 1024 { return } + } + let index: Int = lock.withLock { + times.append(Date()) + return times.count - 1 + } + let response = respond(index) + var header = "HTTP/1.1 \(response.status) \(response.status == 200 ? "OK" : "Status")\r\n" + header += "Content-Type: video/mp2t\r\n" + for (field, value) in response.headers { + header += "\(field): \(value)\r\n" + } + header += "Content-Length: \(response.body.count)\r\nConnection: close\r\n\r\n" + _ = header.withCString { write(conn, $0, strlen($0)) } + response.body.withUnsafeBytes { raw in + var offset = 0 + while offset < raw.count { + let n = write(conn, raw.baseAddress!.advanced(by: offset), raw.count - offset) + guard n > 0 else { return } + offset += n + } + } + } + } + + /// Valid sync bytes at 188-byte cadence, enough to be read in several reader calls. + private static func streamBytes(count packets: Int) -> Data { + var data = Data(capacity: packets * 188) + for packet in 0.. AVIOReader { + AVIOReader(url: url, label: "live", prefetchEnabled: true, isLive: true) + } + + /// Reads until `expectedBytes` have come through or the reader fails; the bytes, and the code + /// the read ended with when it did not deliver them. + private static func drain(_ reader: AVIOReader, expectedBytes: Int) -> (bytes: Int, failure: Int32?) { + let buf = UnsafeMutablePointer.allocate(capacity: 64 * 1024) + defer { buf.deallocate() } + var total = 0 + while total < expectedBytes { + let n = reader.read(into: buf, size: 64 * 1024) + if n <= 0 { return (total, n) } + total += Int(n) + } + return (total, nil) + } + + private static func gaps(_ times: [Date]) -> [TimeInterval] { + zip(times, times.dropFirst()).map { $1.timeIntervalSince($0) } + } + + // MARK: - Tests + + @Test("a stream refused mid-session is asked for again every second until it is served") + func refusedMidStreamIsRetriedEverySecond() throws { + let first = Self.streamBytes(count: 400) + let second = Self.streamBytes(count: 400) + let origin = try #require(ScriptedOrigin { index in + switch index { + case 0: return .init(status: 200, body: first) + case 1, 2, 3: return .init(status: 509) + default: return .init(status: 200, body: second) + } + }) + defer { origin.stop() } + let reader = Self.makeLiveReader(origin.url) + defer { reader.markClosed(); reader.close() } + try reader.open() + + let drained = Self.drain(reader, expectedBytes: first.count + second.count) + + #expect(drained.failure == nil) + #expect(drained.bytes >= first.count + second.count) + let times = origin.requestTimes + #expect(times.count >= 5) + // The three refusals are retried a second apart, give or take the request itself and a + // loaded test host; never up the 0.5, 1, 2, 4 s ladder, whose third rung alone is 4 s. + let retryGaps = Array(Self.gaps(times).dropFirst()).prefix(3) + #expect(retryGaps.count == 3) + for gap in retryGaps { + #expect(gap >= 0.9 && gap <= 3.5, "retry gap \(gap)s") + } + #expect(retryGaps.reduce(0, +) <= 7, "retry gaps \(Array(retryGaps))") + // The refusals armed no pacer and lowered no budget: the books hold no refusal for this + // origin (a witness stamp carries no count). + let books = OriginRequestBudget.shared.snapshot(for: origin.url) + #expect((books?.refusals ?? 0) == 0) + #expect(books?.limit == nil) + } + + @Test("the origin's Retry-After stretches the wait, up to five seconds") + func retryAfterStretchesTheWait() throws { + let first = Self.streamBytes(count: 400) + let second = Self.streamBytes(count: 400) + let origin = try #require(ScriptedOrigin { index in + switch index { + case 0: return .init(status: 200, body: first) + case 1: return .init(status: 503, headers: ["Retry-After": "2"]) + case 2: return .init(status: 503, headers: ["Retry-After": "30"]) + default: return .init(status: 200, body: second) + } + }) + defer { origin.stop() } + let reader = Self.makeLiveReader(origin.url) + defer { reader.markClosed(); reader.close() } + try reader.open() + + let drained = Self.drain(reader, expectedBytes: first.count + second.count) + + #expect(drained.failure == nil) + let retryGaps = Array(Self.gaps(origin.requestTimes).dropFirst()) + #expect(retryGaps.count >= 2) + if retryGaps.count >= 2 { + #expect(retryGaps[0] >= 1.9 && retryGaps[0] <= 4.5, "Retry-After 2 gap \(retryGaps[0])s") + #expect(retryGaps[1] >= 4.9 && retryGaps[1] <= 8, "Retry-After 30 capped gap \(retryGaps[1])s") + } + } + + @Test("a stream refused for longer than the budget ends as a read error") + func refusedPastTheBudgetGivesUp() throws { + let first = Self.streamBytes(count: 400) + let origin = try #require(ScriptedOrigin { index in + index == 0 ? .init(status: 200, body: first) : .init(status: 509) + }) + defer { origin.stop() } + AVIOReader.liveRefusalRetryBudgetForTesting = 3 + defer { AVIOReader.liveRefusalRetryBudgetForTesting = nil } + let reader = Self.makeLiveReader(origin.url) + defer { reader.markClosed(); reader.close() } + try reader.open() + + let drained = Self.drain(reader, expectedBytes: first.count + 1) + + #expect(drained.bytes == first.count) + #expect(drained.failure == FFmpegErr.eio) + // Asked about three times over the budget, one a second, then given up. + let refusals = origin.requestTimes.count - 1 + #expect(refusals >= 3 && refusals <= 6, "\(refusals) refused attempts") + } + + @Test("a live open the origin refuses fails typed with its status at once") + func refusedOpenFailsTyped() throws { + let origin = try #require(ScriptedOrigin { _ in .init(status: 509) }) + defer { origin.stop() } + let reader = Self.makeLiveReader(origin.url) + defer { reader.markClosed(); reader.close() } + + let started = Date() + let failure = #expect(throws: AVIOReaderError.self) { try reader.open() } + #expect(failure == .httpStatus(509)) + #expect(Date().timeIntervalSince(started) < 5) + #expect(origin.requestTimes.count == 1) + } + + @Test("a live open answered 403 fails typed too") + func forbiddenOpenFailsTyped() throws { + let origin = try #require(ScriptedOrigin { _ in .init(status: 403) }) + defer { origin.stop() } + let reader = Self.makeLiveReader(origin.url) + defer { reader.markClosed(); reader.close() } + + let failure = #expect(throws: AVIOReaderError.self) { try reader.open() } + #expect(failure == .httpStatus(403)) + } +} From d35e5dfbccae5c7a5b5a592d391e1947208151ef Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Santos?= Date: Fri, 16 Oct 2026 04:11:07 +0200 Subject: [PATCH 2/2] test(live): judge the live refusal retries by their outcome, not by the wall clock The three loopback cases bounded the gaps between the origin's requests (a second each, Retry-After stretching them to 2 and 5 s, a count within the budget), and CI's runner took three times as long as a Mac to run the same suite, so the bounds decided what the tests reported. They now run with the reconnect backoff scaled down, as the other reader suites do, and assert what came through, how many requests it took, and that the origin's books hold no refusal; the wait itself is a pure rule (AVIOReader.liveRefusalWait) tested as one. --- Sources/AetherEngine/Demuxer/AVIOReader.swift | 8 +- .../LiveStreamRefusalRetryTests.swift | 104 ++++++++---------- 2 files changed, 51 insertions(+), 61 deletions(-) diff --git a/Sources/AetherEngine/Demuxer/AVIOReader.swift b/Sources/AetherEngine/Demuxer/AVIOReader.swift index 6016f98d..77fbb264 100644 --- a/Sources/AetherEngine/Demuxer/AVIOReader.swift +++ b/Sources/AetherEngine/Demuxer/AVIOReader.swift @@ -775,6 +775,12 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { static let liveRefusalRetryBudget: TimeInterval = 60 /// Test seam: a shorter budget, so the give-up is exercised without a minute of wall clock. nonisolated(unsafe) static var liveRefusalRetryBudgetForTesting: TimeInterval? + + /// The wait before a live stream's refused request is made again: a second, or the origin's + /// Retry-After (0 for none) when that is longer, never more than `liveRefusalMaximumRetryWait`. + static func liveRefusalWait(retryAfter: TimeInterval) -> TimeInterval { + min(max(liveRefusalRetryInterval, retryAfter), liveRefusalMaximumRetryWait) + } // Rate-limited attempts that keep the pinned redirect target before one attempt through the // source URL is spent on a fresh redirect. Three paced attempts (~7 s of ladder) ride out the // lingering-slot 509 of a connection-capped panel (#307 follow-up: the slot frees in seconds); @@ -2779,7 +2785,7 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { /// non-blocking one in `chargeFaultedRunwayRefill`, so both pace an origin identically. private func backoffDelay(streak: Int, retryAfter: TimeInterval, isRateLimited: Bool = false) -> TimeInterval { if isLive, isRateLimited { - return min(max(Self.liveRefusalRetryInterval, retryAfter), Self.liveRefusalMaximumRetryWait) * backoffScale + return Self.liveRefusalWait(retryAfter: retryAfter) * backoffScale } let expo = streak <= 0 ? 0.0 : min(Double(1 << min(streak, 4)) * 0.5, 8.0) return min(max(expo, retryAfter), 15.0) * backoffScale diff --git a/Tests/AetherEngineTests/LiveStreamRefusalRetryTests.swift b/Tests/AetherEngineTests/LiveStreamRefusalRetryTests.swift index bd27a48b..732ff6c3 100644 --- a/Tests/AetherEngineTests/LiveStreamRefusalRetryTests.swift +++ b/Tests/AetherEngineTests/LiveStreamRefusalRetryTests.swift @@ -4,9 +4,13 @@ import Foundation /// A live stream read straight off its URL (an IPTV panel's continuous MPEG-TS form of a channel) /// and refused for a while, as a connection-limited panel refuses an account's requests with 509 -/// while another device holds its last slot, keeps its session: the reader asks again every second -/// and carries on the moment it is served, the refusals arm no pacer, and a refused open is the -/// origin's answer, typed, rather than a minute of the same request. +/// while another device holds its last slot, keeps its session: the reader asks again on a short, +/// steady wait until it is served, the refusals arm no pacer, and a refused open is the origin's +/// answer, typed, rather than a minute of the same request. +/// +/// The loopback cases run with the reconnect backoff scaled down, the way the other reader suites +/// do, and are judged by what came through and how many requests it took, never by the wall clock. +/// The wait itself is a pure rule, tested as one. @Suite("Live stream reader rides out refusals", .offCooperativePool, .timeLimit(.minutes(2))) struct LiveStreamRefusalRetryTests { @@ -24,7 +28,7 @@ struct LiveStreamRefusalRetryTests { private let listener: LoopbackListener private let respond: @Sendable (_ index: Int) -> Response private let lock = NSLock() - private var times: [Date] = [] + private var count = 0 /// `respond` gets how many requests came before this one. init?(respond: @escaping @Sendable (_ index: Int) -> Response) { @@ -45,8 +49,8 @@ struct LiveStreamRefusalRetryTests { listener.stop() } - var requestTimes: [Date] { - lock.withLock { times } + var requestCount: Int { + lock.withLock { count } } var url: URL { @@ -64,8 +68,8 @@ struct LiveStreamRefusalRetryTests { if request.count > 64 * 1024 { return } } let index: Int = lock.withLock { - times.append(Date()) - return times.count - 1 + count += 1 + return count - 1 } let response = respond(index) var header = "HTTP/1.1 \(response.status) \(response.status == 200 ? "OK" : "Status")\r\n" @@ -115,20 +119,30 @@ struct LiveStreamRefusalRetryTests { return (total, nil) } - private static func gaps(_ times: [Date]) -> [TimeInterval] { - zip(times, times.dropFirst()).map { $1.timeIntervalSince($0) } + // MARK: - The wait + + @Test("the wait is a second, or the origin's Retry-After up to five seconds") + func waitIsASecondOrTheRetryAfterUpToFive() { + #expect(AVIOReader.liveRefusalWait(retryAfter: 0) == 1) + #expect(AVIOReader.liveRefusalWait(retryAfter: 0.3) == 1) + #expect(AVIOReader.liveRefusalWait(retryAfter: 2) == 2) + #expect(AVIOReader.liveRefusalWait(retryAfter: 5) == 5) + #expect(AVIOReader.liveRefusalWait(retryAfter: 30) == 5) } - // MARK: - Tests + // MARK: - The stream - @Test("a stream refused mid-session is asked for again every second until it is served") - func refusedMidStreamIsRetriedEverySecond() throws { + @Test("a stream refused mid-session is asked for again until it is served, and arms no pacer") + func refusedMidStreamIsRetriedUntilServed() throws { + AetherEngine.reconnectBackoffScaleForTesting = 0.02 + defer { AetherEngine.reconnectBackoffScaleForTesting = 1.0 } let first = Self.streamBytes(count: 400) let second = Self.streamBytes(count: 400) + let refusals = 3 let origin = try #require(ScriptedOrigin { index in switch index { case 0: return .init(status: 200, body: first) - case 1, 2, 3: return .init(status: 509) + case 1...refusals: return .init(status: 509) default: return .init(status: 200, body: second) } }) @@ -141,60 +155,28 @@ struct LiveStreamRefusalRetryTests { #expect(drained.failure == nil) #expect(drained.bytes >= first.count + second.count) - let times = origin.requestTimes - #expect(times.count >= 5) - // The three refusals are retried a second apart, give or take the request itself and a - // loaded test host; never up the 0.5, 1, 2, 4 s ladder, whose third rung alone is 4 s. - let retryGaps = Array(Self.gaps(times).dropFirst()).prefix(3) - #expect(retryGaps.count == 3) - for gap in retryGaps { - #expect(gap >= 0.9 && gap <= 3.5, "retry gap \(gap)s") - } - #expect(retryGaps.reduce(0, +) <= 7, "retry gaps \(Array(retryGaps))") + // The first body, the refusals, and the body that followed them: nothing was given up on + // and nothing asked twice over. + #expect(origin.requestCount >= refusals + 2) // The refusals armed no pacer and lowered no budget: the books hold no refusal for this // origin (a witness stamp carries no count). let books = OriginRequestBudget.shared.snapshot(for: origin.url) #expect((books?.refusals ?? 0) == 0) #expect(books?.limit == nil) - } - - @Test("the origin's Retry-After stretches the wait, up to five seconds") - func retryAfterStretchesTheWait() throws { - let first = Self.streamBytes(count: 400) - let second = Self.streamBytes(count: 400) - let origin = try #require(ScriptedOrigin { index in - switch index { - case 0: return .init(status: 200, body: first) - case 1: return .init(status: 503, headers: ["Retry-After": "2"]) - case 2: return .init(status: 503, headers: ["Retry-After": "30"]) - default: return .init(status: 200, body: second) - } - }) - defer { origin.stop() } - let reader = Self.makeLiveReader(origin.url) - defer { reader.markClosed(); reader.close() } - try reader.open() - - let drained = Self.drain(reader, expectedBytes: first.count + second.count) - - #expect(drained.failure == nil) - let retryGaps = Array(Self.gaps(origin.requestTimes).dropFirst()) - #expect(retryGaps.count >= 2) - if retryGaps.count >= 2 { - #expect(retryGaps[0] >= 1.9 && retryGaps[0] <= 4.5, "Retry-After 2 gap \(retryGaps[0])s") - #expect(retryGaps[1] >= 4.9 && retryGaps[1] <= 8, "Retry-After 30 capped gap \(retryGaps[1])s") - } + #expect(!OriginRequestBudget.shared.isPaced(origin.url)) } @Test("a stream refused for longer than the budget ends as a read error") func refusedPastTheBudgetGivesUp() throws { + AetherEngine.reconnectBackoffScaleForTesting = 0.02 + defer { AetherEngine.reconnectBackoffScaleForTesting = 1.0 } + AVIOReader.liveRefusalRetryBudgetForTesting = 1 + defer { AVIOReader.liveRefusalRetryBudgetForTesting = nil } let first = Self.streamBytes(count: 400) let origin = try #require(ScriptedOrigin { index in index == 0 ? .init(status: 200, body: first) : .init(status: 509) }) defer { origin.stop() } - AVIOReader.liveRefusalRetryBudgetForTesting = 3 - defer { AVIOReader.liveRefusalRetryBudgetForTesting = nil } let reader = Self.makeLiveReader(origin.url) defer { reader.markClosed(); reader.close() } try reader.open() @@ -203,11 +185,13 @@ struct LiveStreamRefusalRetryTests { #expect(drained.bytes == first.count) #expect(drained.failure == FFmpegErr.eio) - // Asked about three times over the budget, one a second, then given up. - let refusals = origin.requestTimes.count - 1 - #expect(refusals >= 3 && refusals <= 6, "\(refusals) refused attempts") + // Asked again more than the old count-based ladder's six would allow at this pace, and + // only given up once the budget ran out. + #expect(origin.requestCount - 1 >= 3, "\(origin.requestCount - 1) refused attempts") } + // MARK: - The open + @Test("a live open the origin refuses fails typed with its status at once") func refusedOpenFailsTyped() throws { let origin = try #require(ScriptedOrigin { _ in .init(status: 509) }) @@ -215,11 +199,10 @@ struct LiveStreamRefusalRetryTests { let reader = Self.makeLiveReader(origin.url) defer { reader.markClosed(); reader.close() } - let started = Date() let failure = #expect(throws: AVIOReaderError.self) { try reader.open() } #expect(failure == .httpStatus(509)) - #expect(Date().timeIntervalSince(started) < 5) - #expect(origin.requestTimes.count == 1) + // The one request, not the read loop's ladder on the same one. + #expect(origin.requestCount == 1) } @Test("a live open answered 403 fails typed too") @@ -231,5 +214,6 @@ struct LiveStreamRefusalRetryTests { let failure = #expect(throws: AVIOReaderError.self) { try reader.open() } #expect(failure == .httpStatus(403)) + #expect(origin.requestCount == 1) } }