Wire ITranscodeSessionStore to Redis-backed impl with NullTranscodeSessionStore fallback and DI registration (#21)

* Initial plan * feat: add Redis-backed ITranscodeSessionStore with NullTranscodeSessionStore fallback and DI registration Co-authored-by: ZoltyMat <177592743+ZoltyMat@users.noreply.github.com> * refactor: add code review improvements - lease expiry comment, Redis connection error handling 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:04:04 -04:00
parent a187ab18b8
commit d60ae43b59
8 changed files with 452 additions and 0 deletions
@@ -95,6 +95,7 @@
    <PackageVersion Include="z440.atl.core" Version="7.9.0" />
    <PackageVersion Include="TMDbLib" Version="2.3.0" />
    <PackageVersion Include="UTF.Unknown" Version="2.6.0" />
+    <PackageVersion Include="StackExchange.Redis" Version="2.8.16" />
    <PackageVersion Include="Xunit.Priority" Version="1.1.6" />
    <PackageVersion Include="xunit.runner.visualstudio" Version="2.8.2" />
    <PackageVersion Include="Xunit.SkippableFact" Version="1.5.23" />
@@ -66,6 +66,7 @@

  <ItemGroup>
    <PackageReference Include="Ignore" />
+    <PackageReference Include="StackExchange.Redis" />
  </ItemGroup>

  <ItemGroup>
@@ -0,0 +1,143 @@
+using System;
+using System.Text.Json;
+using System.Threading;
+using System.Threading.Tasks;
+using MediaBrowser.Controller.MediaEncoding;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using StackExchange.Redis;
+
+namespace Emby.Server.Implementations.MediaEncoding;
+
+/// <summary>
+/// A Redis-backed implementation of <see cref="ITranscodeSessionStore"/> that provides
+/// durable, distributed session tracking with lease-based ownership between pods.
+/// </summary>
+public sealed class RedisTranscodeSessionStore : ITranscodeSessionStore
+{
+    private const string KeyPrefix = "jellyfin:transcode:";
+
+    /// <summary>
+    /// Lua script for atomic takeover: reads the stored session, checks whether the lease has
+    /// expired (comparing <c>LeaseExpiresUtc.Ticks</c> against the caller-supplied current ticks),
+    /// and if expired, updates the owner and expiry before returning 1; returns 0 otherwise.
+    /// </summary>
+    private const string TakeoverScript = @"
+local raw = redis.call('GET', KEYS[1])
+if not raw then return 0 end
+local session = cjson.decode(raw)
+local currentTicks = tonumber(ARGV[1])
+if session['LeaseExpiresUtc'] > currentTicks then return 0 end
+session['OwnerPod'] = ARGV[2]
+local leaseDurationMs = tonumber(ARGV[3])
+local newTicks = currentTicks + (leaseDurationMs * 10000)
+session['LeaseExpiresUtc'] = newTicks
+redis.call('SET', KEYS[1], cjson.encode(session), 'PX', leaseDurationMs)
+return 1";
+
+    private readonly IDatabase _db;
+    private readonly TranscodeStoreOptions _options;
+    private readonly ILogger<RedisTranscodeSessionStore> _logger;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="RedisTranscodeSessionStore"/> class.
+    /// </summary>
+    /// <param name="redis">The Redis connection multiplexer.</param>
+    /// <param name="options">The transcode store configuration options.</param>
+    /// <param name="logger">The logger.</param>
+    public RedisTranscodeSessionStore(
+        IConnectionMultiplexer redis,
+        IOptions<TranscodeStoreOptions> options,
+        ILogger<RedisTranscodeSessionStore> logger)
+    {
+        _db = redis.GetDatabase();
+        _options = options.Value;
+        _logger = logger;
+    }
+
+    /// <inheritdoc />
+    public async Task SetAsync(TranscodeSession session, CancellationToken cancellationToken = default)
+    {
+        var key = GetKey(session.PlaySessionId);
+        var json = JsonSerializer.Serialize(session);
+        var leaseDurationMs = (long)_options.LeaseDurationSeconds * 1000;
+        await _db.StringSetAsync(key, json, TimeSpan.FromMilliseconds(leaseDurationMs)).ConfigureAwait(false);
+        _logger.LogDebug("Set transcode session {PlaySessionId} in Redis.", session.PlaySessionId);
+    }
+
+    /// <inheritdoc />
+    public async Task<TranscodeSession?> TryGetAsync(string playSessionId, CancellationToken cancellationToken = default)
+    {
+        var key = GetKey(playSessionId);
+        var raw = await _db.StringGetAsync(key).ConfigureAwait(false);
+        if (!raw.HasValue)
+        {
+            return null;
+        }
+
+        var session = JsonSerializer.Deserialize<TranscodeSession>(raw.ToString());
+
+        // Check LeaseExpiresUtc in addition to Redis TTL to guard against the window between
+        // Redis TTL evaluation and the GET result being returned to the caller.
+        if (session is null || session.LeaseExpiresUtc <= DateTime.UtcNow)
+        {
+            return null;
+        }
+
+        return session;
+    }
+
+    /// <inheritdoc />
+    public async Task RenewLeaseAsync(string playSessionId, CancellationToken cancellationToken = default)
+    {
+        var key = GetKey(playSessionId);
+        var raw = await _db.StringGetAsync(key).ConfigureAwait(false);
+        if (!raw.HasValue)
+        {
+            return;
+        }
+
+        var session = JsonSerializer.Deserialize<TranscodeSession>(raw.ToString());
+        if (session is null)
+        {
+            return;
+        }
+
+        var leaseDurationMs = (long)_options.LeaseDurationSeconds * 1000;
+        session.LeaseExpiresUtc = DateTime.UtcNow.AddMilliseconds(leaseDurationMs);
+        var json = JsonSerializer.Serialize(session);
+        await _db.StringSetAsync(key, json, TimeSpan.FromMilliseconds(leaseDurationMs)).ConfigureAwait(false);
+        _logger.LogDebug("Renewed lease for transcode session {PlaySessionId}.", playSessionId);
+    }
+
+    /// <inheritdoc />
+    public async Task DeleteAsync(string playSessionId, CancellationToken cancellationToken = default)
+    {
+        var key = GetKey(playSessionId);
+        await _db.KeyDeleteAsync(key).ConfigureAwait(false);
+        _logger.LogDebug("Deleted transcode session {PlaySessionId} from Redis.", playSessionId);
+    }
+
+    /// <inheritdoc />
+    public async Task<bool> TryTakeoverAsync(string playSessionId, string claimingPod, CancellationToken cancellationToken = default)
+    {
+        var key = GetKey(playSessionId);
+        var leaseDurationMs = (long)_options.LeaseDurationSeconds * 1000;
+        var currentTicks = DateTime.UtcNow.Ticks;
+
+        var result = (long?)await _db.ScriptEvaluateAsync(
+            TakeoverScript,
+            keys: new RedisKey[] { key },
+            values: new RedisValue[] { currentTicks, claimingPod, leaseDurationMs }).ConfigureAwait(false);
+
+        var succeeded = result == 1;
+        if (succeeded)
+        {
+            _logger.LogInformation("Pod {ClaimingPod} successfully took over transcode session {PlaySessionId}.", claimingPod, playSessionId);
+        }
+
+        return succeeded;
+    }
+
+    private static string GetKey(string playSessionId) => KeyPrefix + playSessionId;
+}
@@ -2,6 +2,7 @@ using System;
 using System.Collections.Generic;
 using System.Reflection;
 using Emby.Server.Implementations;
+using Emby.Server.Implementations.MediaEncoding;
 using Emby.Server.Implementations.Session;
 using Jellyfin.Api.WebSocketListeners;
 using Jellyfin.Database.Implementations;
@@ -23,6 +24,7 @@ using MediaBrowser.Controller.Drawing;
 using MediaBrowser.Controller.Events;
 using MediaBrowser.Controller.Library;
 using MediaBrowser.Controller.Lyrics;
+using MediaBrowser.Controller.MediaEncoding;
 using MediaBrowser.Controller.Net;
 using MediaBrowser.Controller.Security;
 using MediaBrowser.Controller.Trickplay;
@@ -31,6 +33,7 @@ using MediaBrowser.Providers.Lyric;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
+using StackExchange.Redis;

 namespace Jellyfin.Server
 {
@@ -39,6 +42,8 @@ namespace Jellyfin.Server
    /// </summary>
    public class CoreAppHost : ApplicationHost
    {
+        private readonly IConfiguration _startupConfig;
+
        /// <summary>
        /// Initializes a new instance of the <see cref="CoreAppHost" /> class.
        /// </summary>
@@ -57,6 +62,7 @@ namespace Jellyfin.Server
                options,
                startupConfig)
        {
+            _startupConfig = startupConfig;
        }

        /// <inheritdoc/>
@@ -98,6 +104,31 @@ namespace Jellyfin.Server

            serviceCollection.AddScoped<IAuthenticationManager, AuthenticationManager>();

+            // Transcode session store: Redis-backed when configured, no-op otherwise.
+            serviceCollection.Configure<TranscodeStoreOptions>(_startupConfig.GetSection("Jellyfin:TranscodeStore"));
+            var redisConnectionString = _startupConfig["Jellyfin:TranscodeStore:RedisConnectionString"];
+            if (!string.IsNullOrEmpty(redisConnectionString))
+            {
+                serviceCollection.AddSingleton<IConnectionMultiplexer>(sp =>
+                {
+                    try
+                    {
+                        return ConnectionMultiplexer.Connect(redisConnectionString);
+                    }
+                    catch (Exception ex)
+                    {
+                        sp.GetRequiredService<ILogger<CoreAppHost>>()
+                            .LogError(ex, "Failed to connect to Redis. Check the Jellyfin:TranscodeStore:RedisConnectionString configuration.");
+                        throw;
+                    }
+                });
+                serviceCollection.AddSingleton<ITranscodeSessionStore, RedisTranscodeSessionStore>();
+            }
+            else
+            {
+                serviceCollection.AddSingleton<ITranscodeSessionStore, NullTranscodeSessionStore>();
+            }
+
            foreach (var type in GetExportTypes<ILyricProvider>())
            {
                serviceCollection.AddSingleton(typeof(ILyricProvider), type);
@@ -59,6 +59,7 @@
    <PackageReference Include="Serilog.Sinks.Console" />
    <PackageReference Include="Serilog.Sinks.File" />
    <PackageReference Include="Serilog.Sinks.Graylog" />
+    <PackageReference Include="StackExchange.Redis" />
  </ItemGroup>

  <ItemGroup>
@@ -0,0 +1,31 @@
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MediaBrowser.Controller.MediaEncoding;
+
+/// <summary>
+/// A no-op implementation of <see cref="ITranscodeSessionStore"/> used in single-instance deployments
+/// where durable session tracking across pods is not required.
+/// </summary>
+public sealed class NullTranscodeSessionStore : ITranscodeSessionStore
+{
+    /// <inheritdoc />
+    public Task<TranscodeSession?> TryGetAsync(string playSessionId, CancellationToken cancellationToken = default)
+        => Task.FromResult<TranscodeSession?>(null);
+
+    /// <inheritdoc />
+    public Task<bool> TryTakeoverAsync(string playSessionId, string claimingPod, CancellationToken cancellationToken = default)
+        => Task.FromResult(false);
+
+    /// <inheritdoc />
+    public Task SetAsync(TranscodeSession session, CancellationToken cancellationToken = default)
+        => Task.CompletedTask;
+
+    /// <inheritdoc />
+    public Task RenewLeaseAsync(string playSessionId, CancellationToken cancellationToken = default)
+        => Task.CompletedTask;
+
+    /// <inheritdoc />
+    public Task DeleteAsync(string playSessionId, CancellationToken cancellationToken = default)
+        => Task.CompletedTask;
+}
@@ -0,0 +1,19 @@
+namespace MediaBrowser.Controller.MediaEncoding;
+
+/// <summary>
+/// Configuration options for the transcode session store.
+/// </summary>
+public sealed class TranscodeStoreOptions
+{
+    /// <summary>
+    /// Gets or sets the Redis connection string.
+    /// A <c>null</c> or empty value indicates single-instance mode, where
+    /// <see cref="NullTranscodeSessionStore"/> is used instead of a Redis-backed store.
+    /// </summary>
+    public string? RedisConnectionString { get; set; }
+
+    /// <summary>
+    /// Gets or sets the duration in seconds for which a transcoding session lease is valid.
+    /// </summary>
+    public int LeaseDurationSeconds { get; set; } = 30;
+}
@@ -0,0 +1,225 @@
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using MediaBrowser.Controller.MediaEncoding;
+using Xunit;
+
+namespace Jellyfin.Server.Implementations.Tests.MediaEncoding;
+
+/// <summary>
+/// Tests for transcode session store contract behavior, using <see cref="InMemoryTranscodeSessionStore"/>
+/// as a reference implementation (no real Redis required).
+/// </summary>
+public class RedisTranscodeSessionStoreTests
+{
+    /// <summary>
+    /// Verifies that <see cref="ITranscodeSessionStore.TryGetAsync"/> returns <c>null</c> after
+    /// a session's lease has expired.
+    /// </summary>
+    [Fact]
+    [Trait("Category", "UnitTest")]
+    public async Task TryGetAsync_AfterLeaseExpires_ReturnsNull()
+    {
+        var store = new InMemoryTranscodeSessionStore();
+        var session = new TranscodeSession
+        {
+            PlaySessionId = "session-1",
+            OwnerPod = "pod-a",
+            LeaseExpiresUtc = DateTime.UtcNow.AddSeconds(-1),
+        };
+
+        await store.SetAsync(session);
+
+        var result = await store.TryGetAsync("session-1");
+
+        Assert.Null(result);
+    }
+
+    /// <summary>
+    /// Verifies that <see cref="ITranscodeSessionStore.TryTakeoverAsync"/> returns <c>false</c>
+    /// when the session's lease is still valid.
+    /// </summary>
+    [Fact]
+    [Trait("Category", "UnitTest")]
+    public async Task TryTakeoverAsync_WhileLeaseValid_ReturnsFalse()
+    {
+        var store = new InMemoryTranscodeSessionStore();
+        var session = new TranscodeSession
+        {
+            PlaySessionId = "session-2",
+            OwnerPod = "pod-a",
+            LeaseExpiresUtc = DateTime.UtcNow.AddSeconds(30),
+        };
+
+        await store.SetAsync(session);
+
+        var result = await store.TryTakeoverAsync("session-2", "pod-b");
+
+        Assert.False(result);
+    }
+
+    /// <summary>
+    /// Verifies that <see cref="ITranscodeSessionStore.TryTakeoverAsync"/> returns <c>true</c>
+    /// and updates the owner when the session's lease has expired.
+    /// </summary>
+    [Fact]
+    [Trait("Category", "UnitTest")]
+    public async Task TryTakeoverAsync_AfterLeaseExpires_ReturnsTrue_AndUpdatesOwner()
+    {
+        var store = new InMemoryTranscodeSessionStore();
+        var session = new TranscodeSession
+        {
+            PlaySessionId = "session-3",
+            OwnerPod = "pod-a",
+            LeaseExpiresUtc = DateTime.UtcNow.AddSeconds(-1),
+        };
+
+        await store.SetAsync(session);
+
+        var result = await store.TryTakeoverAsync("session-3", "pod-b");
+
+        Assert.True(result);
+
+        var updated = await store.TryGetAsync("session-3");
+        Assert.NotNull(updated);
+        Assert.Equal("pod-b", updated.OwnerPod);
+        Assert.True(updated.LeaseExpiresUtc > DateTime.UtcNow);
+    }
+
+    /// <summary>
+    /// Verifies that when multiple pods concurrently attempt to take over an expired session,
+    /// exactly one succeeds.
+    /// </summary>
+    [Fact]
+    [Trait("Category", "UnitTest")]
+    public async Task ConcurrentTryTakeover_OnlyOneWins()
+    {
+        var store = new InMemoryTranscodeSessionStore();
+        var session = new TranscodeSession
+        {
+            PlaySessionId = "session-4",
+            OwnerPod = "pod-a",
+            LeaseExpiresUtc = DateTime.UtcNow.AddSeconds(-1),
+        };
+
+        await store.SetAsync(session);
+
+        const int concurrency = 10;
+        var tasks = new Task<bool>[concurrency];
+        for (int i = 0; i < concurrency; i++)
+        {
+            var podName = $"pod-{i}";
+            tasks[i] = store.TryTakeoverAsync("session-4", podName);
+        }
+
+        var results = await Task.WhenAll(tasks);
+
+        var successCount = 0;
+        foreach (var r in results)
+        {
+            if (r)
+            {
+                successCount++;
+            }
+        }
+
+        Assert.Equal(1, successCount);
+    }
+
+    /// <summary>
+    /// Thread-safe, in-memory implementation of <see cref="ITranscodeSessionStore"/> used within
+    /// this test class to avoid a cross-project reference to Jellyfin.MediaEncoding.Tests.
+    /// </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();
+
+        /// <inheritdoc />
+        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);
+            }
+        }
+
+        /// <inheritdoc />
+        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);
+            }
+        }
+
+        /// <inheritdoc />
+        public Task SetAsync(TranscodeSession session, CancellationToken cancellationToken = default)
+        {
+            lock (_lock)
+            {
+                _sessions[session.PlaySessionId] = session;
+            }
+
+            return Task.CompletedTask;
+        }
+
+        /// <inheritdoc />
+        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;
+        }
+
+        /// <inheritdoc />
+        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,
+            };
+    }
+}