Phase 5.2.2a: Register HLS sessions in ITranscodeSessionStore + lease renewal (#23)

* Initial plan * Phase 5.2.2a: Register HLS sessions in ITranscodeSessionStore + lease renewal Co-authored-by: ZoltyMat <177592743+ZoltyMat@users.noreply.github.com> --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: ZoltyMat <177592743+ZoltyMat@users.noreply.github.com>
2026-03-09 23:50:10 -04:00
parent d60ae43b59
commit c2a11f3e68
2 changed files with 289 additions and 1 deletions
@@ -60,6 +60,7 @@ public class DynamicHlsController : BaseJellyfinApiController
    private readonly IDynamicHlsPlaylistGenerator _dynamicHlsPlaylistGenerator;
    private readonly DynamicHlsHelper _dynamicHlsHelper;
    private readonly EncodingOptions _encodingOptions;
+    private readonly ITranscodeSessionStore _transcodeSessionStore;

    /// <summary>
    /// Initializes a new instance of the <see cref="DynamicHlsController"/> class.
@@ -75,6 +76,7 @@ public class DynamicHlsController : BaseJellyfinApiController
    /// <param name="dynamicHlsHelper">Instance of <see cref="DynamicHlsHelper"/>.</param>
    /// <param name="encodingHelper">Instance of <see cref="EncodingHelper"/>.</param>
    /// <param name="dynamicHlsPlaylistGenerator">Instance of <see cref="IDynamicHlsPlaylistGenerator"/>.</param>
+    /// <param name="transcodeSessionStore">Instance of the <see cref="ITranscodeSessionStore"/> interface used to register and renew HLS transcoding session leases in the durable store.</param>
    public DynamicHlsController(
        ILibraryManager libraryManager,
        IUserManager userManager,
@@ -86,7 +88,8 @@ public class DynamicHlsController : BaseJellyfinApiController
        ILogger<DynamicHlsController> 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();
    }
@@ -318,6 +322,12 @@ public class DynamicHlsController : BaseJellyfinApiController
                                cancellationTokenSource)
                            .ConfigureAwait(false);
                        job.IsLiveOutput = true;
+                        await RegisterTranscodeSessionAsync(
+                                playSessionId ?? string.Empty,
+                                mediaSourceId ?? string.Empty,
+                                cancellationToken)
+                            .ConfigureAwait(false);
+                        StartLeaseRenewal(playSessionId ?? string.Empty, cancellationToken);
                    }
                    catch
                    {
@@ -1543,6 +1553,12 @@ public class DynamicHlsController : BaseJellyfinApiController
                        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
                {
@@ -1570,6 +1586,77 @@ public class DynamicHlsController : BaseJellyfinApiController
    private static double[] GetSegmentLengths(StreamState state)
        => GetSegmentLengthsInternal(state.RunTimeTicks ?? 0, state.SegmentLength);

+    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;
@@ -0,0 +1,201 @@
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using MediaBrowser.Controller.MediaEncoding;
+using Xunit;
+
+namespace Jellyfin.Api.Tests.Controllers
+{
+    /// <summary>
+    /// Tests for HLS session registration, lease renewal, and cleanup behaviour wired into
+    /// <see cref="Jellyfin.Api.Controllers.DynamicHlsController"/> in Phase 5.2.2a.
+    /// These tests verify the <see cref="ITranscodeSessionStore"/> contract used by the controller.
+    /// </summary>
+    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,
+            };
+
+        /// <summary>
+        /// After registering a session via <see cref="ITranscodeSessionStore.SetAsync"/>,
+        /// <see cref="ITranscodeSessionStore.TryGetAsync"/> must return a non-null result with
+        /// matching <see cref="TranscodeSession.PlaySessionId"/> and <see cref="TranscodeSession.OwnerPod"/>.
+        /// </summary>
+        [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);
+
+            var retrieved = await store.TryGetAsync("session-reg-1");
+
+            Assert.NotNull(retrieved);
+            Assert.Equal("session-reg-1", retrieved.PlaySessionId);
+            Assert.Equal("pod-a", retrieved.OwnerPod);
+        }
+
+        /// <summary>
+        /// After calling <see cref="ITranscodeSessionStore.DeleteAsync"/>,
+        /// <see cref="ITranscodeSessionStore.TryGetAsync"/> must return <c>null</c>.
+        /// </summary>
+        [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);
+            await store.DeleteAsync("session-cleanup-1");
+
+            var retrieved = await store.TryGetAsync("session-cleanup-1");
+
+            Assert.Null(retrieved);
+        }
+
+        /// <summary>
+        /// After a session's initial lease window would have expired, calling
+        /// <see cref="ITranscodeSessionStore.RenewLeaseAsync"/> must extend the lease so that
+        /// <see cref="ITranscodeSessionStore.TryGetAsync"/> still returns the session as active.
+        /// </summary>
+        [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);
+
+            // Verify the session is not accessible because the lease has expired.
+            Assert.Null(await store.TryGetAsync("session-renewal-1"));
+
+            // Renew the lease.
+            await store.RenewLeaseAsync("session-renewal-1");
+
+            // After renewal the session must be accessible again.
+            var renewed = await store.TryGetAsync("session-renewal-1");
+            Assert.NotNull(renewed);
+            Assert.Equal("session-renewal-1", renewed.PlaySessionId);
+            Assert.True(renewed.LeaseExpiresUtc > DateTime.UtcNow);
+        }
+
+        /// <summary>
+        /// Minimal thread-safe in-memory implementation of <see cref="ITranscodeSessionStore"/>
+        /// used within this test class to avoid a cross-project reference.
+        /// </summary>
+        private sealed class InMemoryTranscodeSessionStore : ITranscodeSessionStore
+        {
+            private static readonly TimeSpan DefaultLeaseDuration = TimeSpan.FromSeconds(30);
+
+            private readonly Dictionary<string, TranscodeSession> _sessions =
+                new(StringComparer.OrdinalIgnoreCase);
+
+            private readonly Lock _lock = new();
+
+            public Task<TranscodeSession?> TryGetAsync(string playSessionId, CancellationToken cancellationToken = default)
+            {
+                lock (_lock)
+                {
+                    if (_sessions.TryGetValue(playSessionId, out var session) && session.LeaseExpiresUtc > DateTime.UtcNow)
+                    {
+                        return Task.FromResult<TranscodeSession?>(Clone(session));
+                    }
+
+                    return Task.FromResult<TranscodeSession?>(null);
+                }
+            }
+
+            public Task<bool> 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;
+            }
+
+            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,
+                };
+        }
+    }
+}