fix(prewarm): a warm hands over its resolved target, its Matroska tail, and no credentials (AE#551)

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

Sources/AetherEngine/Demuxer/SourcePrewarmFetcher.swift

@@ -78,39 +78,67 @@ 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") }
 
         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 <start>-<end>/<total>`. A `*` total is a range the origin will not size, which is