diff --git a/CHANGELOG.md b/CHANGELOG.md index 84f69ff13..c9d89aaff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,8 @@ the public-API contract. ### Added +- Optional session-cache byte budgets reuse existing volume and multi-session safeguards. A separate diagnostics snapshot reports effective bytes, protected playback-window overflow and typed cleanup results. Cache state is sampled off-main and old-session updates are rejected; defaults and live DVR policy remain unchanged. + - **`LiveTelemetry.objectAudio`: how many Atmos objects play, and how many overhead.** While `objectAudioRendering` plays TrueHD Atmos or DTS:X (a build with the `TrueHDObjectAudio` / `DTSXObjectAudio` traits; without them it stays nil), the 1 Hz snapshot carries an diff --git a/Sources/AetherEngine/AetherEngine+Diagnostics.swift b/Sources/AetherEngine/AetherEngine+Diagnostics.swift index e45dc97c4..270f1db6b 100644 --- a/Sources/AetherEngine/AetherEngine+Diagnostics.swift +++ b/Sources/AetherEngine/AetherEngine+Diagnostics.swift @@ -488,6 +488,25 @@ extension AetherEngine { // MARK: - Live telemetry bridge + /// Lifecycle callbacks can arrive after stop or a new load. Only cleanup may publish after + /// the owning session has detached; the token still prevents it crossing into its successor. + func acceptSessionCacheStatus(_ status: SessionCacheStatus, token: UInt64) { + guard token == sessionCacheStatusToken else { return } + guard status.cleanupResult != .notRun || nativeVideoSession != nil else { return } + diagnostics.sessionCacheStatus = status + } + + /// Read the cache ledger off-main. An old tick must never overwrite a new session or cleanup. + func refreshSessionCacheStatus() async { + guard let session = nativeVideoSession else { return } + let token = sessionCacheStatusToken + let generation = loadGeneration + let status = await BlockingWork.detached(priority: .utility) { session.sessionCacheStatus() }.value + guard !Task.isCancelled, sessionCacheStatusToken == token, loadGeneration == generation, + nativeVideoSession === session else { return } + diagnostics.sessionCacheStatus = status + } + /// Single write-through point: sampler never reaches into `EngineDiagnostics` directly. func applyLiveTelemetry(_ snapshot: LiveTelemetry) { diagnostics.liveTelemetry = snapshot diff --git a/Sources/AetherEngine/AetherEngine+Loading.swift b/Sources/AetherEngine/AetherEngine+Loading.swift index 660e7620b..5a63496ba 100644 --- a/Sources/AetherEngine/AetherEngine+Loading.swift +++ b/Sources/AetherEngine/AetherEngine+Loading.swift @@ -373,6 +373,9 @@ extension AetherEngine { // stale caller cannot adopt its successor's generation. if let generation { try checkLoadCurrent(generation) } playbackBackend = .native + sessionCacheStatusToken &+= 1 + diagnostics.sessionCacheStatus = .nativeRemoteHLS( + requestedBudgetBytes: options.sessionCacheByteBudget.map { max(0, $0) }) // #168 follow-up: detect a superseding load()/stop() between the carriage verdict and the reroute. let bypassGeneration = generation ?? loadGeneration @@ -888,8 +891,17 @@ extension AetherEngine { // the host asked for; without it the flag lasts exactly as long as the pre-opened demuxer. heldSourceConnection: loadedOptions.heldSourceConnection, declaredDurationSeconds: loadedOptions.declaredDurationSeconds, - forwardBufferSegments: loadedOptions.forwardBufferSegments + forwardBufferSegments: loadedOptions.forwardBufferSegments, + sessionCacheByteBudget: loadedOptions.sessionCacheByteBudget ) + sessionCacheStatusToken &+= 1 + let cacheStatusToken = sessionCacheStatusToken + session.onSessionCacheStatusChanged = { [weak self] status in + Task { @MainActor [weak self] in + self?.acceptSessionCacheStatus(status, token: cacheStatusToken) + } + } + // AE#792: the session's live header store, shared with the playback demuxer it was handed, so // `updateRequestHeaders(_:)` reaches that reader and every reopen the producer makes. Set before start(). session.requestHeaders = sessionRequestHeaders @@ -1324,7 +1336,8 @@ extension AetherEngine { // session.start() opens its own Demuxer + prewarm seek (~1-3 s on slow CDN); detach so @MainActor doesn't block. var playbackURL = try await BlockingWork.detached(priority: .userInitiated) { [session] in - try session.start() + do { return try session.start() } + catch { session.stop(reason: .loadFailed); throw error } }.value // AirPlay (#86): while external playback is active, serve the loopback over the device's LAN IP so // the receiver reaches the engine-processed stream (DV/Atmos/subtitles preserved). An HDR/DV master @@ -1334,7 +1347,7 @@ extension AetherEngine { playbackURL = served.url // Superseded while starting: stop and unwind before touching shared state. if loadGeneration != generation { - session.stop() + session.stop(reason: .sourceChanged) try checkLoadCurrent(generation) } self.nativeVideoSession = session @@ -1961,6 +1974,9 @@ extension AetherEngine { generation: UInt64, elType: String? = nil ) async throws { + try checkLoadCurrent(generation) + sessionCacheStatusToken &+= 1 + diagnostics.sessionCacheStatus = .inactive let deinterlaceMode = loadedOptions.deinterlaceMode let waitStarted = DispatchTime.now() if let outcome = await DeinterlaceHardwareWarmup.shared.waitIfNeeded( @@ -2123,6 +2139,7 @@ extension AetherEngine { } if loadGeneration == generation { recordStartupCheckpoint(.sessionConstructed) } // #361 let forwardBufferSegments = loadedOptions.forwardBufferSegments + let sessionCacheByteBudget = loadedOptions.sessionCacheByteBudget let dvrRetention = loadedOptions.softwareDVRRetention try await BlockingWork.detached(priority: .userInitiated) { [host, preopenedDemuxer, url, sourceHTTPHeaders, requestHeaders = sessionRequestHeaders, isLive, dvrWindowSeconds, probesize, maxAnalyzeDuration, sequentialOrigin, heldSourceConnection, declaredDuration, networkPhaseSink] in @@ -2141,7 +2158,8 @@ extension AetherEngine { isLive: isLive, dvrWindowSeconds: dvrWindowSeconds, dvrRetention: dvrRetention, - forwardBufferSegments: forwardBufferSegments + forwardBufferSegments: forwardBufferSegments, + sessionCacheByteBudget: sessionCacheByteBudget ) }.value // Superseded: stop idempotently to tear down the demuxer the detached closure opened, then unwind. @@ -2476,6 +2494,8 @@ extension AetherEngine { stopInternal(resetDisplayCriteria: false, keepNativeHost: !targetSoftwarePath && !mediaServicesWereReset, keepCustomReader: true, keepCurrentItem: keepAudioSwitchItem) + sessionCacheStatusToken &+= 1 + diagnostics.sessionCacheStatus = .inactive pendingInPlaceItemHandover = keepAudioSwitchItem if mediaServicesWereReset { dropAudioPlayerHostAfterMediaServicesReset() } EngineLog.emit("[AetherEngine] reload: stopInternal done (\(elapsedMs(since: reloadStart))ms)", category: .engine) diff --git a/Sources/AetherEngine/AetherEngine+ReloadWithOptions.swift b/Sources/AetherEngine/AetherEngine+ReloadWithOptions.swift index aaa5ba4e5..73fcfe97c 100644 --- a/Sources/AetherEngine/AetherEngine+ReloadWithOptions.swift +++ b/Sources/AetherEngine/AetherEngine+ReloadWithOptions.swift @@ -454,7 +454,7 @@ enum SessionOptionCorrection { "preserveASSMarkup", "prepareNativeSubtitles", "eagerNativeSubtitleReaders", "serveIFramePlaylist", "confirmAtmos", "nativeSubtitlePreferredLanguages", "sequentialOrigin", "maxConcurrentSourceRequests", "heldSourceConnection", "declaredDurationSeconds", "probesize", "maxAnalyzeDuration", "preferredAudioLanguages", - "preferredSubtitleLanguages", "externalSubtitles", "forwardBufferSegments", "progressiveSegmentDelivery", "autoplay", + "preferredSubtitleLanguages", "externalSubtitles", "forwardBufferSegments", "sessionCacheByteBudget", "progressiveSegmentDelivery", "autoplay", "audioDelaySeconds", "teletextPage", "deinterlaceMode", "deinterlaceFieldRate", "preferredDecodePath", "escalatesToSoftwarePath", "sharedOutputRole", "isLiveRejoin", "subtitleSessionCarryover", ] diff --git a/Sources/AetherEngine/AetherEngine.swift b/Sources/AetherEngine/AetherEngine.swift index bdbf1a191..94a0888ac 100644 --- a/Sources/AetherEngine/AetherEngine.swift +++ b/Sources/AetherEngine/AetherEngine.swift @@ -2123,6 +2123,9 @@ public final class AetherEngine: ObservableObject { /// dismissal. A superseded load throws CancellationError at the first checkpoint. var loadGeneration: UInt64 = 0 + /// Separates cache lifecycle callbacks from superseding sessions. + var sessionCacheStatusToken: UInt64 = 0 + /// Distinct across engine instances and refreshed whenever teardown invalidates preview work. public private(set) var scrubPreviewSourceGeneration = UUID() @@ -4163,6 +4166,8 @@ public final class AetherEngine: ObservableObject { // (audio-only fast path, suppressed hosts); a load() that throws before routing leaves it for stop(). claimSoftwarePathTakeover() // AE#629 stopInternal(resetDisplayCriteria: false, keepNativeHost: preserveNativeHost, keepCurrentItem: handOverInPlace) + sessionCacheStatusToken &+= 1 + diagnostics.sessionCacheStatus = .inactive if mediaServicesWereReset { dropAudioPlayerHostAfterMediaServicesReset() } // #35/#93: a genuinely new item has not rendered yet; re-arm the cold-startup wedge suspension. // Scrub/seek/producer-restart never route through load(), so mid-stream #93 detection stays armed. @@ -6593,7 +6598,8 @@ public final class AetherEngine: ObservableObject { releaseHeldPicture(reason: "stop") nextLoadRequestsInPlaceItemHandover = false stopInternal(resetDisplayCriteria: resetDisplayCriteria, - finalTeardown: finalTeardown ?? resetDisplayCriteria) + finalTeardown: finalTeardown ?? resetDisplayCriteria, + cacheCleanupReason: .sessionStopped) state = .idle // #361: the sequence ends here without ever reaching its last checkpoint, which is the whole // point: nothing about a stop should read as a finished startup. Cleared in stop() and not in @@ -7587,7 +7593,8 @@ public final class AetherEngine: ObservableObject { /// never settles and burns the full settle timeout (~12 s of /// black-screen latency per audio switch on the old fixed 5 s /// poll; capped at ~2 s since #117, but still worth skipping). - func stopInternal(resetDisplayCriteria: Bool = true, keepNativeHost: Bool = false, keepCustomReader: Bool = false, keepCurrentItem: Bool = false, finalTeardown: Bool = false) { + func stopInternal(resetDisplayCriteria: Bool = true, keepNativeHost: Bool = false, keepCustomReader: Bool = false, keepCurrentItem: Bool = false, finalTeardown: Bool = false, + cacheCleanupReason: SessionCacheCleanupReason = .sourceChanged) { // AE#560: a recording never outlives its session. This covers stop(), a new load() and // every reload, all of which pass through here before the demuxer goes away. endRecordingIfRunning(reason: .sessionEnded) @@ -7664,7 +7671,10 @@ public final class AetherEngine: ObservableObject { // host has already retired for the next item. Only the session's slot is cleared; the engine // keeps the host's observer and re-arms the next session with it in load(). nativeVideoSession?.setNativeVideoFrameTimeObserver(nil) - nativeVideoSession?.stop() + if diagnostics.sessionCacheStatus.route == .nativeRemoteHLS { + diagnostics.sessionCacheStatus = .inactive + } + nativeVideoSession?.stop(reason: cacheCleanupReason) nativeVideoSession = nil nativeSubtitleRenditionsServed = false dolbyVisionConversion = nil diff --git a/Sources/AetherEngine/Diagnostics/EngineDiagnostics.swift b/Sources/AetherEngine/Diagnostics/EngineDiagnostics.swift index 80cd231ae..03e5104fd 100644 --- a/Sources/AetherEngine/Diagnostics/EngineDiagnostics.swift +++ b/Sources/AetherEngine/Diagnostics/EngineDiagnostics.swift @@ -6,6 +6,9 @@ import Combine /// Stats overlays observe this object; everything else observes the engine and is unaffected by telemetry samples. @MainActor public final class EngineDiagnostics: ObservableObject { + /// Session-only fMP4 cache state, sampled off-main and cleared on a new source. + @Published public internal(set) var sessionCacheStatus: SessionCacheStatus = .inactive + /// 1 Hz snapshot while playing/paused; nil while idle. Cleared in stopInternal so sessions don't inherit stale numbers. @Published public internal(set) var liveTelemetry: LiveTelemetry? diff --git a/Sources/AetherEngine/Diagnostics/LiveTelemetrySampler.swift b/Sources/AetherEngine/Diagnostics/LiveTelemetrySampler.swift index 6c24877a2..027e372c4 100644 --- a/Sources/AetherEngine/Diagnostics/LiveTelemetrySampler.swift +++ b/Sources/AetherEngine/Diagnostics/LiveTelemetrySampler.swift @@ -456,6 +456,7 @@ final class LiveTelemetrySampler { rssMb: AetherEngine.residentMemoryMB() ) engine.applyLiveTelemetry(snapshot) + await engine.refreshSessionCacheStatus() } /// One line per tick on the native path (#93 post-recovery lag diagnosis). Discriminates diff --git a/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift b/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift index a388656f6..3f5289bc5 100644 --- a/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift +++ b/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift @@ -45,6 +45,15 @@ final class SoftwarePlaybackHost { }() + /// Zero still needs a working spool; every choice retains the volume and shared-claim clamp. + nonisolated static func packetCacheBudgetBytes(requestedBytes: Int?, availableBytes: Int64?, + capRelaxed: Bool) -> Int { + let budget = HLSVideoEngine.resolveSessionCacheBudget( + requestedBytes: requestedBytes, volumeAvailableBytes: availableBytes, capRelaxed: capRelaxed) + return min(budget.volumeSafetyLimitBytes, + budget.baseEffectiveBytes > 0 ? budget.baseEffectiveBytes : 32 << 20) + } + // MARK: - Published state (mirrors NativeAVPlayerHost surface) /// Frames enqueued on the display layer; read by LiveTelemetrySampler at 1 Hz for observed FPS. Lock-guarded (demux thread writes, main-actor reads). @@ -998,7 +1007,8 @@ final class SoftwarePlaybackHost { isLive: Bool = false, dvrWindowSeconds: Double? = nil, dvrRetention: SoftwareDVRRetentionOptions? = nil, - forwardBufferSegments: Int? = nil + forwardBufferSegments: Int? = nil, + sessionCacheByteBudget: Int? = nil ) async throws { self.demuxer = dem self.duration = dem.duration @@ -1334,7 +1344,8 @@ final class SoftwarePlaybackHost { .volumeAvailableCapacity.map(Int64.init) let capRelaxed = HLSVideoEngine.retentionCapRelaxed(forwardWindowSegments: segments) let diskClaim = RetentionClaims.shared.claim(volumeAvailableBytes: available) { - HLSVideoEngine.sessionRetentionBudgetBytes(volumeAvailableBytes: $0, capRelaxed: capRelaxed) + Self.packetCacheBudgetBytes(requestedBytes: sessionCacheByteBudget, + availableBytes: $0, capRelaxed: capRelaxed) } bytes = diskClaim.bytes guard bytes > 0 else { return nil } diff --git a/Sources/AetherEngine/PlayerState.swift b/Sources/AetherEngine/PlayerState.swift index d2e18e7df..b35c820f7 100644 --- a/Sources/AetherEngine/PlayerState.swift +++ b/Sources/AetherEngine/PlayerState.swift @@ -882,6 +882,9 @@ public struct LoadOptions: Sendable, Equatable { /// remote server directly. public var forwardBufferSegments: Int? + /// Optional temporary session-cache budget. Nil retains automatic sizing; zero keeps only the working window. + public var sessionCacheByteBudget: Int? = nil + /// Serve each VOD loopback segment while it is being written instead of after its cut. The /// muxer flushes a fragment about every half second and the loopback server sends each one as /// it lands, so AVPlayer can show and start on the first fragments of a segment rather than @@ -1023,6 +1026,7 @@ public struct LoadOptions: Sendable, Equatable { preferredSubtitleLanguages: [String] = [], externalSubtitles: [ExternalSubtitleTrack] = [], forwardBufferSegments: Int? = nil, + sessionCacheByteBudget: Int? = nil, progressiveSegmentDelivery: Bool = false, autoplay: Bool = true, teletextPage: Int? = nil, @@ -1077,6 +1081,7 @@ public struct LoadOptions: Sendable, Equatable { self.preferredSubtitleLanguages = preferredSubtitleLanguages self.externalSubtitles = externalSubtitles self.forwardBufferSegments = forwardBufferSegments + self.sessionCacheByteBudget = sessionCacheByteBudget self.progressiveSegmentDelivery = progressiveSegmentDelivery self.autoplay = autoplay self.teletextPage = teletextPage diff --git a/Sources/AetherEngine/SessionCacheStatus.swift b/Sources/AetherEngine/SessionCacheStatus.swift new file mode 100644 index 000000000..a7db59e32 --- /dev/null +++ b/Sources/AetherEngine/SessionCacheStatus.swift @@ -0,0 +1,125 @@ +import Foundation + +/// Route that owns buffering for the current session. Only `loopbackFMP4` uses Aether's +/// session-scoped `SegmentCache`; native remote HLS remains AVPlayer/server managed. +public enum SessionCacheRoute: String, Sendable, Equatable { + case none + case loopbackFMP4 = "loopback_fmp4" + case nativeRemoteHLS = "native_remote_hls" +} + +/// Whether the active route can apply Aether's session-cache byte policy. +public enum SessionCacheCapability: String, Sendable, Equatable { + case inactive + case active + case unsupported +} + +/// Why a session cache directory was closed. Values describe lifecycle, not persistence: every +/// successful close deletes the entire UUID directory and no later session reuses it. +public enum SessionCacheCleanupReason: String, Sendable, Equatable { + case sessionStopped = "session_stopped" + case sourceChanged = "source_changed" + case loadFailed = "load_failed" +} + +public enum SessionCacheCleanupResult: String, Sendable, Equatable { + case notRun = "not_run" + case succeeded + case failed +} + +/// Typed, credential-free disk failure for the current session cache. +public enum SessionCacheFailureReason: String, Sendable, Equatable { + case sessionDirectoryCreationFailed = "session_directory_creation_failed" + case writeFailed = "write_failed" + case adoptFailed = "adopt_failed" + case staleSweepFailed = "stale_sweep_failed" + case cleanupFailed = "cleanup_failed" +} + +/// Route-aware snapshot of Aether's ephemeral fMP4 segment cache. +/// +/// `baseEffectiveBudgetBytes` is the caller request after the tmp-volume safety clamp. +/// `playbackSafeFloorBytes` is the actual resident hard window that cannot be evicted without +/// risking AVPlayer starvation. `effectiveBudgetBytes` is the larger of those two values, so a +/// high-bitrate hard window is reported honestly rather than silently violating the requested cap. +public struct SessionCacheStatus: Sendable, Equatable { + public let route: SessionCacheRoute + public let capability: SessionCacheCapability + public let requestedBudgetBytes: Int? + public let volumeSafetyLimitBytes: Int? + public let baseEffectiveBudgetBytes: Int + public let playbackSafeFloorBytes: Int + public let effectiveBudgetBytes: Int + public let currentResidentBytes: Int + public let currentForwardBytes: Int + public let forwardWindowSegments: Int + public let backwardWindowSegments: Int + public let producerParked: Bool + public let evictionCount: Int + public let hardWindowFloorExceeded: Bool + public let cleanupReason: SessionCacheCleanupReason? + public let cleanupResult: SessionCacheCleanupResult + public let lastFailure: SessionCacheFailureReason? + + public init( + route: SessionCacheRoute, + capability: SessionCacheCapability, + requestedBudgetBytes: Int?, + volumeSafetyLimitBytes: Int?, + baseEffectiveBudgetBytes: Int, + playbackSafeFloorBytes: Int, + effectiveBudgetBytes: Int, + currentResidentBytes: Int, + currentForwardBytes: Int, + forwardWindowSegments: Int, + backwardWindowSegments: Int, + producerParked: Bool, + evictionCount: Int, + hardWindowFloorExceeded: Bool, + cleanupReason: SessionCacheCleanupReason?, + cleanupResult: SessionCacheCleanupResult, + lastFailure: SessionCacheFailureReason? + ) { + self.route = route + self.capability = capability + self.requestedBudgetBytes = requestedBudgetBytes + self.volumeSafetyLimitBytes = volumeSafetyLimitBytes + self.baseEffectiveBudgetBytes = baseEffectiveBudgetBytes + self.playbackSafeFloorBytes = playbackSafeFloorBytes + self.effectiveBudgetBytes = effectiveBudgetBytes + self.currentResidentBytes = currentResidentBytes + self.currentForwardBytes = currentForwardBytes + self.forwardWindowSegments = forwardWindowSegments + self.backwardWindowSegments = backwardWindowSegments + self.producerParked = producerParked + self.evictionCount = evictionCount + self.hardWindowFloorExceeded = hardWindowFloorExceeded + self.cleanupReason = cleanupReason + self.cleanupResult = cleanupResult + self.lastFailure = lastFailure + } + + public static let inactive = SessionCacheStatus( + route: .none, capability: .inactive, requestedBudgetBytes: nil, + volumeSafetyLimitBytes: nil, baseEffectiveBudgetBytes: 0, + playbackSafeFloorBytes: 0, effectiveBudgetBytes: 0, currentResidentBytes: 0, + currentForwardBytes: 0, forwardWindowSegments: 0, backwardWindowSegments: 0, + producerParked: false, evictionCount: 0, hardWindowFloorExceeded: false, + cleanupReason: nil, cleanupResult: .notRun, lastFailure: nil + ) + + public static func nativeRemoteHLS(requestedBudgetBytes: Int?) -> SessionCacheStatus { + SessionCacheStatus( + route: .nativeRemoteHLS, capability: .unsupported, + requestedBudgetBytes: requestedBudgetBytes, volumeSafetyLimitBytes: nil, + baseEffectiveBudgetBytes: 0, playbackSafeFloorBytes: 0, + effectiveBudgetBytes: 0, currentResidentBytes: 0, currentForwardBytes: 0, + forwardWindowSegments: 0, backwardWindowSegments: 0, producerParked: false, + evictionCount: 0, hardWindowFloorExceeded: false, cleanupReason: nil, + cleanupResult: .notRun, lastFailure: nil + ) + } +} + diff --git a/Sources/AetherEngine/Video/HLSVideoEngine.swift b/Sources/AetherEngine/Video/HLSVideoEngine.swift index 4314ff275..c36fb8a80 100644 --- a/Sources/AetherEngine/Video/HLSVideoEngine.swift +++ b/Sources/AetherEngine/Video/HLSVideoEngine.swift @@ -981,11 +981,52 @@ public final class HLSVideoEngine: @unchecked Sendable { /// `capRelaxed` (#207) drops the 2 GiB default for a host that explicitly asked to pre-buffer more /// than the historical window could hold; the quarter-of-free-space clamp, which is what actually /// protects the volume, always applies. Unknown capacity keeps the conservative cap either way. + struct SessionCacheBudgetResolution: Sendable, Equatable { + let requestedBytes: Int? + let volumeSafetyLimitBytes: Int + let baseEffectiveBytes: Int + } + + /// Resolve a caller-exact session budget without weakening the existing tmp-volume guard. + /// nil keeps the historical automatic policy. Unknown capacity stays conservatively bounded + /// at 2 GiB even for an explicit aggressive-prefetch request. + static func resolveSessionCacheBudget( + requestedBytes: Int?, + volumeAvailableBytes: Int64?, + capRelaxed: Bool = false + ) -> SessionCacheBudgetResolution { + let conservativeCap = 2 << 30 + let quarterOfFree = volumeAvailableBytes.map { available -> Int in + let quarter = max(Int64(0), available / 4) + return Int(min(quarter, Int64(Int.max))) + } + let safetyLimit = quarterOfFree ?? conservativeCap + + if let requestedBytes { + let normalized = max(0, requestedBytes) + return SessionCacheBudgetResolution( + requestedBytes: normalized, + volumeSafetyLimitBytes: safetyLimit, + baseEffectiveBytes: min(normalized, safetyLimit) + ) + } + + let effective = capRelaxed + ? safetyLimit + : min(conservativeCap, safetyLimit) + return SessionCacheBudgetResolution( + requestedBytes: nil, + volumeSafetyLimitBytes: safetyLimit, + baseEffectiveBytes: effective + ) + } + static func sessionRetentionBudgetBytes(volumeAvailableBytes: Int64?, capRelaxed: Bool = false) -> Int { - let cap = 2 << 30 - guard let available = volumeAvailableBytes else { return cap } - let quarterOfFree = max(0, Int(available / 4)) - return capRelaxed ? quarterOfFree : min(cap, quarterOfFree) + resolveSessionCacheBudget( + requestedBytes: nil, + volumeAvailableBytes: volumeAvailableBytes, + capRelaxed: capRelaxed + ).baseEffectiveBytes } /// Largest forward window the default 2 GiB retention budget covers by construction @@ -1048,7 +1089,8 @@ public final class HLSVideoEngine: @unchecked Sendable { sequentialOrigin: Bool = false, heldSourceConnection: Bool = false, declaredDurationSeconds: Double? = nil, - forwardBufferSegments: Int? = nil + forwardBufferSegments: Int? = nil, + sessionCacheByteBudget: Int? = nil ) { self.sourceURL = url self.requestHeaders = RequestHeaderStore(sourceHTTPHeaders) @@ -1112,6 +1154,7 @@ public final class HLSVideoEngine: @unchecked Sendable { self.customSourceReopenFactory = customSourceReopenFactory self.companionAudioReader = companionAudioReader self.forwardWindowSegments = Self.clampedForwardWindow(forwardBufferSegments) + self.requestedSessionCacheBudgetBytes = sessionCacheByteBudget.map { max(0, $0) } } /// Session forward-buffer window in segments. Drives BOTH the producer's race-ahead @@ -1125,6 +1168,12 @@ public final class HLSVideoEngine: @unchecked Sendable { /// playlist advertises stays resident; the producer-side prefetch park it also feeds is /// VOD-only (`advanceMuxer`), so live cannot park on it. private var retentionBudgetBytes: Int = 0 + private let requestedSessionCacheBudgetBytes: Int? + private var sessionCacheVolumeSafetyLimitBytes: Int? + + /// Lifecycle snapshots, including async close completion, are delivered here. The host may + /// sample `sessionCacheStatus()` for current bytes; this callback is primarily for cleanup. + public var onSessionCacheStatusChanged: (@Sendable (SessionCacheStatus) -> Void)? /// #687: this session's entry in the process-wide ledger, released on `stop()`. private var retentionClaim: RetentionClaims.Claim? @@ -1618,12 +1667,18 @@ public final class HLSVideoEngine: @unchecked Sendable { .volumeAvailableCapacityForImportantUsage #endif let capRelaxed = Self.retentionCapRelaxed(forwardWindowSegments: forwardWindowSegments) - // #687: sized from what the other running sessions leave, not from the raw free space. - let claim = RetentionClaims.shared.claim(volumeAvailableBytes: availableBytes) { - Self.sessionRetentionBudgetBytes(volumeAvailableBytes: $0, capRelaxed: capRelaxed) + // Size the custom byte ceiling against capacity left by other running sessions. + var resolvedBudget: SessionCacheBudgetResolution? + let claim = RetentionClaims.shared.claim(volumeAvailableBytes: availableBytes) { remaining in + let budget = Self.resolveSessionCacheBudget( + requestedBytes: isLiveSession ? nil : requestedSessionCacheBudgetBytes, + volumeAvailableBytes: remaining, + capRelaxed: capRelaxed) + resolvedBudget = budget + return budget.baseEffectiveBytes } + let budget = resolvedBudget! let retentionBudget = claim.bytes - self.retentionBudgetBytes = retentionBudget let segmentCache = SegmentCache( forwardWindow: forwardWindowSegments, retentionBudgetBytes: retentionBudget, @@ -1631,10 +1686,18 @@ public final class HLSVideoEngine: @unchecked Sendable { onResidentSetChanged: { [weak self] in self?.noteResidentSetChanged() } ) claim.track { [weak segmentCache] in segmentCache?.totalBytes ?? 0 } + restartLock.lock() + self.retentionBudgetBytes = retentionBudget + self.sessionCacheVolumeSafetyLimitBytes = budget.volumeSafetyLimitBytes self.retentionClaim = claim self.cache = segmentCache + restartLock.unlock() + onSessionCacheStatusChanged?(sessionCacheStatus()) + let requestedBudgetDescription = budget.requestedBytes + .map { "\($0 / (1 << 20)) MiB" } ?? "automatic" EngineLog.emit( - "[HLSVideoEngine] segment retention budget: \(retentionBudget / (1 << 20)) MiB " + "[HLSVideoEngine] segment retention budget: requested=\(requestedBudgetDescription) " + + "effective=\(retentionBudget / (1 << 20)) MiB " + "(volumeAvailable=\(availableBytes.map { "\($0 / (1 << 20)) MiB" } ?? "unknown"), " + "forwardWindow=\(forwardWindowSegments) seg" + (capRelaxed ? ", opt-in prefetch: default cap relaxed" : "") @@ -2280,7 +2343,7 @@ public final class HLSVideoEngine: @unchecked Sendable { ? srv.playlistURL : srv.mediaPlaylistURL guard let url = resolvedURL else { - stop() + stop(reason: .loadFailed) throw HLSVideoEngineError.openFailed(reason: "server URL not ready") } self.servingMasterPlaylist = useMasterPlaylist @@ -2637,6 +2700,38 @@ public final class HLSVideoEngine: @unchecked Sendable { var objectAudioActivity: ObjectAudioActivity? { subsystemSnapshot().audioBridge?.objectAudioActivity } var lastAVGapMs: Double { subsystemSnapshot().producer?.lastAVGapMs ?? 0 } + /// Current route-aware cache contract. Byte fields come from the cache ledger, not segment-count + /// estimates. The close-completion snapshot is also delivered through + /// `onSessionCacheStatusChanged` because the cache object is intentionally detached at stop. + public func sessionCacheStatus() -> SessionCacheStatus { + restartLock.lock() + let cache = self.cache + let budget = retentionBudgetBytes + let safetyLimit = sessionCacheVolumeSafetyLimitBytes + restartLock.unlock() + if let cache { + return cache.status( + route: .loopbackFMP4, + requestedBudgetBytes: requestedSessionCacheBudgetBytes.map { max(0, $0) }, + baseEffectiveBudgetBytes: budget, + volumeSafetyLimitBytes: safetyLimit, + capability: isLiveSession ? .unsupported : .active + ) + } + return SessionCacheStatus( + route: .loopbackFMP4, capability: .inactive, + requestedBudgetBytes: requestedSessionCacheBudgetBytes.map { max(0, $0) }, + volumeSafetyLimitBytes: safetyLimit, + baseEffectiveBudgetBytes: budget, playbackSafeFloorBytes: 0, + effectiveBudgetBytes: budget, currentResidentBytes: 0, + currentForwardBytes: 0, forwardWindowSegments: forwardWindowSegments, + backwardWindowSegments: SegmentCache.defaultBackwardWindow, + producerParked: false, evictionCount: 0, + hardWindowFloorExceeded: false, cleanupReason: nil, cleanupResult: .notRun, + lastFailure: nil + ) + } + public func diagnosticStats() -> DiagnosticStats { let subs = subsystemSnapshot() let abLive = subs.audioBridge?.liveBytes @@ -2733,7 +2828,10 @@ public final class HLSVideoEngine: @unchecked Sendable { initData: initData, segmentURL: seg.fileURL) } - public func stop() { + /// Preserve the original zero-argument function type for hosts storing a stop callback. + public func stop() { stop(reason: .sessionStopped) } + + public func stop(reason: SessionCacheCleanupReason) { cancelResidentRangesPublish() // Sodalite#32: drop the tap routes first so a pump still draining its last packets no-ops // instead of decoding into stores being torn down. @@ -2754,6 +2852,11 @@ public final class HLSVideoEngine: @unchecked Sendable { server = nil let c = cache cache = nil + let cacheStatusCallback = onSessionCacheStatusChanged + let cacheRequestedBudget = requestedSessionCacheBudgetBytes.map { max(0, $0) } + let cacheEffectiveBudget = retentionBudgetBytes + let cacheVolumeSafetyLimit = sessionCacheVolumeSafetyLimitBytes + let cacheCapability: SessionCacheCapability = isLiveSession ? .unsupported : .active let ab = audioBridge audioBridge = nil let d = demuxer @@ -2807,7 +2910,16 @@ public final class HLSVideoEngine: @unchecked Sendable { iFrames?.shutdown() _ = p?.waitForFinish(timeout: 3.0) s?.stop() - c?.close() + if let c { + _ = c.close(reason: reason) + cacheStatusCallback?(c.status( + route: .loopbackFMP4, + requestedBudgetBytes: cacheRequestedBudget, + baseEffectiveBudgetBytes: cacheEffectiveBudget, + volumeSafetyLimitBytes: cacheVolumeSafetyLimit, + capability: cacheCapability + )) + } ab?.close() d?.close() sd?.close() @@ -2817,7 +2929,7 @@ public final class HLSVideoEngine: @unchecked Sendable { } deinit { - stop() + stop(reason: .sessionStopped) } // MARK: - I-frame rendition (AE#682) diff --git a/Sources/AetherEngine/Video/SegmentCache.swift b/Sources/AetherEngine/Video/SegmentCache.swift index 7d7cddc43..e0de5f659 100644 --- a/Sources/AetherEngine/Video/SegmentCache.swift +++ b/Sources/AetherEngine/Video/SegmentCache.swift @@ -10,6 +10,14 @@ import Foundation // across the producer/provider threads and capture in @Sendable closures. final class SegmentCache: @unchecked Sendable { + static let defaultBackwardWindow = 20 + + struct StaleSweepResult: Sendable, Equatable { + let inspectedCount: Int + let removedCount: Int + let failureCount: Int + } + /// AE#412: where a stored segment's first random-access point sits, as an offset from the /// segment's ADVERTISED start (its plan boundary). An offset, not an absolute time, so it is /// independent of the item / source / display axes and survives an epoch that opened early. @@ -73,6 +81,11 @@ final class SegmentCache: @unchecked Sendable { private var initVersions: [(versionID: Int, fromSegment: Int, data: Data)] = [] private var closed = false + private var _producerParked = false + private var _evictionCount = 0 + private var _lastFailure: SessionCacheFailureReason? + private var _lastCleanupReason: SessionCacheCleanupReason? + private var _lastCleanupResult: SessionCacheCleanupResult = .notRun /// Declared by provider at top of each mediaSegment(at:); non-monotonic (backward scrub is valid). private var currentTargetIndex: Int = -1 @@ -153,8 +166,16 @@ final class SegmentCache: @unchecked Sendable { } /// (10, 20)=30 entries, ~300 MB at 4K HDR HEVC ~10 MB/seg. - init(forwardWindow: Int = 10, backwardWindow: Int = 20, retentionBudgetBytes: Int = 0, - baseDirectory: URL? = nil, nativeLiveDVRPolicy: LiveDVRRetentionPolicy? = nil, onResidentSetChanged: (@Sendable () -> Void)? = nil) { + init( + forwardWindow: Int = 10, + backwardWindow: Int = SegmentCache.defaultBackwardWindow, + retentionBudgetBytes: Int = 0, + baseDirectory: URL? = nil, + sessionID: String = UUID().uuidString, + now: Date = Date(), + nativeLiveDVRPolicy: LiveDVRRetentionPolicy? = nil, + onResidentSetChanged: (@Sendable () -> Void)? = nil + ) { self.forwardWindow = forwardWindow self.backwardWindow = backwardWindow self.retentionBudgetBytes = retentionBudgetBytes @@ -164,24 +185,29 @@ final class SegmentCache: @unchecked Sendable { // aether-segments/ prefix lets sweepStaleSessionDirs() find sibling dirs from crashed sessions. let baseDir = baseDirectory ?? URL(fileURLWithPath: NSTemporaryDirectory(), isDirectory: true) .appendingPathComponent("aether-segments", isDirectory: true) - let sessionID = UUID().uuidString self.sessionDir = baseDir.appendingPathComponent(sessionID, isDirectory: true) do { try FileManager.default.createDirectory(at: sessionDir, withIntermediateDirectories: true, attributes: nil) } catch { + _lastFailure = .sessionDirectoryCreationFailed EngineLog.emit("[SegmentCache] session dir create failed at \(sessionDir.path): \(error)", category: .session) if Self.isOutOfSpace(error) { storageExhaustedLatch = true } } - // Before the sweep, so a sibling constructed in the same breath cannot read this session - // as unheld. The age check covers the remaining microseconds: a directory this young is - // never a sweep candidate. + // AE#451: protect this session from a sibling's stale sweep. self.lockFD = Self.acquireLiveMarker(sessionDir: sessionDir) - - Self.sweepStaleSessionDirs(baseDir: baseDir, currentSession: sessionID) + let sweep = Self.sweepStaleSessionDirs(baseDir: baseDir, currentSession: sessionID, now: now) + if sweep.failureCount > 0 { + _lastFailure = .staleSweepFailed + EngineLog.emit( + "[SegmentCache] bounded stale sweep: inspected=\(sweep.inspectedCount) " + + "removed=\(sweep.removedCount) failures=\(sweep.failureCount)", + category: .session + ) + } } deinit { @@ -201,25 +227,53 @@ final class SegmentCache: @unchecked Sendable { if fd >= 0 { Darwin.close(fd) } } - private static func sweepStaleSessionDirs(baseDir: URL, currentSession: String) { + /// Bounded crash-remnant cleanup. Only expired sibling directories are eligible; the current + /// UUID and fresh sessions are never removed. Preserve upstream creation-time age and held-lock + /// liveness checks. Both directory inspection and deletion are capped + /// so a cache root with many crash remnants cannot create an unbounded startup I/O burst. + static func sweepStaleSessionDirs( + baseDir: URL, + currentSession: String, + now: Date = Date(), + maxEntries: Int = 128, + maxRemovals: Int = 16 + ) -> StaleSweepResult { let fm = FileManager.default - guard let entries = try? fm.contentsOfDirectory(at: baseDir, - includingPropertiesForKeys: [.creationDateKey], - options: [.skipsHiddenFiles]) else { - return + guard maxEntries > 0, maxRemovals > 0 else { + return StaleSweepResult(inspectedCount: 0, removedCount: 0, failureCount: 0) } - let cutoff = Date().addingTimeInterval(-3600) - for entry in entries where entry.lastPathComponent != currentSession { - let created = (try? entry.resourceValues(forKeys: [.creationDateKey]))?.creationDate - guard created == nil || created! < cutoff else { continue } - // AE#451: age says how long it has been there, not whether anyone is still using it. - if SessionDirectoryLiveness.isLive(entry) { - EngineLog.emit("[SegmentCache] sweep spared live session dir \(entry.lastPathComponent)", - category: .session) - continue + var failures = 0 + guard let enumerator = fm.enumerator( + at: baseDir, + includingPropertiesForKeys: [.isDirectoryKey, .creationDateKey], + options: [.skipsHiddenFiles, .skipsSubdirectoryDescendants], + errorHandler: { _, _ in failures += 1; return false } + ) else { + return StaleSweepResult(inspectedCount: 0, removedCount: 0, failureCount: max(1, failures)) + } + + let cutoff = now.addingTimeInterval(-3600) + var inspected = 0 + var removed = 0 + while inspected < maxEntries, let entry = enumerator.nextObject() as? URL { + inspected += 1 + guard entry.lastPathComponent != currentSession, removed < maxRemovals else { continue } + let values: URLResourceValues + do { values = try entry.resourceValues(forKeys: [.isDirectoryKey, .creationDateKey]) } + catch { failures += 1; continue } + guard values.isDirectory == true, + let created = values.creationDate, + created < cutoff else { continue } + // Old does not mean abandoned: spare sessions whose marker is held (AE#451). + if SessionDirectoryLiveness.isLive(entry) { continue } + do { + try fm.removeItem(at: entry) + removed += 1 + } catch { + failures += 1 } - try? fm.removeItem(at: entry) } + return StaleSweepResult(inspectedCount: inspected, removedCount: removed, failureCount: failures) } // MARK: - Writer side @@ -291,7 +345,8 @@ final class SegmentCache: @unchecked Sendable { return true } - func store(index: Int, data: Data) { + @discardableResult + func store(index: Int, data: Data) -> Bool { let fileURL = nextSegmentFileURL(index: index) var writeOK: Bool do { @@ -314,17 +369,19 @@ final class SegmentCache: @unchecked Sendable { category: .session) if Self.isOutOfSpace(finalError) { noteStorageExhausted() } writeOK = false + condition.lock() + _lastFailure = .writeFailed + condition.unlock() } else { writeOK = true } } - condition.lock() // store racing close() must not resurrect bookkeeping; entry would point into deleted sessionDir. guard !closed else { condition.unlock() try? FileManager.default.removeItem(at: fileURL) - return + return false } // A re-store of a resident index changes bytes, not residency; only an insertion or an // eviction moves the set, and both are already known here without walking it. @@ -348,13 +405,15 @@ final class SegmentCache: @unchecked Sendable { if let supersededFile { try? FileManager.default.removeItem(at: supersededFile) } for url in doomed { try? FileManager.default.removeItem(at: url) } if residentSetChanged { onResidentSetChanged?() } + return writeOK } /// Adopt a staging file via rename(2). Page cache pages stay warm; skips a Swift Data round trip. /// /// `videoReach` (AE#412) is what this segment's video offers a cold arrival; nil leaves the /// previous claim in place only if the index is re-adopted without one, which no caller does. - func adopt(index: Int, stagingPath: URL, byteCount: Int, videoReach: VideoReach? = nil) { + @discardableResult + func adopt(index: Int, stagingPath: URL, byteCount: Int, videoReach: VideoReach? = nil) -> Bool { let fileURL = nextSegmentFileURL(index: index) var renameOK: Bool do { @@ -369,9 +428,11 @@ final class SegmentCache: @unchecked Sendable { category: .session) try? FileManager.default.removeItem(at: stagingPath) renameOK = false + condition.lock() + _lastFailure = .adoptFailed + condition.unlock() } } - condition.lock() // Sealed or not, this staging file is no longer being written. if inProgress[index] == stagingPath { inProgress.removeValue(forKey: index) } @@ -387,7 +448,7 @@ final class SegmentCache: @unchecked Sendable { condition.broadcast() condition.unlock() try? FileManager.default.removeItem(at: fileURL) - return + return false } var residentSetChanged = false var supersededFile: URL? @@ -415,11 +476,19 @@ final class SegmentCache: @unchecked Sendable { if let supersededFile { try? FileManager.default.removeItem(at: supersededFile) } for url in doomed { try? FileManager.default.removeItem(at: url) } if residentSetChanged { onResidentSetChanged?() } + return renameOK } - func close() { + @discardableResult + func close(reason: SessionCacheCleanupReason = .sessionStopped) -> SessionCacheCleanupResult { condition.lock() + if closed { + let prior = _lastCleanupResult + condition.unlock() + return prior + } closed = true + _lastCleanupReason = reason let expiryTimer = nativeLiveExpiryTimer nativeLiveExpiryTimer = nil let dir = sessionDir @@ -434,15 +503,28 @@ final class SegmentCache: @unchecked Sendable { sealedStagingOrder.removeAll(keepingCapacity: false) _totalBytes = 0 _highestStoredIndex = -1 + _producerParked = false condition.broadcast() condition.unlock() expiryTimer?.cancel() releaseLiveMarker() - try? FileManager.default.removeItem(at: dir) - // A closed cache holds nothing, and that is a resident-set change like any other. The engine - // clears its published band on teardown anyway; this keeps the cache honest on its own. + let result: SessionCacheCleanupResult + do { + if FileManager.default.fileExists(atPath: dir.path) { + try FileManager.default.removeItem(at: dir) + } + result = .succeeded + } catch { + result = .failed + EngineLog.emit("[SegmentCache] session cleanup failed: \(error)", category: .session) + } + condition.lock() + _lastCleanupResult = result + if result == .failed { _lastFailure = .cleanupFailed } + condition.unlock() if hadEntries { onResidentSetChanged?() } + return result } // MARK: - Reader side @@ -663,6 +745,7 @@ final class SegmentCache: @unchecked Sendable { entries.removeValue(forKey: k) doomed.append(url) } + _evictionCount += doomed.count condition.unlock() for url in doomed { try? FileManager.default.removeItem(at: url) @@ -696,6 +779,61 @@ final class SegmentCache: @unchecked Sendable { return currentTargetIndex } + var producerParked: Bool { + condition.lock() + defer { condition.unlock() } + return _producerParked + } + + var lastCleanupReason: SessionCacheCleanupReason? { + condition.lock() + defer { condition.unlock() } + return _lastCleanupReason + } + + var lastFailure: SessionCacheFailureReason? { + condition.lock() + defer { condition.unlock() } + return _lastFailure + } + + func status( + route: SessionCacheRoute, + requestedBudgetBytes: Int?, + baseEffectiveBudgetBytes: Int, + volumeSafetyLimitBytes: Int?, + capability: SessionCacheCapability = .active + ) -> SessionCacheStatus { + condition.lock() + defer { condition.unlock() } + let lo = currentTargetIndex - backwardWindow + let hi = max(currentTargetIndex + forwardWindow, _highestStoredIndex) + var hardFloor = 0 + for (index, bytes) in entryBytes where index >= lo && index <= hi { + hardFloor += bytes + } + let effective = max(baseEffectiveBudgetBytes, hardFloor) + return SessionCacheStatus( + route: route, + capability: closed ? .inactive : capability, + requestedBudgetBytes: requestedBudgetBytes, + volumeSafetyLimitBytes: volumeSafetyLimitBytes, + baseEffectiveBudgetBytes: baseEffectiveBudgetBytes, + playbackSafeFloorBytes: hardFloor, + effectiveBudgetBytes: effective, + currentResidentBytes: _totalBytes, + currentForwardBytes: currentForwardBytes(), + forwardWindowSegments: forwardWindow, + backwardWindowSegments: backwardWindow, + producerParked: _producerParked, + evictionCount: _evictionCount, + hardWindowFloorExceeded: hardFloor > baseEffectiveBudgetBytes, + cleanupReason: _lastCleanupReason, + cleanupResult: _lastCleanupResult, + lastFailure: _lastFailure + ) + } + /// How often a pump has passed a plan index without opening a segment for it (#358). /// /// The keyframe-gated cutter opens a segment at the IRAP that reaches a boundary, so a boundary @@ -895,9 +1033,15 @@ final class SegmentCache: @unchecked Sendable { func awaitPrefetchDiskHeadroom(head: Int, budgetBytes: Int, timeout: TimeInterval = 1.0) -> Bool { condition.lock() defer { condition.unlock() } - if !shouldParkLocked(head: head, budgetBytes: budgetBytes) { return true } + if !shouldParkLocked(head: head, budgetBytes: budgetBytes) { + _producerParked = false + return true + } + _producerParked = true _ = condition.wait(until: Date().addingTimeInterval(timeout)) - return !shouldParkLocked(head: head, budgetBytes: budgetBytes) + let released = !shouldParkLocked(head: head, budgetBytes: budgetBytes) + _producerParked = !released + return released } /// Must be called with condition held. @@ -1030,6 +1174,7 @@ final class SegmentCache: @unchecked Sendable { doomed.append(url) } } + _evictionCount += doomed.count return doomed } for (k, url) in entries { @@ -1041,6 +1186,7 @@ final class SegmentCache: @unchecked Sendable { doomed.append(url) } } + _evictionCount += doomed.count // Collected under the lock, deleted by the caller AFTER // unlocking: removeItem is filesystem I/O on the segment-serve // hot path (fetch waiters + the pump's backpressure wait park on diff --git a/Tests/AetherEngineTests/SessionCacheContractTests.swift b/Tests/AetherEngineTests/SessionCacheContractTests.swift new file mode 100644 index 000000000..04fa1a359 --- /dev/null +++ b/Tests/AetherEngineTests/SessionCacheContractTests.swift @@ -0,0 +1,342 @@ +import Foundation +import Testing +@testable import AetherEngine + +@Suite("Session cache public contract") +struct SessionCachePublicContractTests { + @Test("LoadOptions carries an exact caller-requested session budget") + func loadOptionsBudget() { + let auto = LoadOptions(sessionCacheByteBudget: 512 << 20) + let disk = LoadOptions(sessionCacheByteBudget: 16 << 30) + let disabled = LoadOptions(sessionCacheByteBudget: 0) + #expect(auto.sessionCacheByteBudget == 512 << 20) + #expect(disk.sessionCacheByteBudget == 16 << 30) + #expect(disabled.sessionCacheByteBudget == 0) + } + + @Test("Exact requests clamp to one quarter of free tmp capacity") + func requestedBudgetClampsToVolume() { + let result = HLSVideoEngine.resolveSessionCacheBudget( + requestedBytes: 8 << 30, + volumeAvailableBytes: 16 << 30, + capRelaxed: true + ) + #expect(result.requestedBytes == 8 << 30) + #expect(result.volumeSafetyLimitBytes == 4 << 30) + #expect(result.baseEffectiveBytes == 4 << 30) + } + + @Test("Unknown capacity keeps the conservative 2 GiB safety ceiling") + func unknownCapacityIsConservative() { + let result = HLSVideoEngine.resolveSessionCacheBudget( + requestedBytes: 16 << 30, + volumeAvailableBytes: nil, + capRelaxed: true + ) + #expect(result.requestedBytes == 16 << 30) + #expect(result.baseEffectiveBytes == 2 << 30) + } + + @Test("Zero expresses window-only caching") + func zeroBudgetIsWindowOnly() { + let result = HLSVideoEngine.resolveSessionCacheBudget( + requestedBytes: 0, + volumeAvailableBytes: 100 << 30, + capRelaxed: false + ) + #expect(result.baseEffectiveBytes == 0) + } + + @Test("Native remote HLS reports unsupported rather than zero-byte support") + func nativeRemoteHLSIsUnsupported() { + let status = SessionCacheStatus.nativeRemoteHLS(requestedBudgetBytes: 512 << 20) + #expect(status.route == .nativeRemoteHLS) + #expect(status.capability == .unsupported) + #expect(status.requestedBudgetBytes == 512 << 20) + #expect(status.currentResidentBytes == 0) + } +} + +@Suite("Session cache lifecycle and pressure contract", .offCooperativePool, .timeLimit(.minutes(1))) +struct SessionCacheLifecycleContractTests { + private func makeRoot() throws -> URL { + let root = FileManager.default.temporaryDirectory + .appendingPathComponent("aether-cache-contract-\(UUID().uuidString)", isDirectory: true) + try FileManager.default.createDirectory(at: root, withIntermediateDirectories: true) + return root + } + + @Test("Runtime hard window raises effective bytes instead of starving playback") + func hardWindowRaisesEffectiveBudget() throws { + let root = try makeRoot() + defer { try? FileManager.default.removeItem(at: root) } + let cache = SegmentCache( + forwardWindow: 1_000, + backwardWindow: 20, + retentionBudgetBytes: 50, + baseDirectory: root, + sessionID: "current" + ) + defer { _ = cache.close(reason: .sessionStopped) } + cache.declareTarget(0) + for index in 0...10 { + #expect(cache.store(index: index, data: Data(repeating: 1, count: 10))) + } + let status = cache.status( + route: .loopbackFMP4, + requestedBudgetBytes: 50, + baseEffectiveBudgetBytes: 50, + volumeSafetyLimitBytes: 1_000 + ) + #expect(status.playbackSafeFloorBytes == 110) + #expect(status.effectiveBudgetBytes == 110) + #expect(status.hardWindowFloorExceeded) + } + + @Test("Producer park is observable and releases when the consumer advances") + func producerParkStatus() throws { + let root = try makeRoot() + defer { try? FileManager.default.removeItem(at: root) } + let cache = SegmentCache( + forwardWindow: 1_000, + backwardWindow: 2, + retentionBudgetBytes: 50, + baseDirectory: root, + sessionID: "current" + ) + defer { _ = cache.close(reason: .sessionStopped) } + cache.declareTarget(0) + for index in 0...10 { _ = cache.store(index: index, data: Data(repeating: 1, count: 10)) } + #expect(cache.awaitPrefetchDiskHeadroom(head: 11, budgetBytes: 50, timeout: 0.01) == false) + #expect(cache.producerParked) + cache.declareTarget(9) + #expect(cache.awaitPrefetchDiskHeadroom(head: 11, budgetBytes: 50, timeout: 0.01)) + #expect(cache.producerParked == false) + } + + @Test("Close deletes the session directory and records its reason") + func closeDeletesSession() throws { + let root = try makeRoot() + defer { try? FileManager.default.removeItem(at: root) } + let cache = SegmentCache(baseDirectory: root, sessionID: "current") + #expect(cache.store(index: 0, data: Data(repeating: 1, count: 32))) + let result = cache.close(reason: .sourceChanged) + #expect(result == .succeeded) + #expect(FileManager.default.fileExists(atPath: cache.sessionDir.path) == false) + #expect(cache.lastCleanupReason == .sourceChanged) + } + + @Test("A failed session directory makes writes fail closed without false accounting") + func writeFailureFailsClosed() throws { + let rootFile = FileManager.default.temporaryDirectory + .appendingPathComponent("aether-cache-file-\(UUID().uuidString)") + try Data([1]).write(to: rootFile) + defer { try? FileManager.default.removeItem(at: rootFile) } + let cache = SegmentCache(baseDirectory: rootFile, sessionID: "impossible") + #expect(cache.store(index: 0, data: Data(repeating: 1, count: 32)) == false) + #expect(cache.totalBytes == 0) + #expect(cache.lastFailure == .writeFailed) + _ = cache.close(reason: .loadFailed) + } + + @Test("A failed replacement preserves the previous generation and its accounting") + func adoptFailurePreservesResidentGeneration() throws { + let root = try makeRoot() + defer { try? FileManager.default.removeItem(at: root) } + let cache = SegmentCache(baseDirectory: root, sessionID: "current") + defer { _ = cache.close(reason: .sessionStopped) } + #expect(cache.store(index: 0, data: Data(repeating: 1, count: 32))) + let resident = try #require(cache.peek(index: 0)) + let missing = root.appendingPathComponent("missing-staging.m4s") + #expect(cache.adopt(index: 0, stagingPath: missing, byteCount: 64) == false) + #expect(cache.totalBytes == 32) + #expect(cache.peek(index: 0) == resident) + #expect(resident == Data(repeating: 1, count: 32)) + #expect(cache.lastFailure == .adoptFailed) + } + + @Test("Bounded stale sweep removes only expired siblings and preserves current and fresh") + func boundedStaleSweep() throws { + let root = try makeRoot() + defer { try? FileManager.default.removeItem(at: root) } + let now = Date() + let current = root.appendingPathComponent("current", isDirectory: true) + let fresh = root.appendingPathComponent("fresh", isDirectory: true) + try FileManager.default.createDirectory(at: current, withIntermediateDirectories: true) + try FileManager.default.createDirectory(at: fresh, withIntermediateDirectories: true) + var stale: [URL] = [] + for index in 0..<5 { + let url = root.appendingPathComponent("stale-\(index)", isDirectory: true) + try FileManager.default.createDirectory(at: url, withIntermediateDirectories: true) + try FileManager.default.setAttributes( + [.creationDate: now.addingTimeInterval(-7_200 - Double(index))], + ofItemAtPath: url.path + ) + stale.append(url) + } + + let first = SegmentCache.sweepStaleSessionDirs( + baseDir: root, + currentSession: "current", + now: now, + maxEntries: 7, + maxRemovals: 2 + ) + #expect(first.inspectedCount <= 7) + #expect(first.removedCount == 2) + #expect(FileManager.default.fileExists(atPath: current.path)) + #expect(FileManager.default.fileExists(atPath: fresh.path)) + #expect(stale.filter { FileManager.default.fileExists(atPath: $0.path) }.count == 3) + } +} + +@Suite("Session budgets preserve automatic and multi-session policy") +struct SessionCacheBudgetCompatibilityTests { + @Test(arguments: [Int64(-1), 0, 1024, 16 << 30], [false, true]) + func automaticUnchanged(available: Int64, relaxed: Bool) { + let expected = relaxed ? max(0, Int(available / 4)) : min(2 << 30, max(0, Int(available / 4))) + #expect(HLSVideoEngine.resolveSessionCacheBudget(requestedBytes: nil, + volumeAvailableBytes: available, capRelaxed: relaxed).baseEffectiveBytes == expected) + } + + @Test func negativeAndUnknown() { + #expect(HLSVideoEngine.resolveSessionCacheBudget(requestedBytes: -1, + volumeAvailableBytes: 100).baseEffectiveBytes == 0) + #expect(HLSVideoEngine.resolveSessionCacheBudget(requestedBytes: nil, + volumeAvailableBytes: nil, capRelaxed: true).baseEffectiveBytes == 2 << 30) + } + + @Test func sharedClaimsLeaveHeadroomForLaterSessions() { + let ledger = RetentionClaims() + let first = ledger.claim(volumeAvailableBytes: 1600) { + HLSVideoEngine.resolveSessionCacheBudget(requestedBytes: 1000, volumeAvailableBytes: $0).baseEffectiveBytes + } + let second = ledger.claim(volumeAvailableBytes: 1600) { + HLSVideoEngine.resolveSessionCacheBudget(requestedBytes: 1000, volumeAvailableBytes: $0).baseEffectiveBytes + } + #expect(first.bytes == 400) + #expect(second.bytes == 300) + first.release() + second.release() + let shrunk = ledger.claim(volumeAvailableBytes: 100) { + HLSVideoEngine.resolveSessionCacheBudget(requestedBytes: 1000, volumeAvailableBytes: $0).baseEffectiveBytes + } + #expect(shrunk.bytes == 25) + shrunk.release() + } +} + +@MainActor +@Suite(.timeLimit(.minutes(1))) +struct SessionCachePublicationTests { + @Test func oldCallbackCannotReplaceNewSessionOrStoppedState() throws { + let engine = try AetherEngine() + let old = engine.sessionCacheStatusToken + engine.sessionCacheStatusToken &+= 1 + let expected = SessionCacheStatus.nativeRemoteHLS(requestedBudgetBytes: 123) + engine.diagnostics.sessionCacheStatus = expected + engine.acceptSessionCacheStatus(.inactive, token: old) + #expect(engine.diagnostics.sessionCacheStatus == expected) + // A queued startup snapshot cannot resurrect a cache after stop detached the session. + engine.acceptSessionCacheStatus(.inactive, token: engine.sessionCacheStatusToken) + #expect(engine.diagnostics.sessionCacheStatus == expected) + engine.stop() + #expect(engine.diagnostics.sessionCacheStatus == .inactive) + } + + @Test func actualLoadStopReportsDeletionAndNewLoadRejectsOldCleanup() async throws { + let engine = try AetherEngine() + defer { engine.stop() } + func source() throws -> MediaSource { + .custom(DataIOReader(data: try ProbeTestFixtures.hdr10Plus()), formatHint: "mp4") + } + _ = try await engine.load(source: source(), options: LoadOptions(sessionCacheByteBudget: 0)) + let old = try #require(engine.nativeVideoSession) + let directory = try #require(old.cache).sessionDir + let oldToken = engine.sessionCacheStatusToken + await engine.refreshSessionCacheStatus() + #expect(engine.diagnostics.sessionCacheStatus.route == .loopbackFMP4) + #expect(engine.diagnostics.sessionCacheStatus.baseEffectiveBudgetBytes == 0) + engine.stop() + try await waitFor { engine.diagnostics.sessionCacheStatus.cleanupResult != .notRun } + let cleaned = engine.diagnostics.sessionCacheStatus + #expect(cleaned.cleanupReason == .sessionStopped) + #expect(cleaned.cleanupResult == .succeeded) + #expect(cleaned.capability == .inactive) + #expect(!FileManager.default.fileExists(atPath: directory.path)) + _ = try await engine.load(source: source(), options: LoadOptions(sessionCacheByteBudget: 1024)) + await engine.refreshSessionCacheStatus() + let current = engine.diagnostics.sessionCacheStatus + engine.acceptSessionCacheStatus(cleaned, token: oldToken) + #expect(engine.diagnostics.sessionCacheStatus == current) + #expect(current.requestedBudgetBytes == 1024) + } +} + +extension SessionCacheLifecycleContractTests { + @Test func liveBudgetCapabilityAndClosedStateAreExplicit() throws { + let root = try makeRoot() + defer { try? FileManager.default.removeItem(at: root) } + let cache = SegmentCache(baseDirectory: root) + let session = HLSVideoEngine(url: URL(string: "https://example.invalid/live")!, + isLiveSession: true, sessionCacheByteBudget: 123) + session.cache = cache + #expect(session.sessionCacheStatus().capability == .unsupported) + _ = cache.close(reason: .sessionStopped) + #expect(session.sessionCacheStatus().capability == .inactive) + } + + @Test func cleanupFailureDoesNotClaimSuccessfulDeletion() throws { + let root = try makeRoot() + let cache = SegmentCache(baseDirectory: root) + defer { + try? FileManager.default.setAttributes([.posixPermissions: 0o700], ofItemAtPath: root.path) + try? FileManager.default.removeItem(at: root) + } + try FileManager.default.setAttributes([.posixPermissions: 0o500], ofItemAtPath: root.path) + #expect(cache.close(reason: .loadFailed) == .failed) + let status = cache.status(route: .loopbackFMP4, requestedBudgetBytes: nil, + baseEffectiveBudgetBytes: 0, volumeSafetyLimitBytes: nil) + #expect(status.cleanupReason == .loadFailed) + #expect(status.lastFailure == .cleanupFailed) + #expect(FileManager.default.fileExists(atPath: cache.sessionDir.path)) + } + + @Test func sweepEnumerationFailureIsReported() throws { + let root = try makeRoot() + defer { try? FileManager.default.removeItem(at: root) } + let missing = root.appendingPathComponent("missing") + let result = SegmentCache.sweepStaleSessionDirs(baseDir: missing, currentSession: "unused") + #expect(result.failureCount > 0) + #expect(result.removedCount == 0) + } +} + +extension SessionCacheBudgetCompatibilityTests { + @Test(arguments: [Int64(0), 1024, 16 << 30], [false, true]) + func softwareAutomaticIsUnchanged(available: Int64, relaxed: Bool) { + #expect(SoftwarePlaybackHost.packetCacheBudgetBytes(requestedBytes: nil, + availableBytes: available, capRelaxed: relaxed) == + HLSVideoEngine.sessionRetentionBudgetBytes(volumeAvailableBytes: available, capRelaxed: relaxed)) + } + + @Test func softwareWindowOnlyStillRespectsVolumeSafety() { + #expect(SoftwarePlaybackHost.packetCacheBudgetBytes(requestedBytes: 0, + availableBytes: 16 << 30, capRelaxed: true) == 32 << 20) + #expect(SoftwarePlaybackHost.packetCacheBudgetBytes(requestedBytes: 0, + availableBytes: 1024, capRelaxed: true) == 256) + #expect(SoftwarePlaybackHost.packetCacheBudgetBytes(requestedBytes: 100, + availableBytes: 1024, capRelaxed: true) == 100) + #expect(SoftwarePlaybackHost.packetCacheBudgetBytes(requestedBytes: Int.max, + availableBytes: nil, capRelaxed: true) == 2 << 30) + } +} + +extension SessionCachePublicContractTests { + @Test func stopRetainsZeroArgumentCallbackCompatibility() { + let session = HLSVideoEngine(url: URL(string: "https://example.invalid/unused")!) + let stop: () -> Void = session.stop + stop() + #expect(session.sessionCacheStatus().capability == .inactive) + } +} diff --git a/docs/api.md b/docs/api.md index 16362779c..78d8947de 100644 --- a/docs/api.md +++ b/docs/api.md @@ -1378,3 +1378,13 @@ Public for the CLI, the test suite, or a diagnostic overlay, and outside the sha - **Dolby Vision Profile 7 FEL tooling** (`@_spi(Diagnostics)`): `FelProbeResult` / `felDecodeProbe` (`aetherctl feltest`) and `felResidualEnabled` (`aetherctl play --cycle-el`), which exist only with the `DualLayerDolbyVision` trait, `DoviRpuSurvey` / `DoviRpuSurveyResult` / `DoviRpuFrameSurvey` / `doviRpuSurvey` (`aetherctl dvsurvey`), and, in `DEBUG` builds with the `DualLayerDolbyVision` trait only, `armDolbyVisionPinTest` / `dolbyVisionPinIsArmed` / `dolbyVisionPinIsActive` (`aetherctl play --dv-pin`). Hosts import none of them; the host surface of the FEL route is `LoadOptions.preferDualLayerDolbyVision` and, in `SessionReloadRefusal`, the new case `.dualLayerRouteNeedsFreshLoad`. See [Dolby Vision FEL](dolby-vision-fel.md) and [CLI](cli.md). - **`HLSLiveIngestReader`'s internals** (`terminalError`, `upstreamTargetDuration`, `observedLiveCadenceSeconds`, `closedLiveCadenceSeconds`, `upstreamSegmentDurationSeconds`, `companionAudioReader`): fixture and diagnostic reads. The last two are the closed evidence the served TARGETDURATION is sealed from (AE#447); `upstreamTargetDuration` is the upstream's own claim, reported in the seal line and derived from nowhere. - **`SubtitleChannel`**: the primary / secondary selector on the engine's internal subtitle routing. No public signature takes one; a host picks the channel by calling the primary or the secondary method. + +### Session cache budget and status + +`LoadOptions.sessionCacheByteBudget` is an optional byte request for the temporary VOD cache. Nil preserves automatic sizing: the existing 2 GiB default (relaxed for an explicitly large forward window), volume safety clamp and process-wide `RetentionClaims` allocation remain in force. Explicit requests are nonnegative and clamped to one quarter of capacity left after other sessions' unwritten claims; unknown capacity keeps the 2 GiB ceiling. Zero requests window-only operation, not zero memory or zero disk I/O. The software disk path keeps a bounded working spool (its existing packet-retention mechanism, not the fMP4 segment window); the tvOS memory-store default and live DVR retention policies are unchanged. + +Observe `diagnostics.sessionCacheStatus` on `EngineDiagnostics`, separately from the engine's UI state. `SessionCacheStatus` reports `route` (`SessionCacheRoute`: `none`, `loopbackFMP4`, `nativeRemoteHLS`), `capability` (`SessionCacheCapability`: `inactive`, `active`, `unsupported`), `requestedBudgetBytes`, `volumeSafetyLimitBytes`, `baseEffectiveBudgetBytes`, `playbackSafeFloorBytes`, `effectiveBudgetBytes`, `currentResidentBytes`, `currentForwardBytes`, `forwardWindowSegments`, `backwardWindowSegments`, `producerParked`, `evictionCount` and `hardWindowFloorExceeded`. A protected playback window can exceed the byte request; the snapshot reports that instead of claiming a strict cap which could starve playback. Live fMP4 sessions report `unsupported` for this VOD byte policy and keep their separate live DVR limits; closed caches report `inactive` with the final cleanup result. The detailed snapshot describes fMP4 segment caching; remote HLS is AVPlayer/server managed and reports unsupported, not successful zero-byte caching. Software packet storage retains its existing separate telemetry. + +`cleanupReason` (`SessionCacheCleanupReason`: `sessionStopped`, `sourceChanged`, `loadFailed`) and `cleanupResult` (`SessionCacheCleanupResult`: `notRun`, `succeeded`, `failed`) describe asynchronous deletion after the producer stops. `lastFailure` (`SessionCacheFailureReason`: `sessionDirectoryCreationFailed`, `writeFailed`, `adoptFailed`, `staleSweepFailed`, `cleanupFailed`) carries a credential-free category. `inactive` is the empty snapshot; `nativeRemoteHLS(requestedBudgetBytes:)` constructs the unsupported route state. Old-session callbacks cannot replace a newer session's state. + +Direct `HLSVideoEngine` users may call `sessionCacheStatus()` off the main thread and receive lifecycle updates through `onSessionCacheStatusChanged`. `stop(reason:)` records the cleanup reason; the existing zero-argument `stop()` and its callback type remain available. Engine sampling reads the in-memory byte ledger off-main at the existing telemetry cadence, without enumerating files on the UI thread. Successful cleanup removes the complete session UUID directory. A bounded startup sweep removes eligible abandoned siblings; it never adopts their bytes or supplies an offline/cross-launch cache.