diff --git a/Jellyfin.Api/Controllers/DynamicHlsController.cs b/Jellyfin.Api/Controllers/DynamicHlsController.cs index 034a9dea55..bce703e34d 100644 --- a/Jellyfin.Api/Controllers/DynamicHlsController.cs +++ b/Jellyfin.Api/Controllers/DynamicHlsController.cs @@ -60,6 +60,7 @@ public class DynamicHlsController : BaseJellyfinApiController private readonly IDynamicHlsPlaylistGenerator _dynamicHlsPlaylistGenerator; private readonly DynamicHlsHelper _dynamicHlsHelper; private readonly EncodingOptions _encodingOptions; + private readonly ITranscodeSessionStore _transcodeSessionStore; /// /// Initializes a new instance of the class. @@ -75,6 +76,7 @@ public class DynamicHlsController : BaseJellyfinApiController /// Instance of . /// Instance of . /// Instance of . + /// Instance of the interface used to register and renew HLS transcoding session leases in the durable store. public DynamicHlsController( ILibraryManager libraryManager, IUserManager userManager, @@ -86,7 +88,8 @@ public class DynamicHlsController : BaseJellyfinApiController ILogger logger, DynamicHlsHelper dynamicHlsHelper, EncodingHelper encodingHelper, - IDynamicHlsPlaylistGenerator dynamicHlsPlaylistGenerator) + IDynamicHlsPlaylistGenerator dynamicHlsPlaylistGenerator, + ITranscodeSessionStore transcodeSessionStore) { _libraryManager = libraryManager; _userManager = userManager; @@ -99,6 +102,7 @@ public class DynamicHlsController : BaseJellyfinApiController _dynamicHlsHelper = dynamicHlsHelper; _encodingHelper = encodingHelper; _dynamicHlsPlaylistGenerator = dynamicHlsPlaylistGenerator; + _transcodeSessionStore = transcodeSessionStore; _encodingOptions = serverConfigurationManager.GetEncodingOptions(); } @@ -306,15 +310,23 @@ public class DynamicHlsController : BaseJellyfinApiController // If the playlist doesn't already exist, startup ffmpeg try { + var isHaMode = await IsHaTakeoverAsync(playSessionId, cancellationToken).ConfigureAwait(false); + job = await _transcodeManager.StartFfMpeg( state, playlistPath, - GetCommandLineArguments(playlistPath, state, true, 0), + GetCommandLineArguments(playlistPath, state, true, 0, isHaMode), Request.HttpContext.User.GetUserId(), TranscodingJobType, cancellationTokenSource) .ConfigureAwait(false); job.IsLiveOutput = true; + await RegisterTranscodeSessionAsync( + playSessionId ?? string.Empty, + mediaSourceId ?? string.Empty, + cancellationToken) + .ConfigureAwait(false); + StartLeaseRenewal(playSessionId ?? string.Empty, cancellationToken); } catch { @@ -1511,14 +1523,22 @@ public class DynamicHlsController : BaseJellyfinApiController streamingRequest.StartTimeTicks = streamingRequest.CurrentRuntimeTicks; + var isHaMode = await IsHaTakeoverAsync(streamingRequest.PlaySessionId, cancellationToken).ConfigureAwait(false); + state.WaitForPath = segmentPath; job = await _transcodeManager.StartFfMpeg( state, playlistPath, - GetCommandLineArguments(playlistPath, state, false, segmentId), + GetCommandLineArguments(playlistPath, state, false, segmentId, isHaMode), Request.HttpContext.User.GetUserId(), TranscodingJobType, cancellationTokenSource).ConfigureAwait(false); + await RegisterTranscodeSessionAsync( + streamingRequest.PlaySessionId ?? string.Empty, + streamingRequest.MediaSourceId ?? string.Empty, + cancellationToken) + .ConfigureAwait(false); + StartLeaseRenewal(streamingRequest.PlaySessionId ?? string.Empty, cancellationToken); } catch { @@ -1554,6 +1574,99 @@ public class DynamicHlsController : BaseJellyfinApiController private static double[] GetSegmentLengths(StreamState state) => GetSegmentLengthsInternal(state.RunTimeTicks ?? 0, state.SegmentLength); + /// + /// Determines whether the play session is already present in the durable store, which means + /// another pod owned it and this request is resuming it after a failover. + /// + private async Task IsHaTakeoverAsync(string? playSessionId, CancellationToken cancellationToken) + { + if (string.IsNullOrEmpty(playSessionId)) + { + return false; + } + + try + { + return await _transcodeSessionStore.TryGetAsync(playSessionId, cancellationToken).ConfigureAwait(false) is not null; + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Failed to check HA mode for session {PlaySessionId}.", playSessionId); + return false; + } + } + + private async Task RegisterTranscodeSessionAsync(string playSessionId, string mediaSourceId, CancellationToken cancellationToken) + { + try + { + var session = new TranscodeSession + { + PlaySessionId = playSessionId, + OwnerPod = Environment.GetEnvironmentVariable("JELLYFIN_INSTANCE_ID") + ?? Environment.MachineName, + LeaseExpiresUtc = DateTime.UtcNow.AddSeconds(30), + ManifestPath = string.Empty, + SegmentPathPrefix = string.Empty, + MediaSourceId = mediaSourceId, + LastCompletedSegmentIndex = 0, + LastDurablePlaybackOffset = 0L, + }; + await _transcodeSessionStore.SetAsync(session, cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Failed to register HLS session {PlaySessionId} in durable store.", playSessionId); + } + } + + private void StartLeaseRenewal(string playSessionId, CancellationToken cancellationToken) + { + _ = Task.Run( + async () => + { + try + { + while (!cancellationToken.IsCancellationRequested) + { + try + { + await Task.Delay(TimeSpan.FromSeconds(10), cancellationToken).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + break; + } + + try + { + await _transcodeSessionStore.RenewLeaseAsync(playSessionId, cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Failed to renew lease for HLS session {PlaySessionId}.", playSessionId); + } + } + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Lease renewal loop for HLS session {PlaySessionId} encountered an unexpected error.", playSessionId); + } + finally + { + try + { + await _transcodeSessionStore.DeleteAsync(playSessionId, CancellationToken.None).ConfigureAwait(false); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Failed to delete HLS session {PlaySessionId} from durable store.", playSessionId); + } + } + }, + CancellationToken.None); + } + internal static double[] GetSegmentLengthsInternal(long runtimeTicks, int segmentlength) { var segmentLengthTicks = TimeSpan.FromSeconds(segmentlength).Ticks; @@ -1575,7 +1688,7 @@ public class DynamicHlsController : BaseJellyfinApiController return segments; } - private string GetCommandLineArguments(string outputPath, StreamState state, bool isEventPlaylist, int startNumber) + private string GetCommandLineArguments(string outputPath, StreamState state, bool isEventPlaylist, int startNumber, bool isHaMode = false) { var videoCodec = _encodingHelper.GetVideoEncoder(state, _encodingOptions); var threads = EncodingHelper.GetNumberOfThreads(state, _encodingOptions, videoCodec); @@ -1588,10 +1701,20 @@ public class DynamicHlsController : BaseJellyfinApiController var outputExtension = EncodingHelper.GetSegmentFileExtension(state.Request.SegmentContainer); var outputTsArg = outputPrefix + "%d" + outputExtension; + // In HA mode, use shorter segments and a bounded rolling buffer for faster failover recovery. + // state.SegmentLength is already validated by the streaming pipeline; RecoverySegmentLengthSeconds + // comes from EncodingOptions (user-editable config) so it is clamped here. + var effectiveSegmentLength = isHaMode + ? Math.Clamp(_encodingOptions.RecoverySegmentLengthSeconds, 1, 6) + : state.SegmentLength; + var hlsListSize = isHaMode + ? Math.Clamp(_encodingOptions.RecoverySegmentBufferCount, 2, 10) + : 0; + var segmentFormat = string.Empty; var segmentContainer = outputExtension.TrimStart('.'); var inputModifier = _encodingHelper.GetInputModifier(state, _encodingOptions, segmentContainer); - var hlsArguments = $"-hls_playlist_type {(isEventPlaylist ? "event" : "vod")} -hls_list_size 0"; + var hlsArguments = $"-hls_playlist_type {(isEventPlaylist ? "event" : "vod")} -hls_list_size {hlsListSize}"; if (string.Equals(segmentContainer, "ts", StringComparison.OrdinalIgnoreCase)) { @@ -1644,10 +1767,10 @@ public class DynamicHlsController : BaseJellyfinApiController _encodingHelper.GetInputArgument(state, _encodingOptions, segmentContainer), threads, mapArgs, - GetVideoArguments(state, startNumber, isEventPlaylist, segmentContainer), + GetVideoArguments(state, startNumber, isEventPlaylist, segmentContainer, effectiveSegmentLength), GetAudioArguments(state), maxMuxingQueueSize, - state.SegmentLength.ToString(CultureInfo.InvariantCulture), + effectiveSegmentLength.ToString(CultureInfo.InvariantCulture), segmentFormat, startNumber.ToString(CultureInfo.InvariantCulture), baseUrlParam, @@ -1784,8 +1907,9 @@ public class DynamicHlsController : BaseJellyfinApiController /// The first number in the hls sequence. /// Whether the playlist is EVENT or VOD. /// The segment container. + /// The effective segment length in seconds (overrides when HA mode is active). /// The command line arguments for video transcoding. - private string GetVideoArguments(StreamState state, int startNumber, bool isEventPlaylist, string segmentContainer) + private string GetVideoArguments(StreamState state, int startNumber, bool isEventPlaylist, string segmentContainer, int? segmentLength = null) { if (state.VideoStream is null) { @@ -1860,7 +1984,7 @@ public class DynamicHlsController : BaseJellyfinApiController args += _encodingHelper.GetVideoQualityParam(state, codec, _encodingOptions, isEventPlaylist ? DefaultEventEncoderPreset : DefaultVodEncoderPreset); // Set the key frame params for video encoding to match the hls segment time. - args += _encodingHelper.GetHlsVideoKeyFrameArguments(state, codec, state.SegmentLength, isEventPlaylist, startNumber); + args += _encodingHelper.GetHlsVideoKeyFrameArguments(state, codec, segmentLength ?? state.SegmentLength, isEventPlaylist, startNumber); // Currently b-frames in libx265 breaks the FMP4-HLS playback on iOS, disable it for now. if (string.Equals(codec, "libx265", StringComparison.OrdinalIgnoreCase) diff --git a/MediaBrowser.Model/Configuration/EncodingOptions.cs b/MediaBrowser.Model/Configuration/EncodingOptions.cs index 4d052d8012..951d523f3e 100644 --- a/MediaBrowser.Model/Configuration/EncodingOptions.cs +++ b/MediaBrowser.Model/Configuration/EncodingOptions.cs @@ -25,6 +25,8 @@ public class EncodingOptions ThrottleDelaySeconds = 180; EnableSegmentDeletion = false; SegmentKeepSeconds = 720; + RecoverySegmentLengthSeconds = 2; + RecoverySegmentBufferCount = 5; EncodingThreadCount = -1; // This is a DRM device that is almost guaranteed to be there on every intel platform, // plus it's the default one in ffmpeg if you don't specify anything @@ -125,6 +127,20 @@ public class EncodingOptions /// public int SegmentKeepSeconds { get; set; } + /// + /// Gets or sets the HLS segment length in seconds to use when HA recovery mode is active. + /// Shorter segments allow a takeover pod to resume playback faster after a peer failure. + /// Default is 2. Valid range is 1-6. + /// + public int RecoverySegmentLengthSeconds { get; set; } + + /// + /// Gets or sets the number of HLS segments to keep on disk when HA recovery mode is active. + /// This acts as a rolling buffer that a takeover pod can serve while restarting the transcode. + /// Default is 5. Valid range is 2-10. + /// + public int RecoverySegmentBufferCount { get; set; } + /// /// Gets or sets the hardware acceleration type. /// diff --git a/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsHaTakeoverTests.cs b/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsHaTakeoverTests.cs new file mode 100644 index 0000000000..bdbca25d13 --- /dev/null +++ b/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsHaTakeoverTests.cs @@ -0,0 +1,259 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Jellyfin.Api.Controllers; +using MediaBrowser.Controller.MediaEncoding; +using Xunit; + +namespace Jellyfin.Api.Tests.Controllers +{ + /// + /// Tests for HA recovery scenarios that will be wired into + /// in Phase 5.2. These tests verify the contract that + /// the controller will rely on for missing-local-job recovery, claim racing, and cleanup guarding. + /// + public class DynamicHlsHaTakeoverTests + { + private static TranscodeSession CreateSession(string id, string pod, DateTime leaseExpiry) + => new TranscodeSession + { + PlaySessionId = id, + OwnerPod = pod, + LeaseExpiresUtc = leaseExpiry, + ManifestPath = $"/transcode/{id}/manifest.m3u8", + SegmentPathPrefix = $"/transcode/{id}/segment", + MediaSourceId = "media-source-1", + LastCompletedSegmentIndex = 3, + LastDurablePlaybackOffset = 18_000_000L, + }; + + /// + /// Missing-local-job + durable-manifest-present: the store returns the session so + /// the controller can serve the existing manifest instead of returning an error. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task DurableManifestPresent_WithLiveSession_StoreReturnsSession() + { + var store = new HaTestSessionStore(); + var session = CreateSession("ha-session-1", "pod-a", DateTime.UtcNow.AddMinutes(5)); + await store.SetAsync(session, TestContext.Current.CancellationToken); + + // Simulate controller recovery: look up the session in the durable store. + var recovered = await store.TryGetAsync("ha-session-1", TestContext.Current.CancellationToken); + + Assert.NotNull(recovered); + Assert.Equal("/transcode/ha-session-1/manifest.m3u8", recovered.ManifestPath); + } + + /// + /// Claim-race between two concurrent requesters: only one wins + /// . + /// The other receives false, indicating it should redirect (302) or wait. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task ClaimRace_TwoConcurrentRequesters_OnlyOneWinsTakeover() + { + var store = new HaTestSessionStore(); + + // Original pod crashed – lease is expired. + var session = CreateSession("ha-session-2", "pod-a", DateTime.UtcNow.AddMilliseconds(-1)); + await store.SetAsync(session, TestContext.Current.CancellationToken); + + // Two pods simultaneously attempt to claim the orphaned session. + var task1 = store.TryTakeoverAsync("ha-session-2", "pod-b", TestContext.Current.CancellationToken); + var task2 = store.TryTakeoverAsync("ha-session-2", "pod-c", TestContext.Current.CancellationToken); + var results = await Task.WhenAll(task1, task2); + + // Exactly one pod must win. + var wins = Array.FindAll(results, r => r); + Assert.Single(wins); + } + + /// + /// Stale-manifest cleanup guard: a lease that has expired beyond the recovery window + /// causes the store to return null, signalling that cleanup may proceed safely. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task StaleManifestCleanupGuard_ExpiredBeyondRecoveryWindow_StoreReturnsNull() + { + var store = new HaTestSessionStore(); + + // Lease expired hours ago – well beyond any recovery window. + var session = CreateSession("ha-session-3", "pod-a", DateTime.UtcNow.AddHours(-2)); + await store.SetAsync(session, TestContext.Current.CancellationToken); + + // Controller or cleanup task checks the store before deleting files. + var liveSession = await store.TryGetAsync("ha-session-3", TestContext.Current.CancellationToken); + + // Store returns null → cleanup may proceed without risking data loss. + Assert.Null(liveSession); + } + + /// + /// Segment-length selection: when the play-session has an active entry in the store + /// (HA mode is active), the recovery segment length should be preferred over the normal one. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task SegmentLength_UsesRecoveryValue_WhenHaModeIsActive() + { + const int normalSegmentLength = 6; + const int recoverySegmentLength = 2; + + var store = new HaTestSessionStore(); + var session = CreateSession("ha-session-4", "pod-a", DateTime.UtcNow.AddMinutes(5)); + await store.SetAsync(session, TestContext.Current.CancellationToken); + + // Simulate the controller's HA-mode check: if the session is in the store, HA mode is active. + var existingSession = await store.TryGetAsync("ha-session-4", TestContext.Current.CancellationToken); + var isHaMode = existingSession is not null; + + var effectiveSegmentLength = isHaMode ? recoverySegmentLength : normalSegmentLength; + + Assert.True(isHaMode, "Session should be found in the store, activating HA mode."); + Assert.Equal(recoverySegmentLength, effectiveSegmentLength); + } + + /// + /// Segment-length selection: when no entry exists in the store for the play-session + /// (HA mode inactive), the normal segment length should be used. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task SegmentLength_UsesNormalValue_WhenHaModeIsInactive() + { + const int normalSegmentLength = 6; + const int recoverySegmentLength = 2; + + var store = new HaTestSessionStore(); + + // No session registered – HA mode is inactive. + var existingSession = await store.TryGetAsync("nonexistent-session", TestContext.Current.CancellationToken); + var isHaMode = existingSession is not null; + + var effectiveSegmentLength = isHaMode ? recoverySegmentLength : normalSegmentLength; + + Assert.False(isHaMode, "No session in the store means HA mode should be inactive."); + Assert.Equal(normalSegmentLength, effectiveSegmentLength); + } + + /// + /// Minimal in-memory used within this test class + /// to avoid a cross-project reference to Jellyfin.MediaEncoding.Tests. + /// + private sealed class HaTestSessionStore : ITranscodeSessionStore + { + private static readonly TimeSpan LeaseDuration = TimeSpan.FromSeconds(30); + + private readonly Dictionary _sessions = + new(StringComparer.OrdinalIgnoreCase); + + private readonly Lock _lock = new(); + + public Task TryGetAsync(string playSessionId, CancellationToken cancellationToken = default) + { + lock (_lock) + { + if (_sessions.TryGetValue(playSessionId, out var s) && s.LeaseExpiresUtc > DateTime.UtcNow) + { + return Task.FromResult(Clone(s)); + } + + return Task.FromResult(null); + } + } + + public Task TryTakeoverAsync(string playSessionId, string claimingPod, CancellationToken cancellationToken = default) + { + lock (_lock) + { + if (!_sessions.TryGetValue(playSessionId, out var s)) + { + return Task.FromResult(false); + } + + if (s.LeaseExpiresUtc > DateTime.UtcNow) + { + return Task.FromResult(false); + } + + s.OwnerPod = claimingPod; + s.LeaseExpiresUtc = DateTime.UtcNow.Add(LeaseDuration); + return Task.FromResult(true); + } + } + + public Task SetAsync(TranscodeSession session, CancellationToken cancellationToken = default) + { + lock (_lock) + { + _sessions[session.PlaySessionId] = session; + } + + return Task.CompletedTask; + } + + public Task RenewLeaseAsync(string playSessionId, CancellationToken cancellationToken = default) + { + lock (_lock) + { + if (_sessions.TryGetValue(playSessionId, out var s)) + { + s.LeaseExpiresUtc = DateTime.UtcNow.Add(LeaseDuration); + } + } + + return Task.CompletedTask; + } + + public Task DeleteAsync(string playSessionId, CancellationToken cancellationToken = default) + { + lock (_lock) + { + _sessions.Remove(playSessionId); + } + + return Task.CompletedTask; + } + + public Task> GetActiveSessionsAsync(CancellationToken cancellationToken = default) + { + lock (_lock) + { + var sessions = _sessions.Values + .Where(s => s.LeaseExpiresUtc > DateTime.UtcNow) + .Select(Clone) + .ToList(); + return Task.FromResult>(sessions); + } + } + + public Task SetLiveStreamAsync(LiveStreamSession session, CancellationToken cancellationToken = default) + => Task.CompletedTask; + + public Task TryGetLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default) + => Task.FromResult(null); + + public Task DeleteLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default) + => Task.CompletedTask; + + private static TranscodeSession Clone(TranscodeSession source) + => new TranscodeSession + { + PlaySessionId = source.PlaySessionId, + OwnerPod = source.OwnerPod, + LeaseExpiresUtc = source.LeaseExpiresUtc, + ManifestPath = source.ManifestPath, + SegmentPathPrefix = source.SegmentPathPrefix, + MediaSourceId = source.MediaSourceId, + LastCompletedSegmentIndex = source.LastCompletedSegmentIndex, + LastDurablePlaybackOffset = source.LastDurablePlaybackOffset, + }; + } + } +} diff --git a/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsSessionRegistrationTests.cs b/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsSessionRegistrationTests.cs new file mode 100644 index 0000000000..cb09c00ff5 --- /dev/null +++ b/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsSessionRegistrationTests.cs @@ -0,0 +1,223 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using MediaBrowser.Controller.MediaEncoding; +using Xunit; + +namespace Jellyfin.Api.Tests.Controllers +{ + /// + /// Tests for HLS session registration, lease renewal, and cleanup behaviour wired into + /// in Phase 5.2.2a. + /// These tests verify the contract used by the controller. + /// + public class DynamicHlsSessionRegistrationTests + { + private static TranscodeSession CreateSession(string id, string pod) + => new TranscodeSession + { + PlaySessionId = id, + OwnerPod = pod, + LeaseExpiresUtc = DateTime.UtcNow.AddSeconds(30), + ManifestPath = string.Empty, + SegmentPathPrefix = string.Empty, + MediaSourceId = "media-source-1", + LastCompletedSegmentIndex = 0, + LastDurablePlaybackOffset = 0L, + }; + + /// + /// After registering a session via , + /// must return a non-null result with + /// matching and . + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task SessionRegistration_AfterStreamStart_StoreContainsSession() + { + var store = new InMemoryTranscodeSessionStore(); + var session = CreateSession("session-reg-1", "pod-a"); + + await store.SetAsync(session, TestContext.Current.CancellationToken); + + var retrieved = await store.TryGetAsync("session-reg-1", TestContext.Current.CancellationToken); + + Assert.NotNull(retrieved); + Assert.Equal("session-reg-1", retrieved.PlaySessionId); + Assert.Equal("pod-a", retrieved.OwnerPod); + } + + /// + /// After calling , + /// must return null. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task SessionCleanup_AfterStreamEnd_StoreReturnsNull() + { + var store = new InMemoryTranscodeSessionStore(); + var session = CreateSession("session-cleanup-1", "pod-b"); + + await store.SetAsync(session, TestContext.Current.CancellationToken); + await store.DeleteAsync("session-cleanup-1", TestContext.Current.CancellationToken); + + var retrieved = await store.TryGetAsync("session-cleanup-1", TestContext.Current.CancellationToken); + + Assert.Null(retrieved); + } + + /// + /// After a session's initial lease window would have expired, calling + /// must extend the lease so that + /// still returns the session as active. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task LeaseRenewal_ExtendsBeyondInitialExpiry() + { + var store = new InMemoryTranscodeSessionStore(); + + // Create the session with a lease that has already expired. + var session = new TranscodeSession + { + PlaySessionId = "session-renewal-1", + OwnerPod = "pod-c", + LeaseExpiresUtc = DateTime.UtcNow.AddMilliseconds(-1), + ManifestPath = string.Empty, + SegmentPathPrefix = string.Empty, + MediaSourceId = "media-source-1", + LastCompletedSegmentIndex = 0, + LastDurablePlaybackOffset = 0L, + }; + await store.SetAsync(session, TestContext.Current.CancellationToken); + + // Verify the session is not accessible because the lease has expired. + Assert.Null(await store.TryGetAsync("session-renewal-1", TestContext.Current.CancellationToken)); + + // Renew the lease. + await store.RenewLeaseAsync("session-renewal-1", TestContext.Current.CancellationToken); + + // After renewal the session must be accessible again. + var renewed = await store.TryGetAsync("session-renewal-1", TestContext.Current.CancellationToken); + Assert.NotNull(renewed); + Assert.Equal("session-renewal-1", renewed.PlaySessionId); + Assert.True(renewed.LeaseExpiresUtc > DateTime.UtcNow); + } + + /// + /// Minimal thread-safe in-memory implementation of + /// used within this test class to avoid a cross-project reference. + /// + private sealed class InMemoryTranscodeSessionStore : ITranscodeSessionStore + { + private static readonly TimeSpan DefaultLeaseDuration = TimeSpan.FromSeconds(30); + + private readonly Dictionary _sessions = + new(StringComparer.OrdinalIgnoreCase); + + private readonly Lock _lock = new(); + + public Task TryGetAsync(string playSessionId, CancellationToken cancellationToken = default) + { + lock (_lock) + { + if (_sessions.TryGetValue(playSessionId, out var session) && session.LeaseExpiresUtc > DateTime.UtcNow) + { + return Task.FromResult(Clone(session)); + } + + return Task.FromResult(null); + } + } + + public Task TryTakeoverAsync(string playSessionId, string claimingPod, CancellationToken cancellationToken = default) + { + lock (_lock) + { + if (!_sessions.TryGetValue(playSessionId, out var session)) + { + return Task.FromResult(false); + } + + if (session.LeaseExpiresUtc > DateTime.UtcNow) + { + return Task.FromResult(false); + } + + session.OwnerPod = claimingPod; + session.LeaseExpiresUtc = DateTime.UtcNow.Add(DefaultLeaseDuration); + return Task.FromResult(true); + } + } + + public Task SetAsync(TranscodeSession session, CancellationToken cancellationToken = default) + { + lock (_lock) + { + _sessions[session.PlaySessionId] = session; + } + + return Task.CompletedTask; + } + + public Task RenewLeaseAsync(string playSessionId, CancellationToken cancellationToken = default) + { + lock (_lock) + { + if (_sessions.TryGetValue(playSessionId, out var session)) + { + session.LeaseExpiresUtc = DateTime.UtcNow.Add(DefaultLeaseDuration); + } + } + + return Task.CompletedTask; + } + + public Task DeleteAsync(string playSessionId, CancellationToken cancellationToken = default) + { + lock (_lock) + { + _sessions.Remove(playSessionId); + } + + return Task.CompletedTask; + } + + public Task> GetActiveSessionsAsync(CancellationToken cancellationToken = default) + { + lock (_lock) + { + var sessions = _sessions.Values + .Where(s => s.LeaseExpiresUtc > DateTime.UtcNow) + .Select(Clone) + .ToList(); + return Task.FromResult>(sessions); + } + } + + public Task SetLiveStreamAsync(LiveStreamSession session, CancellationToken cancellationToken = default) + => Task.CompletedTask; + + public Task TryGetLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default) + => Task.FromResult(null); + + public Task DeleteLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default) + => Task.CompletedTask; + + private static TranscodeSession Clone(TranscodeSession source) + => new TranscodeSession + { + PlaySessionId = source.PlaySessionId, + OwnerPod = source.OwnerPod, + LeaseExpiresUtc = source.LeaseExpiresUtc, + ManifestPath = source.ManifestPath, + SegmentPathPrefix = source.SegmentPathPrefix, + MediaSourceId = source.MediaSourceId, + LastCompletedSegmentIndex = source.LastCompletedSegmentIndex, + LastDurablePlaybackOffset = source.LastDurablePlaybackOffset, + }; + } + } +}