diff --git a/Directory.Packages.props b/Directory.Packages.props index 7508a5a86..efe9b4164 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -95,6 +95,7 @@ + diff --git a/Emby.Server.Implementations/Emby.Server.Implementations.csproj b/Emby.Server.Implementations/Emby.Server.Implementations.csproj index 15843730e..abdb0679b 100644 --- a/Emby.Server.Implementations/Emby.Server.Implementations.csproj +++ b/Emby.Server.Implementations/Emby.Server.Implementations.csproj @@ -66,6 +66,7 @@ + diff --git a/Emby.Server.Implementations/MediaEncoding/RedisTranscodeSessionStore.cs b/Emby.Server.Implementations/MediaEncoding/RedisTranscodeSessionStore.cs new file mode 100644 index 000000000..bd82faf49 --- /dev/null +++ b/Emby.Server.Implementations/MediaEncoding/RedisTranscodeSessionStore.cs @@ -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; + +/// +/// A Redis-backed implementation of that provides +/// durable, distributed session tracking with lease-based ownership between pods. +/// +public sealed class RedisTranscodeSessionStore : ITranscodeSessionStore +{ + private const string KeyPrefix = "jellyfin:transcode:"; + + /// + /// Lua script for atomic takeover: reads the stored session, checks whether the lease has + /// expired (comparing LeaseExpiresUtc.Ticks against the caller-supplied current ticks), + /// and if expired, updates the owner and expiry before returning 1; returns 0 otherwise. + /// + 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 _logger; + + /// + /// Initializes a new instance of the class. + /// + /// The Redis connection multiplexer. + /// The transcode store configuration options. + /// The logger. + public RedisTranscodeSessionStore( + IConnectionMultiplexer redis, + IOptions options, + ILogger logger) + { + _db = redis.GetDatabase(); + _options = options.Value; + _logger = logger; + } + + /// + 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); + } + + /// + public async Task 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(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; + } + + /// + 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(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); + } + + /// + 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); + } + + /// + public async Task 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; +} diff --git a/Jellyfin.Server/CoreAppHost.cs b/Jellyfin.Server/CoreAppHost.cs index 2548ddea7..6020a1bc3 100644 --- a/Jellyfin.Server/CoreAppHost.cs +++ b/Jellyfin.Server/CoreAppHost.cs @@ -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 /// public class CoreAppHost : ApplicationHost { + private readonly IConfiguration _startupConfig; + /// /// Initializes a new instance of the class. /// @@ -57,6 +62,7 @@ namespace Jellyfin.Server options, startupConfig) { + _startupConfig = startupConfig; } /// @@ -98,6 +104,31 @@ namespace Jellyfin.Server serviceCollection.AddScoped(); + // Transcode session store: Redis-backed when configured, no-op otherwise. + serviceCollection.Configure(_startupConfig.GetSection("Jellyfin:TranscodeStore")); + var redisConnectionString = _startupConfig["Jellyfin:TranscodeStore:RedisConnectionString"]; + if (!string.IsNullOrEmpty(redisConnectionString)) + { + serviceCollection.AddSingleton(sp => + { + try + { + return ConnectionMultiplexer.Connect(redisConnectionString); + } + catch (Exception ex) + { + sp.GetRequiredService>() + .LogError(ex, "Failed to connect to Redis. Check the Jellyfin:TranscodeStore:RedisConnectionString configuration."); + throw; + } + }); + serviceCollection.AddSingleton(); + } + else + { + serviceCollection.AddSingleton(); + } + foreach (var type in GetExportTypes()) { serviceCollection.AddSingleton(typeof(ILyricProvider), type); diff --git a/Jellyfin.Server/Jellyfin.Server.csproj b/Jellyfin.Server/Jellyfin.Server.csproj index 14ab114fb..4d20655b0 100644 --- a/Jellyfin.Server/Jellyfin.Server.csproj +++ b/Jellyfin.Server/Jellyfin.Server.csproj @@ -59,6 +59,7 @@ + diff --git a/MediaBrowser.Controller/MediaEncoding/NullTranscodeSessionStore.cs b/MediaBrowser.Controller/MediaEncoding/NullTranscodeSessionStore.cs new file mode 100644 index 000000000..e7dcdd1d1 --- /dev/null +++ b/MediaBrowser.Controller/MediaEncoding/NullTranscodeSessionStore.cs @@ -0,0 +1,31 @@ +using System.Threading; +using System.Threading.Tasks; + +namespace MediaBrowser.Controller.MediaEncoding; + +/// +/// A no-op implementation of used in single-instance deployments +/// where durable session tracking across pods is not required. +/// +public sealed class NullTranscodeSessionStore : ITranscodeSessionStore +{ + /// + public Task TryGetAsync(string playSessionId, CancellationToken cancellationToken = default) + => Task.FromResult(null); + + /// + public Task TryTakeoverAsync(string playSessionId, string claimingPod, CancellationToken cancellationToken = default) + => Task.FromResult(false); + + /// + public Task SetAsync(TranscodeSession session, CancellationToken cancellationToken = default) + => Task.CompletedTask; + + /// + public Task RenewLeaseAsync(string playSessionId, CancellationToken cancellationToken = default) + => Task.CompletedTask; + + /// + public Task DeleteAsync(string playSessionId, CancellationToken cancellationToken = default) + => Task.CompletedTask; +} diff --git a/MediaBrowser.Controller/MediaEncoding/TranscodeStoreOptions.cs b/MediaBrowser.Controller/MediaEncoding/TranscodeStoreOptions.cs new file mode 100644 index 000000000..23d457bd2 --- /dev/null +++ b/MediaBrowser.Controller/MediaEncoding/TranscodeStoreOptions.cs @@ -0,0 +1,19 @@ +namespace MediaBrowser.Controller.MediaEncoding; + +/// +/// Configuration options for the transcode session store. +/// +public sealed class TranscodeStoreOptions +{ + /// + /// Gets or sets the Redis connection string. + /// A null or empty value indicates single-instance mode, where + /// is used instead of a Redis-backed store. + /// + public string? RedisConnectionString { get; set; } + + /// + /// Gets or sets the duration in seconds for which a transcoding session lease is valid. + /// + public int LeaseDurationSeconds { get; set; } = 30; +} diff --git a/tests/Jellyfin.Server.Implementations.Tests/MediaEncoding/RedisTranscodeSessionStoreTests.cs b/tests/Jellyfin.Server.Implementations.Tests/MediaEncoding/RedisTranscodeSessionStoreTests.cs new file mode 100644 index 000000000..3bc51b002 --- /dev/null +++ b/tests/Jellyfin.Server.Implementations.Tests/MediaEncoding/RedisTranscodeSessionStoreTests.cs @@ -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; + +/// +/// Tests for transcode session store contract behavior, using +/// as a reference implementation (no real Redis required). +/// +public class RedisTranscodeSessionStoreTests +{ + /// + /// Verifies that returns null after + /// a session's lease has expired. + /// + [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); + } + + /// + /// Verifies that returns false + /// when the session's lease is still valid. + /// + [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); + } + + /// + /// Verifies that returns true + /// and updates the owner when the session's lease has expired. + /// + [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); + } + + /// + /// Verifies that when multiple pods concurrently attempt to take over an expired session, + /// exactly one succeeds. + /// + [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[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); + } + + /// + /// Thread-safe, in-memory implementation of used within + /// this test class to avoid a cross-project reference to Jellyfin.MediaEncoding.Tests. + /// + 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; + } + + 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, + }; + } +}