fix(hls): anchor injected subtitle renditions to the origin's timestamps

CHANGELOG.md

@@ -30,8 +30,8 @@ the public-API contract.
 
 ### Fixed
 
+- External subtitles on a remote HLS source no longer appear early when the origin's media timestamps do not start at the playlist's time 0. The WebVTT renditions the engine injects (#316) carried no `X-TIMESTAMP-MAP`, so AVPlayer pinned cue time 0 to timestamp 0, and an ffmpeg MPEG-TS transcode cut with `-copyts -max_delay 5000000` showed every cue 10 s early. The engine now reads the first media timestamp of the segment the load opens on, off the load path, and serves the renditions with a matching `X-TIMESTAMP-MAP`; an origin already on its playlist timeline, one whose segment starts before its playlist slot (a transcode restarted at the previous keyframe, which AE#616 corrects), or one it cannot read, keeps the plain WebVTT.
 - Software video now presents the selected frame when seeking while paused. It decodes through preroll without starting the clock or audio, displays the winning seek's frame, and stays paused instead of retaining the previous picture until Play.
-
 - The software video decoder no longer runs more than 16 frame threads. It used one per core, and each frame thread holds back one decoded frame, so a 32-core Mac waited for 31 frames before showing the first one after a load or seek (about 3 s at 10 fps). FFmpeg also warns above 16 threads. Hosts with 16 or fewer cores keep their current thread count.
 - TrueHD Atmos rendered to APAC (`LoadOptions.objectAudioRendering`) no longer plays 42.7 ms ahead of the video. The bridge dropped the encoder's 2048 frames of priming and stamped the first content packet on the source position, but AVFoundation presents an APAC packet's audio 2048 frames before its timestamp, so every session ran early, at load and after every seek. The priming packets now stay in the stream and take the source position's timestamp.
 - Authorized native HLS uses the engine relay from the initial load, without forwarding origin credentials to the loopback asset. Optional subtitle playlist preparation shares the authorizer and has a bounded deadline across redirects and refreshes.

Sources/AetherEngine/AetherEngine+Loading.swift

@@ -535,7 +535,8 @@ extension AetherEngine {
         // rewritten master, without them the relay stands alone. A required authorization relay
         // that cannot start throws rather than returning nil.
         let playbackURL = try await prepareRemoteHLSStandIn(
-            originURL: url, options: options, expectedGeneration: bypassGeneration) ?? url
+            originURL: url, options: options, startPosition: startPosition,
+            expectedGeneration: bypassGeneration) ?? url
         // With a relay in front, the item AVPlayer fails is a loopback 502 and the refused handshake
         // happened out of its sight, so the classification has to be able to ask the side that made it.
         if let relay = remoteHLSSubtitleProxy?.server.relay {
@@ -650,6 +651,7 @@ extension AetherEngine {
     @MainActor
     private func prepareRemoteHLSStandIn(originURL: URL,
                                          options: LoadOptions,
+                                         startPosition: Double?,
                                          expectedGeneration: UInt64) async throws -> URL? {
         let requiresAuthorizationRelay = options.httpRequestAuthorization != nil
         let mayNeedRelay = EngineTLS.serverTrustEvaluator != nil
@@ -675,7 +677,9 @@ extension AetherEngine {
         let preparation = Task {
             await RemoteHLSSubtitleProxy.prepare(
                 originURL: originURL, tracks: tracks, httpHeaders: options.httpHeaders,
-                needsRelay: needsRelay, httpRequestAuthorization: options.httpRequestAuthorization)
+                needsRelay: needsRelay, httpRequestAuthorization: options.httpRequestAuthorization,
+                // Picks the segment the subtitle timestamp probe reads: the one this load opens on.
+                startPosition: startPosition)
         }
         remoteHLSPreparationTask = preparation
         let result = await withTaskCancellationHandler {

Sources/AetherEngine/Native/RemoteHLSSubtitleProvider.swift

@@ -32,6 +32,20 @@ final class RemoteHLSSubtitleProvider: HLSSegmentProvider, @unchecked Sendable {
     private let fillLock = NSLock()
     private var fillTask: Task<Void, Never>?
 
+    /// Where cue time 0 sits on the origin's media timestamps (`RemoteHLSTimestampAnchor`). Unprobed
+    /// and failed probes both serve the plain body; a running probe holds the `.vtt` answer inside the
+    /// same wait as an unfinished store, because AVPlayer keeps the first answer for the session.
+    private enum TimestampAnchor {
+        case unprobed
+        case probing
+        case resolved(Double?)
+    }
+    /// Guards `timestampAnchor` and `anchorTask`: resolved on the probe's task, read on the server's
+    /// connection thread, cancelled from the engine's teardown.
+    private let anchorLock = NSLock()
+    private var timestampAnchor = TimestampAnchor.unprobed
+    private var anchorTask: Task<Void, Never>?
+
     /// How long a `.vtt` fetch waits for its store to finish. AVPlayer fetches a whole-program VOD
     /// subtitle segment ONCE and never re-fetches it, so serving early means serving truncated for the
     /// rest of the session; the fill normally completes long before the rendition is ever selected.
@@ -78,7 +92,7 @@ final class RemoteHLSSubtitleProvider: HLSSegmentProvider, @unchecked Sendable {
     /// machinery the loopback path uses (#266), so a container holding several subtitle streams is read
     /// once and a single bad stream index cannot blank its siblings.
     func startFill() {
-        cancelFill()
+        cancelDecode()
         let jobs = Self.fillJobs(tracks: tracks, stores: stores, defaultHeaders: defaultHeaders)
         guard !jobs.isEmpty else { return }
         let task = Task.detached(priority: .utility) { [jobs] in
@@ -92,14 +106,68 @@ final class RemoteHLSSubtitleProvider: HLSSegmentProvider, @unchecked Sendable {
         fillLock.unlock()
     }
 
+    /// Teardown: stops the decode and a timestamp probe still in flight.
     func cancelFill() {
+        cancelDecode()
+        anchorLock.lock()
+        let probe = anchorTask
+        anchorTask = nil
+        anchorLock.unlock()
+        probe?.cancel()
+    }
+
+    private func cancelDecode() {
         fillLock.lock()
         let task = fillTask
         fillTask = nil
         fillLock.unlock()
         task?.cancel()
     }
 
+    /// Runs `probe` once, off the load path, and anchors every rendition this provider serves to its
+    /// answer. The proxy starts it right after the build, so the segment it reads is the one the
+    /// origin is producing for the player's first request anyway.
+    func startTimestampAnchorProbe(_ probe: @escaping @Sendable () async -> Double?) {
+        anchorLock.lock()
+        timestampAnchor = .probing
+        anchorLock.unlock()
+        let task = Task.detached(priority: .utility) { [weak self] in
+            let anchor = await probe()
+            self?.resolveTimestampAnchor(anchor)
+        }
+        anchorLock.lock()
+        anchorTask = task
+        anchorLock.unlock()
+    }
+
+    private func resolveTimestampAnchor(_ anchor: Double?) {
+        anchorLock.lock()
+        timestampAnchor = .resolved(anchor)
+        anchorLock.unlock()
+    }
+
+    /// Wait for the started probe, the `awaitFill` of the anchor: for tests, not for the session.
+    func awaitTimestampAnchor() async {
+        await currentAnchorTask()?.value
+    }
+
+    private func currentAnchorTask() -> Task<Void, Never>? {
+        anchorLock.lock()
+        defer { anchorLock.unlock() }
+        return anchorTask
+    }
+
+    /// Nil while unprobed, still probing, or when the probe had no answer: all mean the plain body.
+    private var anchorSnapshot: (pending: Bool, seconds: Double?) {
+        anchorLock.lock()
+        defer { anchorLock.unlock() }
+        switch timestampAnchor {
+        case .unprobed: return (false, nil)
+        case .probing: return (true, nil)
+        case .resolved(let seconds): return (false, seconds)
+        }
+    }
+
     /// Wait for the started fill to finish. Nothing in the session waits for it, the `.vtt` handler
     /// polls the store instead; this exists so a test can assert on a finished store without racing
     /// the wall clock. `nativeSubtitleVTT`'s wait is a budget, and a budget loses under a saturated
@@ -180,19 +248,27 @@ final class RemoteHLSSubtitleProvider: HLSSegmentProvider, @unchecked Sendable {
 
     /// Whole-program WebVTT for one sidecar. Cue times are the sidecar's own, moved by the track's
     /// declared `nativeTimelineOffsetSeconds`: this bypass has no loopback producer and therefore no
-    /// playlist shift, and AVPlayer places the cues by media timestamp, so the sidecar's absolute
-    /// seconds line up with the frames. They do NOT line up with item time where an origin's segments
-    /// start before their playlist slot; AE#616 measures that gap off them. An unfinished store is
-    /// answered `.pending`, never served: AVPlayer caches this response for the rest of the session.
+    /// playlist shift. An `X-TIMESTAMP-MAP` ties cue time 0 to where the origin's media timestamps
+    /// put it, so AVPlayer places the cues against the frames. They do NOT line up with item time
+    /// where an origin's segments start before their playlist slot; AE#616 measures that gap off them.
+    /// A probe that has not answered inside the wait serves the plain body, which is what this
+    /// returned before the anchor existed. An unfinished store is answered `.pending`, never served:
+    /// AVPlayer caches this response for the rest of the session.
     func nativeSubtitleVTT(ordinal: Int, segmentIndex: Int) -> NativeSubtitleVTTResponse {
         guard ordinal >= 0, ordinal < stores.count, segmentIndex == 0 else { return .missing }
         let store = stores[ordinal]
         let deadline = Date().addingTimeInterval(vttFillWaitSeconds)
-        while !store.isFinished, Date() < deadline {
+        while !store.isFinished || anchorSnapshot.pending, Date() < deadline {
             usleep(100_000)
         }
         guard store.isFinished else { return .pending }
+        let anchor = anchorSnapshot
+        if anchor.pending {
+            EngineLog.emit(
+                "[AetherEngine] #316: subs_\(ordinal) served before the timestamp probe answered, "
+                + "cue time 0 stays at timestamp 0", category: .engine)
+        }
         let cues = store.allCues()
-        return .ready(WebVTTBuilder.body(cues: cues))
+        return .ready(WebVTTBuilder.body(cues: cues, timestampAnchorSeconds: anchor.seconds))
     }
 }

Sources/AetherEngine/Native/RemoteHLSSubtitleProxy.swift

@@ -20,7 +20,9 @@ import Foundation
 /// one variant for the program duration and the VOD verdict), a rewrite, a socket. Anything that does
 /// not line up, and the sidecars stay overlay-only exactly as before. A load is never failed over a
 /// subtitle feature. A refusal with a relay wanted still stands the relay up on its own, because the
-/// media has nowhere else to go.
+/// media has nowhere else to go. The one read that is not a playlist, the head of a media segment for
+/// the renditions' `X-TIMESTAMP-MAP` (`RemoteHLSTimestampAnchor`), runs after the build, off the load
+/// path.
 enum RemoteHLSSubtitleProxy {
 
     /// A standing stand-in: the caller plays `masterURL` and owns the teardown.
@@ -58,15 +60,17 @@ enum RemoteHLSSubtitleProxy {
                         tracks: [RemoteHLSSubtitleProvider.Track],
                         httpHeaders: [String: String],
                         needsRelay: Bool,
-                        httpRequestAuthorization: HTTPRequestAuthorization? = nil) async -> Prepared? {
+                        httpRequestAuthorization: HTTPRequestAuthorization? = nil,
+                        startPosition: Double? = nil) async -> Prepared? {
         let needsRelay = needsRelay || httpRequestAuthorization != nil
         guard !Task.isCancelled else { return nil }
         guard !tracks.isEmpty || needsRelay else { return nil }
         if !tracks.isEmpty {
             do {
                 let prepared = try await build(
                     originURL: originURL, tracks: tracks, httpHeaders: httpHeaders,
-                    needsRelay: needsRelay, httpRequestAuthorization: httpRequestAuthorization)
+                    needsRelay: needsRelay, httpRequestAuthorization: httpRequestAuthorization,
+                    startPosition: startPosition)
                 EngineLog.emit(
                     "[AetherEngine] #316: serving \(tracks.count) external subtitle rendition(s) over a "
                     + "rewritten master at \(prepared.masterURL.absoluteString), media "
@@ -128,7 +132,8 @@ enum RemoteHLSSubtitleProxy {
                               tracks: [RemoteHLSSubtitleProvider.Track],
                               httpHeaders: [String: String],
                               needsRelay: Bool,
-                              httpRequestAuthorization: HTTPRequestAuthorization?) async throws -> Prepared {
+                              httpRequestAuthorization: HTTPRequestAuthorization?,
+                              startPosition: Double?) async throws -> Prepared {
         let session = makeSession()
         defer { session.finishTasksAndInvalidate() }
         let preflightRelay = httpRequestAuthorization.map {
@@ -143,9 +148,10 @@ enum RemoteHLSSubtitleProxy {
 
         let (body, finalURL) = try await fetchPlaylist(originURL, session: session, headers: httpHeaders, relay: preflightRelay)
         let parsed = try parse(body, at: finalURL)
-        let duration = try await programDuration(of: parsed, at: finalURL,
-                                                 session: session, credentials: credentials,
-                                                 relay: preflightRelay)
+        let variant = try await mediaPlaylist(of: parsed, body: body, at: finalURL,
+                                              session: session, credentials: credentials,
+                                              relay: preflightRelay)
+        let duration = try sumSegmentDurations(variant.media.segments)
 
         try Task.checkCancellation()
         let rewritten = try RemoteHLSMasterRewrite.rewriteDeclaringNames(
@@ -187,6 +193,16 @@ enum RemoteHLSSubtitleProxy {
         // Decode up front: the rendition is fetched the moment the host selects it, and a whole-program
         // .vtt is fetched once and never again.
         provider.startFill()
+        // Anchor the renditions to the origin's media timestamps. Started now and not awaited: the
+        // segment it reads is the one the origin is producing for the player's first request, and the
+        // `.vtt` handler holds its answer for the probe the way it holds it for the decode.
+        if let target = RemoteHLSTimestampAnchor.target(mediaPlaylistBody: variant.body, media: variant.media,
+                                                        at: variant.url, startPosition: startPosition) {
+            provider.startTimestampAnchorProbe {
+                await RemoteHLSTimestampAnchor.probe(target, credentials: credentials,
+                                                     authorization: httpRequestAuthorization)
+            }
+        }
         return Prepared(server: server, provider: provider, masterURL: masterURL,
                         renditionNames: rewritten.renditionNames)
     }
@@ -244,17 +260,19 @@ enum RemoteHLSSubtitleProxy {
         }
     }
 
-    /// Sum of the origin's own EXTINFs. A master is resolved through its first variant; the durations
-    /// are identical across variants, and one small GET buys both the length and the VOD verdict.
-    private static func programDuration(of playlist: HLSPlaylist,
-                                        at url: URL,
-                                        session: URLSession,
-                                        credentials: CredentialScope,
-                                        relay: HLSOriginRelay?) async throws -> Double {
+    /// The media playlist the program duration (the sum of its EXTINFs) and the timestamp anchor are
+    /// read from. A master is resolved through its first variant; the durations are identical across
+    /// variants, and one small GET buys the length, the VOD verdict and the segment list.
+    private static func mediaPlaylist(of playlist: HLSPlaylist,
+                                      body: String,
+                                      at url: URL,
+                                      session: URLSession,
+                                      credentials: CredentialScope,
+                                      relay: HLSOriginRelay?) async throws -> (media: HLSMediaPlaylist, body: String, url: URL) {
         switch playlist {
         case .media(let media):
             guard media.hasEndList else { throw Refusal.notVOD }
-            return try sumSegmentDurations(media.segments)
+            return (media, body, url)
         case .master(let master):
             guard let variant = master.variants.first,
                   let variantURL = HLSPlaylistParser.resolve(uri: variant.uri, against: url) else {
@@ -266,7 +284,7 @@ enum RemoteHLSSubtitleProxy {
                 throw Refusal.unusablePlaylist("variant is not a media playlist")
             }
             guard media.hasEndList else { throw Refusal.notVOD }
-            return try sumSegmentDurations(media.segments)
+            return (media, body, finalURL)
         }
     }