diff --git a/CHANGELOG.md b/CHANGELOG.md index 5ba692b1..9ff756ab 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,19 @@ the public-API contract. ### 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`. - **Profile 7 FEL chroma residual is sampled where the base chroma sits.** With a quarter-resolution enhancement layer (1080p under 4K), the chroma kernel read the enhancement chroma at texel centres, a quarter of an EL chroma texel up and to the left of the co-sited (top-left) chroma location HDR diff --git a/Sources/AetherEngine/Demuxer/AVIOReader.swift b/Sources/AetherEngine/Demuxer/AVIOReader.swift index d2635a77..77fbb264 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,26 @@ 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? + + /// 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); @@ -973,6 +1005,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 +1317,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 +1594,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 +2434,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 +2499,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 +2667,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 +2783,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 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 } @@ -2771,7 +2832,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 +2846,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 +2864,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..732ff6c3 --- /dev/null +++ b/Tests/AetherEngineTests/LiveStreamRefusalRetryTests.swift @@ -0,0 +1,219 @@ +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 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 { + + // 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 count = 0 + + /// `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 requestCount: Int { + lock.withLock { count } + } + + 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 { + count += 1 + return 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) + } + + // 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: - The stream + + @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...refusals: 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) + // 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) + #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() } + 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 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) }) + 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(509)) + // 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") + 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)) + #expect(origin.requestCount == 1) + } +}