diff --git a/Sources/AetherEngine/Demuxer/AVIOReader.swift b/Sources/AetherEngine/Demuxer/AVIOReader.swift index 8b718d05..763963da 100644 --- a/Sources/AetherEngine/Demuxer/AVIOReader.swift +++ b/Sources/AetherEngine/Demuxer/AVIOReader.swift @@ -1049,8 +1049,23 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { if sleepNs > 0 { Thread.sleep(forTimeInterval: Double(sleepNs) / 1_000_000_000) } } + /// The caller's headers this request may carry, given where it is actually going. + /// + /// Not the same set for every target, and that is the point. `RedirectHeaderPolicy` (#126) + /// keeps a media-server credential off a cross-origin redirect target, but it only ever ran on + /// the redirect HOP. Once a session pinned that target (#12), every later request was built + /// straight against it with the full header set, so the credential the hop had just stripped + /// went to the edge on the next range anyway. Measured against a logging origin: the 302 hop + /// arrived `auth=none`, and the post-seek request to the same pinned host 13 s later carried + /// both `Authorization` and `X-Emby-Token`. One policy, applied where the request is built, so + /// a pin cannot outflank it. + private func headers(for target: URL?) -> [String: String] { + RedirectHeaderPolicy.headersToReplay( + extraHeaders: extraHeaders, originalURL: url, redirectURL: target ?? url) + } + private func applyExtraHeaders(_ request: inout URLRequest) { - for (name, value) in extraHeaders { + for (name, value) in headers(for: request.url) { request.setValue(value, forHTTPHeaderField: name) } } @@ -2530,6 +2545,13 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { adoptedWarmSize = warm.contentLength winCond.unlock() SourceContentLengthCache.store(warm.contentLength, for: url) + // The warm followed the redirect chain and knows where it ended. Pinning that target here + // is what keeps this session from resolving it a second time: a resolver 302 measured + // 800 ms on the AE#551 round 2 harness and 3.2 s on the reporter's panel, and it was paid + // per fresh connection, not once. This is the same pin a redirect records (#12), so the + // expiry ladder above handles a lease that has run out in the usual way: drop it and + // re-resolve through the source URL. Credential headers do not follow it (`headers(for:)`). + recordResolvedURL(warm.resolvedURL) EngineLog.emit( "[AVIOReader] \(label) adopted a prewarmed source: head=\(warm.head.data.count)B " + "tail=\(warm.tail?.data.count ?? 0)B of \(warm.contentLength)B; " @@ -2600,7 +2622,7 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { let delegate = TailPrefetchDelegate( expectedLength: Self.tailPrefetchBytes, - extraHeaders: extraHeaders + extraHeaders: headers(for: request.url) ) // #281 retest: one line per open, and the line the field needs. The advertised way to check // this fix was "does a bytes=-65536 request show up", which the engine never printed, so a @@ -2880,7 +2902,7 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { transfer = HeldSourceConnection( url: request.url ?? requestURLForBudget, offset: offset, - extraHeaders: extraHeaders, + extraHeaders: headers(for: request.url ?? requestURLForBudget), userAgent: nil, label: label, generation: generation, @@ -2891,7 +2913,7 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { let delegate = PersistentReadDelegate( reader: self, generation: generation, - extraHeaders: extraHeaders, + extraHeaders: headers(for: request.url), ticket: ticket, originURL: requestURLForBudget ) @@ -3355,7 +3377,7 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { let semaphore = DispatchSemaphore(value: 0) let delegate = StreamingDelegate( - extraHeaders: extraHeaders, + extraHeaders: headers(for: request.url), onResponse: { [weak self] response in // Advisory length for the sequential-origin EOF/EIO distinction; -1 (chunked / // unknown) leaves the clean-end path as the only EOF source. @@ -3702,7 +3724,7 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { timeout: Self.shortFetchSlotWaitSeconds) defer { OriginRequestBudget.shared.release(ticket) } - let delegate = ProbeDelegate(extraHeaders: extraHeaders) + let delegate = ProbeDelegate(extraHeaders: headers(for: request.url)) let task = Self.probeSession.dataTask(with: request) task.delegate = delegate @@ -3936,7 +3958,7 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable { for: slotURL, label: "\(label) fetch", timeout: Self.shortFetchSlotWaitSeconds) defer { OriginRequestBudget.shared.release(ticket) } - let delegate = ChunkFetchDelegate(extraHeaders: extraHeaders, + let delegate = ChunkFetchDelegate(extraHeaders: headers(for: request.url), bodyLimit: Self.expectedBodyBytes(for: request)) let task = Self.chunkSession.dataTask(with: request) task.delegate = delegate diff --git a/Sources/AetherEngine/Demuxer/SourcePrewarmFetcher.swift b/Sources/AetherEngine/Demuxer/SourcePrewarmFetcher.swift index 1f7f141d..b56720a0 100644 --- a/Sources/AetherEngine/Demuxer/SourcePrewarmFetcher.swift +++ b/Sources/AetherEngine/Demuxer/SourcePrewarmFetcher.swift @@ -78,19 +78,26 @@ enum SourcePrewarmFetcher { } if Task.isCancelled { return decline(url, "cancelled") } + // The tail rides the target the head just resolved, headers filtered the way a redirect + // would filter them: re-entering through the source URL would pay the same 302 a second + // time, which on the reporting origin in AE#551 round 2 was 800 ms of pure redirect. + let tailURL = head.respondedURL ?? url + let tailHeaders = RedirectHeaderPolicy.headersToReplay( + extraHeaders: extraHeaders, originalURL: url, redirectURL: tailURL) var tail: ResidentSpan? - if SourcePrewarmPlan.needsTrailingObject(head: head.body), - head.total > Int64(head.body.count) + Int64(tailBytes) { - let start = head.total - Int64(tailBytes) - if let fetched = try? await RangeFetch.run( - url: url, extraHeaders: extraHeaders, - requestedStart: start, requestedLength: tailBytes, - label: "prewarm tail", session: session), - fetched.range.start == start { - tail = ResidentSpan(start: start, data: fetched.body) - } else { - EngineLog.emit("[SourcePrewarm] trailing object not retained for \(url.lastPathComponent); " - + "the head alone is warm", category: .demux) + switch SourcePrewarmPlan.trailing(head: head.body, total: head.total) { + case .none: + break + case .suffix where head.total > Int64(head.body.count) + Int64(tailBytes): + tail = await fetchTail(url: tailURL, extraHeaders: tailHeaders, source: url, + start: head.total - Int64(tailBytes), length: tailBytes) + case .suffix: + break + case .range(let start): + let length = Int(clamping: head.total - start) + if length > 0 { + tail = await fetchTail(url: tailURL, extraHeaders: tailHeaders, source: url, + start: start, length: length) } } if Task.isCancelled { return decline(url, "cancelled") } @@ -98,19 +105,40 @@ enum SourcePrewarmFetcher { let warmed = PrewarmedSource(head: ResidentSpan(start: 0, data: head.body), tail: tail, contentLength: head.total, - requestHeaders: extraHeaders) + requestHeaders: extraHeaders, + resolvedURL: head.respondedURL) guard store.store(warmed, for: url) else { return decline(url, "\(warmed.byteCount) bytes exceed the prewarm store's cap") } EngineLog.emit( "[SourcePrewarm] warmed \(url.lastPathComponent): head=\(head.body.count)B " - + "tail=\(tail?.data.count ?? 0)B of \(head.total)B (#551)", + + "tail=\(tail?.data.count ?? 0)B at \(tail.map { String($0.start) } ?? "-") " + + "of \(head.total)B" + + (warmed.resolvedURL.map { ", resolved to host=\($0.host ?? "?")" } ?? "") + + " (#551)", category: .demux) return SourcePrewarmReport(retainedBytes: warmed.byteCount, contentLength: head.total, declined: nil) } + private static func fetchTail(url: URL, + extraHeaders: [String: String], + source: URL, + start: Int64, + length: Int) async -> ResidentSpan? { + if let fetched = try? await RangeFetch.run( + url: url, extraHeaders: extraHeaders, + requestedStart: start, requestedLength: length, + label: "prewarm tail", session: session), + fetched.range.start == start { + return ResidentSpan(start: start, data: fetched.body) + } + EngineLog.emit("[SourcePrewarm] trailing object not retained for \(source.lastPathComponent); " + + "the head alone is warm", category: .demux) + return nil + } + private static func decline(_ url: URL, _ reason: String) -> SourcePrewarmReport { EngineLog.emit("[SourcePrewarm] \(url.lastPathComponent) not warmed: \(reason) (#551)", category: .demux) @@ -135,6 +163,9 @@ enum RangeFetch { let body: Data let range: (start: Int64, end: Int64) let total: Int64 + /// The URL that actually answered, redirects followed. A warm that resolved a 302 knows the + /// target the session would otherwise resolve again (#551 round 2). + let respondedURL: URL? } static func run(url: URL, @@ -191,6 +222,7 @@ private final class RangeFetchDelegate: NSObject, URLSessionDataDelegate, @unche private var buffer = Data() private var contentRange: (start: Int64, end: Int64, total: Int64)? private var rejection: String? + private var respondedURL: URL? /// Guards the handoff between the caller's thread, which installs the handler, and the /// session's delegate queue, which produces the outcome. Either can be first. @@ -295,6 +327,9 @@ private final class RangeFetchDelegate: NSObject, URLSessionDataDelegate, @unche return } contentRange = parsed + // `http.url` is the URL that answered, redirects followed, which is the one a later load + // should start at instead of resolving the chain again (#551 round 2). + respondedURL = http.url completionHandler(.allow) } @@ -324,7 +359,8 @@ private final class RangeFetchDelegate: NSObject, URLSessionDataDelegate, @unche } return .body(RangeFetch.Result(body: buffer, range: (start: range.start, end: range.end), - total: range.total)) + total: range.total, + respondedURL: respondedURL)) } /// `bytes -/`. A `*` total is a range the origin will not size, which is diff --git a/Sources/AetherEngine/Demuxer/SourcePrewarmPlan.swift b/Sources/AetherEngine/Demuxer/SourcePrewarmPlan.swift index 693abc80..45c4852f 100644 --- a/Sources/AetherEngine/Demuxer/SourcePrewarmPlan.swift +++ b/Sources/AetherEngine/Demuxer/SourcePrewarmPlan.swift @@ -3,13 +3,61 @@ import Foundation /// #551: what a warm still has to fetch once its head is in hand. /// /// A second request is the most expensive thing a speculative path can spend against a metered -/// origin (#377), so the tail is asked for only where a cold open would actually go looking for it: -/// an MP4 whose `moov` sits behind media the warm head does not reach. Matroska is excluded on -/// measurement rather than on principle, since it reads no cues at open whether they sit at the -/// front or the back, and every other container here is left alone because the sniffer cannot say -/// anything true about it. +/// origin (#377), so the tail is asked for only where the session that opens this source will +/// actually go looking for it. Two layouts do: +/// +/// - an MP4 whose `moov` sits behind media the warm head does not reach, which is the layout #281 +/// was reported about, and +/// - a Matroska whose SeekHead names a level-1 object that lies past the warm head. Its Cues are +/// the usual one, and `HLSVideoEngine`'s cue prewarm makes that read a certainty rather than a +/// possibility: right after the open it seeks to the middle of the title so libavformat loads +/// the index, which is priced in its own comment as "1-2 byte-range reads". `read_header` adds +/// the same shape for every non-Cues object the SeekHead points at (Tags, Chapters, +/// Attachments), because `matroska_execute_seekhead` parses those on the spot and defers only +/// the Cues. +/// +/// Matroska was excluded here until AE#551 round 2, on the reading that it reads no cues at open. +/// That is true of `read_header` alone and false of the session: measured against a logging origin, +/// a warmed 63 MB MKV still spent a request on `bytes=63704296-63708626`, which is its Cues +/// element to the byte. enum SourcePrewarmPlan { + /// How many bytes past the first trailing object a warm is willing to pull. + /// + /// The MP4 case asks for a fixed 64 KB suffix; the Matroska case asks for everything from the + /// first trailing object to the end of the source, because that is the span the reads land in + /// and only the origin knows where the objects end. That span is normally tiny (4.3 KB of Cues + /// on the 7 minute fixture this was measured on, tens of KB on a feature title at one cue per + /// two seconds), so a source that wants more than this is one whose trailing objects are a + /// download rather than an index, and the warm declines instead of spending the link on it. + static let maxTrailingSpanBytes: Int64 = 1024 * 1024 + + /// What a warm should fetch after its head. + enum Trailing: Equatable { + /// Nothing: the head covers what the open reads, or the container cannot say. + case none + /// The last `SourcePrewarmFetcher.tailBytes` of the source (MP4 with `moov` at the end). + case suffix + /// Everything from `start` to the end of the source (Matroska trailing objects). + case range(start: Int64) + } + + /// The decision, given the warm head and where it ends. + /// + /// - Parameters: + /// - head: the warm's head bytes, starting at byte zero. + /// - total: the source's total size, out of the head response's `Content-Range`. + static func trailing(head: Data, total: Int64) -> Trailing { + if isMP4Family(head) { + return needsTrailingObject(head: head) ? .suffix : .none + } + if isMatroska(head), let start = firstTrailingObject(head: head, headEnd: Int64(head.count)) { + guard start < total, total - start <= maxTrailingSpanBytes else { return .none } + return .range(start: start) + } + return .none + } + /// Walks the top-level box chain inside `head` and reports whether the source's trailing object /// is still unaccounted for. False for anything that is not an MP4-family head, and false as /// soon as `moov` is found inside the warm bytes. @@ -41,6 +89,160 @@ enum SourcePrewarmPlan { return true } + // MARK: - Matroska + + private static let idEBMLHeader: UInt64 = 0x1A45_DFA3 + private static let idSegment: UInt64 = 0x1853_8067 + private static let idSeekHead: UInt64 = 0x114D_9B74 + private static let idSeek: UInt64 = 0x4DBB + private static let idSeekID: UInt64 = 0x53AB + private static let idSeekPosition: UInt64 = 0x53AC + private static let idCluster: UInt64 = 0x1F43_B675 + + static func isMatroska(_ head: Data) -> Bool { + head.count >= 4 && readID(head, at: 0)?.value == idEBMLHeader + } + + /// The lowest absolute offset at or past `headEnd` that the head's SeekHead points at. + /// + /// Clusters are excluded: they are media, and a session reads them because it is playing, not + /// because it is opening. Everything else the SeekHead names is an object `matroska_read_header` + /// or the cue prewarm goes and parses, so the span from the earliest of them to the end of the + /// source is exactly the ground those reads land in. + /// + /// Positions are relative to the Segment's DATA start, which is what `segment_start` means in + /// libavformat's matroska demuxer, so the walk has to find the Segment before it can resolve + /// one. + static func firstTrailingObject(head: Data, headEnd: Int64) -> Int64? { + guard let ebml = readElement(head, at: 0), ebml.id == idEBMLHeader, let ebmlSize = ebml.size + else { return nil } + var offset = ebml.headerLength + Int(clamping: ebmlSize) + // Void / CRC-32 between the EBML header and the Segment are legal; walk until the Segment. + while offset < head.count, let element = readElement(head, at: offset) { + if element.id == idSegment { + let dataStart = offset + element.headerLength + return firstTrailingObject(inSegment: head, dataStart: dataStart, headEnd: headEnd) + } + guard let size = element.size else { return nil } + offset += element.headerLength + Int(clamping: size) + } + return nil + } + + private static func firstTrailingObject(inSegment head: Data, + dataStart: Int, + headEnd: Int64) -> Int64? { + var offset = dataStart + var earliest: Int64? + while offset < head.count, let element = readElement(head, at: offset) { + if element.id == idSeekHead { + guard let size = element.size else { return earliest } + let start = offset + element.headerLength + let end = min(head.count, start + Int(clamping: size)) + // A SeekHead truncated by the warm head's edge still yields the entries that fit. + for position in seekPositions(head, from: start, to: end) { + let absolute = Int64(dataStart) + position + guard absolute >= headEnd else { continue } + earliest = min(earliest ?? absolute, absolute) + } + } + guard let size = element.size else { return earliest } + let next = offset + element.headerLength + Int(clamping: size) + guard next > offset else { return earliest } + offset = next + } + return earliest + } + + /// The `SeekPosition` of every `Seek` entry in the range whose `SeekID` is not a Cluster. + private static func seekPositions(_ head: Data, from: Int, to end: Int) -> [Int64] { + var positions: [Int64] = [] + var offset = from + while offset < end, let entry = readElement(head, at: offset) { + guard let entrySize = entry.size else { return positions } + let entryEnd = min(end, offset + entry.headerLength + Int(clamping: entrySize)) + if entry.id == idSeek { + var seekID: UInt64? + var position: Int64? + var inner = offset + entry.headerLength + while inner < entryEnd, let field = readElement(head, at: inner) { + guard let size = field.size else { break } + let valueStart = inner + field.headerLength + let valueEnd = min(entryEnd, valueStart + Int(clamping: size)) + if field.id == idSeekID { + seekID = beUInt(head, from: valueStart, to: valueEnd) + } else if field.id == idSeekPosition { + position = beUInt(head, from: valueStart, to: valueEnd).map { Int64(clamping: $0) } + } + inner = valueStart + Int(clamping: size) + } + if let position, position >= 0, seekID != idCluster { + positions.append(position) + } + } + offset = entryEnd > offset ? entryEnd : end + } + return positions + } + + private struct Element { + let id: UInt64 + /// Nil for the unknown-size form, which only a Segment or Cluster uses. + let size: UInt64? + let headerLength: Int + } + + private static func readElement(_ data: Data, at offset: Int) -> Element? { + guard let id = readID(data, at: offset), + let size = readSize(data, at: offset + id.length) + else { return nil } + return Element(id: id.value, size: size.value, headerLength: id.length + size.length) + } + + private static func readID(_ data: Data, at offset: Int) -> (value: UInt64, length: Int)? { + guard offset >= 0, offset < data.count else { return nil } + let base = data.startIndex + offset + let first = data[base] + guard first != 0 else { return nil } + var length = 0 + for i in 0..<4 where first & (0x80 >> UInt8(i)) != 0 { + length = i + 1 + break + } + guard length > 0, offset + length <= data.count else { return nil } + var value: UInt64 = 0 + for i in 0.. (value: UInt64?, length: Int)? { + guard offset >= 0, offset < data.count else { return nil } + let base = data.startIndex + offset + let first = data[base] + guard first != 0 else { return nil } + var length = 0 + for i in 0..<8 where first & (0x80 >> UInt8(i)) != 0 { + length = i + 1 + break + } + guard length > 0, offset + length <= data.count else { return nil } + var value = UInt64(first & (0xFF >> UInt8(length))) + for i in 1.. UInt64? { + guard from >= 0, end <= data.count, from < end, end - from <= 8 else { return nil } + let base = data.startIndex + var value: UInt64 = 0 + for i in from.. Bool { diff --git a/Sources/AetherEngine/Demuxer/SourcePrewarmStore.swift b/Sources/AetherEngine/Demuxer/SourcePrewarmStore.swift index b55ded60..98664874 100644 --- a/Sources/AetherEngine/Demuxer/SourcePrewarmStore.swift +++ b/Sources/AetherEngine/Demuxer/SourcePrewarmStore.swift @@ -20,6 +20,14 @@ struct PrewarmedSource: Sendable { /// different sizes, under one URL. A session whose headers differ therefore does not adopt /// these bytes, because nothing here could tell that they are the wrong ones. let requestHeaders: [String: String] + /// The URL that actually served the warm, redirects followed, or nil where none were. + /// + /// A resolver URL that 302s to a signed edge target is the shape half of IPTV is built out of, + /// and the warm resolves that chain. Without this the session resolves it a second time, which + /// on the reporting origin in AE#551 round 2 cost 3.2 s of redirect TTFB the warm had already + /// paid. The session adopts it as its pinned target (#12), so the existing expiry ladder is + /// what handles a lease that has since run out. + let resolvedURL: URL? var byteCount: Int { head.data.count + (tail?.data.count ?? 0) } } diff --git a/Sources/aetherctl/main.swift b/Sources/aetherctl/main.swift index ba17a777..cded0064 100644 --- a/Sources/aetherctl/main.swift +++ b/Sources/aetherctl/main.swift @@ -848,7 +848,10 @@ if first == "play" { let budget = prewarmBytes ?? AetherEngine.defaultPrewarmByteBudget let started = Date() let done = DispatchSemaphore(value: 0) - Task { + // Detached, not `Task {}`: top-level code is MainActor-isolated under the Swift 6 language + // mode, so an inheriting task enqueues on the main actor that `done.wait()` is blocking, + // and the warm never starts. That deadlock is why this flag measured nothing (#551). + Task.detached { let report = await AetherEngine.prewarm(url: target, httpHeaders: playHeaders, byteBudget: budget) let ms = Int(Date().timeIntervalSince(started) * 1000) diff --git a/Tests/AetherEngineTests/Issue551RedirectHandoffTests.swift b/Tests/AetherEngineTests/Issue551RedirectHandoffTests.swift new file mode 100644 index 00000000..c8ade591 --- /dev/null +++ b/Tests/AetherEngineTests/Issue551RedirectHandoffTests.swift @@ -0,0 +1,139 @@ +import Testing +import Foundation +@testable import AetherEngine + +/// AE#551 round 2: what a warm knows about WHERE the bytes live, and what a request may carry +/// once it goes there. +/// +/// The report: a stable resolver URL that 302s to a temporary CDN target was resolved by the warm +/// and then resolved again by the load, for ~3.2 s of redirect TTFB the warm had already paid. +/// Measured here as request counts against the source origin, which is the observable a loopback +/// origin can answer honestly. +/// +/// The second half was found while fixing the first: pinning a cross-origin target and then +/// building requests against it replays the media server's credential to that target, which is +/// exactly what `RedirectHeaderPolicy` (#126) exists to prevent on the hop. One policy now runs +/// where the request is built, so a pin cannot outflank it. +/// +/// `.serialized`: the prewarm store and the origin budget are process-wide. +@Suite("#551 the warm's resolved target, and what travels to it", .serialized) +struct Issue551RedirectHandoffTests { + + private let fileSize: Int64 = 64 * 1024 * 1024 + private let credentials = ["Authorization": "Bearer SOURCE-ONLY", + "X-Emby-Token": "SOURCE-ONLY"] + + private func read(_ reader: AVIOReader, upTo target: Int) -> Int { + let sliceCap = 128 * 1024 + let buf = UnsafeMutablePointer.allocate(capacity: sliceCap) + defer { buf.deallocate() } + 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) + } + return got + } + + // MARK: - The handoff + + @Test("the load starts at the target the warm resolved instead of resolving it again", + .timeLimit(.minutes(2))) + func loadAdoptsTheWarmsResolvedTarget() async throws { + SourcePrewarmStore.shared.clear() + let cdn = try #require(ThrottledOriginServer(totalSize: fileSize)) + defer { cdn.stop() } + let cdnPort = cdn.port + let sourceMaybe = ThrottledOriginServer( + totalSize: fileSize, + respond: { _, _, _ in .redirect(to: "http://127.0.0.1:\(cdnPort)/cdn/movie.bin") }) + let source = try #require(sourceMaybe) + defer { source.stop() } + let url = URL(string: "http://127.0.0.1:\(source.port)/movie.bin")! + + _ = await SourcePrewarmFetcher.warm(url: url, extraHeaders: [:], + byteBudget: 256 * 1024, into: .shared) + let resolvesForTheWarm = source.rangeRequestCount + #expect(resolvesForTheWarm >= 1, "the warm has to have resolved the chain itself") + + let reader = AVIOReader(url: url) + defer { reader.markClosed(); reader.close() } + try reader.open() + #expect(read(reader, upTo: 384 * 1024) == 384 * 1024) + + #expect(source.rangeRequestCount == resolvesForTheWarm, + "the session resolved the chain a second time: \(source.requestLog)") + #expect(cdn.requestLog.contains(where: { $0.start >= 256 * 1024 }), + "the data connection never reached the resolved target: \(cdn.requestLog)") + } + + @Test("a warm that never saw a redirect pins nothing", .timeLimit(.minutes(2))) + func aDirectWarmPinsNothing() async throws { + SourcePrewarmStore.shared.clear() + let server = try #require(ThrottledOriginServer(totalSize: fileSize)) + defer { server.stop() } + let url = URL(string: "http://127.0.0.1:\(server.port)/movie.bin")! + + _ = await SourcePrewarmFetcher.warm(url: url, extraHeaders: [:], + byteBudget: 256 * 1024, into: .shared) + let warmed = try #require(SourcePrewarmStore.shared.take(for: url)) + #expect(warmed.resolvedURL?.absoluteString == url.absoluteString || warmed.resolvedURL == nil, + "a direct origin must not hand back a different target: \(String(describing: warmed.resolvedURL))") + } + + // MARK: - What travels to the pinned target + + @Test("a credential header does not reach a cross-origin target the session pinned", + .timeLimit(.minutes(2))) + func credentialsStayOffThePinnedTarget() async throws { + let firstRange: Int64 = 256 * 1024 + let cdn = try #require(ThrottledOriginServer(totalSize: fileSize)) + defer { cdn.stop() } + let cdnPort = cdn.port + let sourceMaybe = ThrottledOriginServer( + totalSize: fileSize, + respond: { _, _, _ in .redirect(to: "http://127.0.0.1:\(cdnPort)/cdn/movie.bin") }) + let source = try #require(sourceMaybe) + defer { source.stop() } + + var headers = credentials + // A non-credential header a header-dependent proxy needs (#8): it must still travel, or + // the fix would have broken the thing the policy deliberately keeps. + headers["Referer"] = "https://app.example" + let reader = AVIOReader(url: URL(string: "http://127.0.0.1:\(source.port)/movie.bin")!, + extraHeaders: headers, + boundedInitialFetch: firstRange) + defer { reader.markClosed(); reader.close() } + try reader.open() + // Past the bounded first range, so at least one request is built against the PINNED target + // rather than followed onto it through a 302. + #expect(read(reader, upTo: Int(firstRange) + 128 * 1024) == Int(firstRange) + 128 * 1024) + + let atTarget = cdn.requestHeaders + #expect(!atTarget.isEmpty, "the target served nothing, so the test proves nothing") + #expect(!atTarget.contains(where: { $0["authorization"] != nil }), + "the media server's credential reached the CDN: \(atTarget)") + #expect(!atTarget.contains(where: { $0["x-emby-token"] != nil }), + "the media server's token reached the CDN: \(atTarget)") + #expect(atTarget.allSatisfy { $0["referer"] == "https://app.example" }, + "a non-credential header stopped travelling: \(atTarget)") + #expect(source.requestHeaders.allSatisfy { $0["authorization"] == "Bearer SOURCE-ONLY" }, + "the source itself must still be authenticated: \(source.requestHeaders)") + } + + @Test("the same-origin case keeps every header it always had", .timeLimit(.minutes(2))) + func sameOriginKeepsCredentials() async throws { + let server = try #require(ThrottledOriginServer(totalSize: fileSize)) + defer { server.stop() } + let reader = AVIOReader(url: URL(string: "http://127.0.0.1:\(server.port)/movie.bin")!, + extraHeaders: credentials) + defer { reader.markClosed(); reader.close() } + try reader.open() + #expect(read(reader, upTo: 128 * 1024) == 128 * 1024) + + #expect(!server.requestHeaders.isEmpty) + #expect(server.requestHeaders.allSatisfy { $0["authorization"] == "Bearer SOURCE-ONLY" }, + "an unredirected source lost its credential: \(server.requestHeaders)") + } +} diff --git a/Tests/AetherEngineTests/SourcePrewarmPlanTests.swift b/Tests/AetherEngineTests/SourcePrewarmPlanTests.swift index 78854200..93f3d7b8 100644 --- a/Tests/AetherEngineTests/SourcePrewarmPlanTests.swift +++ b/Tests/AetherEngineTests/SourcePrewarmPlanTests.swift @@ -5,10 +5,17 @@ import Foundation /// #551: whether a warm needs a second request for the source's trailing object. /// /// The question is worth asking because the answer is usually no, and a second request against a -/// metered origin is the most expensive thing a speculative path can spend. Matroska never reads -/// its cues at open whether they sit at the front or the back (measured across the #281 fixture -/// matrix), and a fast-start MP4 carries its `moov` inside the warm head already. What is left is -/// exactly the layout #281 was reported about: an MP4 whose `moov` is at the end. +/// metered origin is the most expensive thing a speculative path can spend. A fast-start MP4 +/// carries its `moov` inside the warm head already; an MP4 whose `moov` is at the end is the +/// layout #281 was reported about. +/// +/// Matroska was excluded here until AE#551 round 2 on the reading that it never reads its cues at +/// open. That holds for `matroska_read_header` and not for the session: `HLSVideoEngine` seeks to +/// the middle of the title right after the open so libavformat loads the index, and +/// `matroska_execute_seekhead` parses every non-Cues object the SeekHead points at on the spot. +/// Measured against a logging origin, a warmed 63 MB MKV still spent a request on +/// `bytes=63704296-63708626`, its Cues element to the byte. So the SeekHead is now read, and what +/// it names past the warm head is what the warm fetches. @Suite("Source prewarm tail plan (#551)") struct SourcePrewarmPlanTests { @@ -36,13 +43,156 @@ struct SourcePrewarmPlanTests { #expect(SourcePrewarmPlan.needsTrailingObject(head: head.prefix(1 << 20)) == true) } - @Test("Matroska never needs the tail, at either end") - func matroskaNeedsNoTail() { + @Test("Matroska is not an MP4 box chain, so the box walker says nothing about it") + func matroskaIsNotWalkedAsMP4() { var head = Data([0x1A, 0x45, 0xDF, 0xA3]) head.append(Data(count: 4096)) #expect(SourcePrewarmPlan.needsTrailingObject(head: head) == false) } + // MARK: - Matroska (AE#551 round 2) + + private func vint4(_ value: Int) -> Data { + Data([UInt8(0x10 | ((value >> 24) & 0x0F)), UInt8((value >> 16) & 0xFF), + UInt8((value >> 8) & 0xFF), UInt8(value & 0xFF)]) + } + + private func idBytes(_ id: UInt64) -> Data { + var bytes: [UInt8] = [] + var started = false + for shift in stride(from: 56, through: 0, by: -8) { + let byte = UInt8((id >> UInt64(shift)) & 0xFF) + if byte != 0 { started = true } + if started { bytes.append(byte) } + } + return Data(bytes) + } + + private func element(_ id: UInt64, _ payload: Data) -> Data { + var d = idBytes(id) + d.append(vint4(payload.count)) + d.append(payload) + return d + } + + private func beBytes(_ value: Int64) -> Data { + var d = Data() + withUnsafeBytes(of: value.bigEndian) { d.append(contentsOf: $0) } + return d + } + + private func seekEntry(id: UInt64, position: Int64) -> Data { + var payload = element(0x53AB, idBytes(id)) + payload.append(element(0x53AC, beBytes(position))) + return element(0x4DBB, payload) + } + + /// An MKV head: EBML header, then a Segment whose SeekHead names the given objects. The + /// positions are Segment-DATA relative, the way libavformat resolves them. + private func matroskaHead(entries: [(id: UInt64, position: Int64)], + padTo: Int) -> (head: Data, segmentDataStart: Int) { + var head = element(0x1A45_DFA3, Data(count: 31)) + var seekHeadPayload = Data() + for entry in entries { seekHeadPayload.append(seekEntry(id: entry.id, position: entry.position)) } + let seekHead = element(0x114D_9B74, seekHeadPayload) + var segmentPayload = seekHead + // Tracks, to prove the walk steps over what it does not care about. + segmentPayload.append(element(0x1654_AE6B, Data(count: 64))) + let segmentHeader = idBytes(0x1853_8067) + vint4(1 << 27) + let segmentDataStart = head.count + segmentHeader.count + head.append(segmentHeader) + head.append(segmentPayload) + if head.count < padTo { head.append(Data(count: padTo - head.count)) } + return (head, segmentDataStart) + } + + private let cuesID: UInt64 = 0x1C53_BB6B + private let tagsID: UInt64 = 0x1254_C367 + private let clusterID: UInt64 = 0x1F43_B675 + + @Test("a Matroska whose Cues sit past the warm head is warmed from the Cues to the end") + func matroskaTrailingCuesAreFetched() { + let headBytes = 64 * 1024 + let total: Int64 = 63_708_627 + // The fixture this was measured on: Cues 4331 bytes before the end. + let cuesAbsolute = total - 4331 + // Positions are Segment-DATA relative, so the fixture has to state the entry that way. + let dataStart = Int64(matroskaHead(entries: [], padTo: headBytes).segmentDataStart) + let fixture = matroskaHead(entries: [(cuesID, cuesAbsolute - dataStart)], padTo: headBytes) + #expect(SourcePrewarmPlan.trailing(head: fixture.head, total: total) + == .range(start: cuesAbsolute)) + } + + @Test("a Matroska whose objects all sit inside the warm head needs nothing more") + func matroskaWithEverythingInTheHeadNeedsNothing() { + let fixture = matroskaHead(entries: [(cuesID, 2048), (tagsID, 4096)], padTo: 64 * 1024) + #expect(SourcePrewarmPlan.trailing(head: fixture.head, total: 63_708_627) == SourcePrewarmPlan.Trailing.none) + } + + @Test("the earliest trailing object is what the span starts at") + func earliestTrailingObjectWins() { + let headBytes = 64 * 1024 + let total: Int64 = 8 * 1024 * 1024 + let probe = matroskaHead(entries: [], padTo: headBytes) + let dataStart = Int64(probe.segmentDataStart) + let tags = total - 300_000 + let cues = total - 500_000 + let fixture = matroskaHead( + entries: [(tagsID, tags - dataStart), (cuesID, cues - dataStart)], padTo: headBytes) + #expect(SourcePrewarmPlan.trailing(head: fixture.head, total: total) == .range(start: cues)) + } + + /// A Cluster is media. A session reads one because it is playing, not because it is opening, + /// and warming from the first cluster to the end of a film is a download. + @Test("a Cluster entry is not a trailing object") + func clusterEntriesAreIgnored() { + let headBytes = 64 * 1024 + let total: Int64 = 63_708_627 + let probe = matroskaHead(entries: [], padTo: headBytes) + let dataStart = Int64(probe.segmentDataStart) + let fixture = matroskaHead(entries: [(clusterID, 1_000_000 - dataStart)], padTo: headBytes) + #expect(SourcePrewarmPlan.trailing(head: fixture.head, total: total) == SourcePrewarmPlan.Trailing.none) + } + + /// The cap is what keeps this a warm rather than a transfer: an index that starts right behind + /// the head and runs for tens of megabytes is not something to fetch speculatively. + @Test("a trailing span wider than the cap is declined") + func spanBeyondTheCapIsDeclined() { + let headBytes = 64 * 1024 + let total: Int64 = 63_708_627 + let probe = matroskaHead(entries: [], padTo: headBytes) + let dataStart = Int64(probe.segmentDataStart) + let justInside = total - SourcePrewarmPlan.maxTrailingSpanBytes + let justOutside = justInside - 1 + let inside = matroskaHead(entries: [(cuesID, justInside - dataStart)], padTo: headBytes) + let outside = matroskaHead(entries: [(cuesID, justOutside - dataStart)], padTo: headBytes) + #expect(SourcePrewarmPlan.trailing(head: inside.head, total: total) == .range(start: justInside)) + #expect(SourcePrewarmPlan.trailing(head: outside.head, total: total) == SourcePrewarmPlan.Trailing.none) + } + + @Test("a Matroska head without a SeekHead is left alone") + func matroskaWithoutSeekHeadNeedsNothing() { + var head = element(0x1A45_DFA3, Data(count: 31)) + head.append(idBytes(0x1853_8067) + vint4(1 << 27)) + head.append(element(0x1654_AE6B, Data(count: 64))) + head.append(Data(count: 4096)) + #expect(SourcePrewarmPlan.trailing(head: head, total: 63_708_627) == SourcePrewarmPlan.Trailing.none) + } + + // MARK: - The MP4 answer, through the same entry point + + @Test("the MP4 layouts answer through trailing() the way they did through needsTrailingObject()") + func mp4AnswersThroughTrailing() { + var faststart = box("ftyp", size: 32) + faststart.append(box("moov", size: 4096)) + faststart.append(box("mdat", size: 65536)) + #expect(SourcePrewarmPlan.trailing(head: faststart, total: 64 << 20) == SourcePrewarmPlan.Trailing.none) + + var moovEnd = box("ftyp", size: 32) + moovEnd.append(box("mdat", size: 512 * 1024 * 1024)) + #expect(SourcePrewarmPlan.trailing(head: moovEnd.prefix(1 << 20), total: 512 << 20) == .suffix) + } + @Test("a container the sniffer does not know is left alone") func unknownNeedsNoTail() { let ts = Data([UInt8](repeating: 0x47, count: 4096)) diff --git a/Tests/AetherEngineTests/SourcePrewarmStoreTests.swift b/Tests/AetherEngineTests/SourcePrewarmStoreTests.swift index 2e81c472..c9c49b81 100644 --- a/Tests/AetherEngineTests/SourcePrewarmStoreTests.swift +++ b/Tests/AetherEngineTests/SourcePrewarmStoreTests.swift @@ -18,7 +18,8 @@ struct SourcePrewarmStoreTests { PrewarmedSource(head: ResidentSpan(start: start, data: Data(count: head)), tail: nil, contentLength: contentLength, - requestHeaders: [:]) + requestHeaders: [:], + resolvedURL: nil) } @Test("a stored source is served once and then gone") @@ -86,7 +87,8 @@ struct SourcePrewarmStoreTests { let withTail = PrewarmedSource(head: ResidentSpan(start: 0, data: Data(count: 1000)), tail: ResidentSpan(start: 9000, data: Data(count: 500)), contentLength: 9500, - requestHeaders: [:]) + requestHeaders: [:], + resolvedURL: nil) store.store(withTail, for: url("a")) #expect(store.retainedBytes == 1500) diff --git a/Tests/AetherEngineTests/ThrottledOriginServer.swift b/Tests/AetherEngineTests/ThrottledOriginServer.swift index 6fba2a9c..d2735fed 100644 --- a/Tests/AetherEngineTests/ThrottledOriginServer.swift +++ b/Tests/AetherEngineTests/ThrottledOriginServer.swift @@ -57,6 +57,9 @@ final class ThrottledOriginServer: @unchecked Sendable { private var _requestedRanges: [(start: Int64, end: Int64?)] = [] private var _requestLog: [(path: String, start: Int64, end: Int64?)] = [] private var _rangeHeaderPresent: [Bool] = [] + /// #551 round 2: the headers each request arrived with, lowercased names. A credential that + /// must not reach a target is only provably absent at the target. + private var _requestHeaders: [[String: String]] = [] private var _inflight = 0 private var _peakInflight = 0 private var _refusedForConcurrency = 0 @@ -104,6 +107,12 @@ final class ThrottledOriginServer: @unchecked Sendable { return _requestLog } + /// #551 round 2: every request's headers, in `requestLog` order, names lowercased. + var requestHeaders: [[String: String]] { + lock.lock(); defer { lock.unlock() } + return _requestHeaders + } + /// Whether each logged request carried a Range header at all. A range-less GET is logged in /// `requestLog` as (start 0, end nil), indistinguishable from `bytes=0-`; the sequential-origin /// reader's whole contract is that it never sends Range, so its tests assert on THIS. @@ -263,6 +272,7 @@ final class ThrottledOriginServer: @unchecked Sendable { _requestedRanges.append((offset, rangeEnd)) _requestLog.append((path, offset, rangeEnd)) _rangeHeaderPresent.append(hadRangeHeader) + _requestHeaders.append(Self.parseHeaders(request)) let requestIndex = _requestLog.count - 1 // #388: in flight from the moment this origin has a request to answer until its body is // written. A request parked in `readRequestHeader` on a kept-alive socket is not one. @@ -367,6 +377,18 @@ final class ThrottledOriginServer: @unchecked Sendable { return true } + private static func parseHeaders(_ request: String) -> [String: String] { + var headers: [String: String] = [:] + for line in request.components(separatedBy: "\r\n").dropFirst() { + guard let colon = line.firstIndex(of: ":") else { continue } + let name = line[.. String? { var buf = [UInt8](repeating: 0, count: 64 * 1024) var collected = Data() diff --git a/docs/api.md b/docs/api.md index 67389717..c727001c 100644 --- a/docs/api.md +++ b/docs/api.md @@ -433,7 +433,9 @@ A cold open is not free. On a non-fast-start MP4 it is two to three sequential r | `AetherEngine.defaultPrewarmByteBudget` | `8 * 1024 * 1024`. A byte budget and not a duration, because a duration needs the bitrate, which is known only after the probe this is trying to get ahead of. | | `SourcePrewarmReport` | `retainedBytes`, `contentLength`, `declined`, `isWarm`. `declined` is one sentence naming why nothing was retained, and the engine logs it either way. | -What it fetches: one ranged GET from byte zero for the budget, plus a second one for the trailing object only where the head says a cold open would go looking for it (an MP4 whose `moov` sits behind the media). Matroska is never given a second request, because it reads no cues at open whether they sit at the front or the back. +What it fetches: one ranged GET from byte zero for the budget, plus a second one for the trailing object only where the head says the session that opens this source will go looking for it. Two layouts do: an MP4 whose `moov` sits behind the media, and a Matroska whose SeekHead names a level-1 object past the warm head (its Cues, usually). The Matroska span runs from that object to the end of the source and is declined above 1 MB, because a trailing object that large is a download rather than an index. + +A warm also hands the load **where the bytes live**. A resolver URL that answers 302 with a temporary edge target is resolved once, by the warm, and the session starts at that target instead of resolving the chain again; a lease that has since run out falls back to the source URL through the same ladder a mid-session expiry uses. Credential headers (`Authorization`, `Cookie`, `X-Emby-Token` and the rest of the #126 set) never travel to a cross-origin target, whether it was reached through a redirect or pinned from a warm. Three limits are part of the contract rather than implementation detail: