diff --git a/Directory.Packages.props b/Directory.Packages.props index a00f24b1a7..1732f575a5 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -78,6 +78,7 @@ + diff --git a/Emby.Server.Implementations/Emby.Server.Implementations.csproj b/Emby.Server.Implementations/Emby.Server.Implementations.csproj index f312fb4db9..f0a5f1e2c3 100644 --- a/Emby.Server.Implementations/Emby.Server.Implementations.csproj +++ b/Emby.Server.Implementations/Emby.Server.Implementations.csproj @@ -65,6 +65,7 @@ + diff --git a/Emby.Server.Implementations/MediaEncoding/RedisTranscodeSessionStore.cs b/Emby.Server.Implementations/MediaEncoding/RedisTranscodeSessionStore.cs new file mode 100644 index 0000000000..6ac5c01205 --- /dev/null +++ b/Emby.Server.Implementations/MediaEncoding/RedisTranscodeSessionStore.cs @@ -0,0 +1,270 @@ +using System; +using System.Collections.Generic; +using System.Linq; +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:"; + private const string LiveStreamKeyPrefix = "jellyfin:livestream:"; + + /// + /// 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 IConnectionMultiplexer _redis; + 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) + { + _redis = redis; + _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; + + private static string GetLiveStreamKey(string liveStreamId, string sessionIdOrPlaySessionId) + => LiveStreamKeyPrefix + liveStreamId + ":" + sessionIdOrPlaySessionId; + + /// + public async Task> GetActiveSessionsAsync(CancellationToken cancellationToken = default) + { + var sessions = new List(); + var servers = _redis.GetServers(); + + foreach (var server in servers) + { + if (!server.IsConnected) + { + continue; + } + + var keys = new List(); + await foreach (var key in server.KeysAsync(database: _db.Database, pattern: KeyPrefix + "*", pageSize: 1000).WithCancellation(cancellationToken).ConfigureAwait(false)) + { + keys.Add(key); + } + + var tasks = keys.Select(key => _db.StringGetAsync(key)).ToList(); + var values = await Task.WhenAll(tasks).ConfigureAwait(false); + + foreach (var raw in values) + { + if (!raw.HasValue) + { + continue; + } + + TranscodeSession? session; + try + { + session = JsonSerializer.Deserialize(raw.ToString()); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Failed to deserialize transcode session from Redis."); + continue; + } + + if (session is not null) + { + sessions.Add(session); + } + } + } + + return sessions; + } + + /// + public async Task SetLiveStreamAsync(LiveStreamSession session, CancellationToken cancellationToken = default) + { + var key = GetLiveStreamKey(session.LiveStreamId, session.SessionId); + var json = JsonSerializer.Serialize(session); + // Live stream records use the same lease duration as transcode sessions. + var leaseDurationMs = (long)_options.LeaseDurationSeconds * 1000; + await _db.StringSetAsync(key, json, TimeSpan.FromMilliseconds(leaseDurationMs)).ConfigureAwait(false); + + // Also index by play session id so the caller can look up by either key. + if (!string.IsNullOrEmpty(session.PlaySessionId)) + { + var playKey = GetLiveStreamKey(session.LiveStreamId, session.PlaySessionId); + await _db.StringSetAsync(playKey, json, TimeSpan.FromMilliseconds(leaseDurationMs)).ConfigureAwait(false); + } + + _logger.LogDebug( + "Set live stream session {LiveStreamId}/{SessionId} in Redis.", + session.LiveStreamId, + session.SessionId); + } + + /// + public async Task TryGetLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default) + { + var key = GetLiveStreamKey(liveStreamId, sessionIdOrPlaySessionId); + var raw = await _db.StringGetAsync(key).ConfigureAwait(false); + if (!raw.HasValue) + { + return null; + } + + return JsonSerializer.Deserialize(raw.ToString()); + } + + /// + public async Task DeleteLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default) + { + var key = GetLiveStreamKey(liveStreamId, sessionIdOrPlaySessionId); + var raw = await _db.StringGetAsync(key).ConfigureAwait(false); + if (raw.HasValue) + { + var session = JsonSerializer.Deserialize(raw.ToString()); + if (session is not null) + { + // Remove both the session-id key and the play-session-id key if present. + var keysToDelete = new System.Collections.Generic.List + { + GetLiveStreamKey(liveStreamId, session.SessionId) + }; + + if (!string.IsNullOrEmpty(session.PlaySessionId)) + { + keysToDelete.Add(GetLiveStreamKey(liveStreamId, session.PlaySessionId)); + } + + await _db.KeyDeleteAsync(keysToDelete.ToArray()).ConfigureAwait(false); + _logger.LogDebug( + "Deleted live stream session {LiveStreamId}/{SessionId} from Redis.", + liveStreamId, + session.SessionId); + return; + } + } + + // Fallback: delete just the key that was supplied. + await _db.KeyDeleteAsync(key).ConfigureAwait(false); + } +} diff --git a/Jellyfin.Server/CoreAppHost.cs b/Jellyfin.Server/CoreAppHost.cs index 2548ddea7c..6020a1bc33 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 9f5bf01a05..00fdc1d894 100644 --- a/Jellyfin.Server/Jellyfin.Server.csproj +++ b/Jellyfin.Server/Jellyfin.Server.csproj @@ -56,6 +56,7 @@ + diff --git a/MediaBrowser.Controller/MediaEncoding/ITranscodeSessionStore.cs b/MediaBrowser.Controller/MediaEncoding/ITranscodeSessionStore.cs new file mode 100644 index 0000000000..e72d327cd1 --- /dev/null +++ b/MediaBrowser.Controller/MediaEncoding/ITranscodeSessionStore.cs @@ -0,0 +1,104 @@ +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; + +namespace MediaBrowser.Controller.MediaEncoding; + +/// +/// Provides a durable store for HLS transcoding session state, enabling +/// HA recovery and lease-based ownership between pods. +/// +public interface ITranscodeSessionStore +{ + /// + /// Attempts to retrieve a transcoding session by its play session identifier. + /// + /// The play session identifier. + /// A cancellation token. + /// + /// The if it exists and its lease has not expired; + /// otherwise null. + /// + Task TryGetAsync(string playSessionId, CancellationToken cancellationToken = default); + + /// + /// Attempts to take over ownership of an existing session by claiming the lease for + /// . Takeover succeeds only when the session exists and + /// its current lease has already expired. + /// + /// The play session identifier. + /// The name of the pod attempting to claim ownership. + /// A cancellation token. + /// + /// true if the takeover succeeded (the claiming pod now holds the lease); + /// false if the session does not exist, its lease is still valid, or another + /// concurrent caller already claimed it. + /// + Task TryTakeoverAsync(string playSessionId, string claimingPod, CancellationToken cancellationToken = default); + + /// + /// Persists a new or updated transcoding session. + /// + /// The session to store. + /// A cancellation token. + /// A representing the asynchronous operation. + Task SetAsync(TranscodeSession session, CancellationToken cancellationToken = default); + + /// + /// Renews the lease for an existing session, extending its + /// by the store's configured lease duration. + /// + /// The play session identifier. + /// A cancellation token. + /// A representing the asynchronous operation. + Task RenewLeaseAsync(string playSessionId, CancellationToken cancellationToken = default); + + /// + /// Removes a transcoding session from the store. + /// + /// The play session identifier. + /// A cancellation token. + /// A representing the asynchronous operation. + Task DeleteAsync(string playSessionId, CancellationToken cancellationToken = default); + + /// + /// Returns all currently active transcoding sessions from the store. + /// + /// A cancellation token. + /// + /// An enumerable of objects representing all active sessions. + /// Returns an empty enumerable if no sessions are active or if the store cannot be reached. + /// + Task> GetActiveSessionsAsync(CancellationToken cancellationToken = default); + + /// + /// Persists a live stream session record so that takeover pods can identify and close + /// streams that were opened on a pod that has since crashed or been evicted. + /// + /// The live stream session to store. + /// A cancellation token. + /// A representing the asynchronous operation. + Task SetLiveStreamAsync(LiveStreamSession session, CancellationToken cancellationToken = default); + + /// + /// Attempts to retrieve a live stream session by its live stream identifier and the + /// session or play-session identifier that owns it. + /// + /// The live stream identifier. + /// The session identifier or play-session identifier. + /// A cancellation token. + /// + /// The if it exists; otherwise null. + /// + Task TryGetLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default); + + /// + /// Removes the live stream session record for the given live stream and session identifier. + /// This is called when the stream is closed, either by the owning pod or a takeover pod. + /// + /// The live stream identifier. + /// The session identifier or play-session identifier. + /// A cancellation token. + /// A representing the asynchronous operation. + Task DeleteLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default); +} diff --git a/MediaBrowser.Controller/MediaEncoding/LiveStreamSession.cs b/MediaBrowser.Controller/MediaEncoding/LiveStreamSession.cs new file mode 100644 index 0000000000..6549d4adb1 --- /dev/null +++ b/MediaBrowser.Controller/MediaEncoding/LiveStreamSession.cs @@ -0,0 +1,36 @@ +using System; + +namespace MediaBrowser.Controller.MediaEncoding; + +/// +/// Represents a durable record of an open live stream session, enabling HA pod recovery +/// when the owning pod crashes or is evicted. +/// +public sealed class LiveStreamSession +{ + /// + /// Gets or sets the live stream identifier (e.g. a TV tuner channel token). + /// + public string LiveStreamId { get; set; } = string.Empty; + + /// + /// Gets or sets the session identifier of the client that opened this live stream. + /// + public string SessionId { get; set; } = string.Empty; + + /// + /// Gets or sets the play session identifier associated with this live stream, + /// or an empty string when the client did not supply one. + /// + public string PlaySessionId { get; set; } = string.Empty; + + /// + /// Gets or sets the name of the pod that currently holds this live stream open. + /// + public string OwnerPod { get; set; } = string.Empty; + + /// + /// Gets or sets the UTC time at which this record was created. + /// + public DateTime OpenedAtUtc { get; set; } +} diff --git a/MediaBrowser.Controller/MediaEncoding/NullTranscodeSessionStore.cs b/MediaBrowser.Controller/MediaEncoding/NullTranscodeSessionStore.cs new file mode 100644 index 0000000000..a92ab8098d --- /dev/null +++ b/MediaBrowser.Controller/MediaEncoding/NullTranscodeSessionStore.cs @@ -0,0 +1,49 @@ +using System; +using System.Collections.Generic; +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; + + /// + public Task> GetActiveSessionsAsync(CancellationToken cancellationToken = default) + => Task.FromResult>(Array.Empty()); + + /// + 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; +} diff --git a/MediaBrowser.Controller/MediaEncoding/TranscodeSession.cs b/MediaBrowser.Controller/MediaEncoding/TranscodeSession.cs new file mode 100644 index 0000000000..690f36e7cd --- /dev/null +++ b/MediaBrowser.Controller/MediaEncoding/TranscodeSession.cs @@ -0,0 +1,49 @@ +using System; + +namespace MediaBrowser.Controller.MediaEncoding; + +/// +/// Represents a durable record of an HLS transcoding session for HA pod recovery. +/// +public sealed class TranscodeSession +{ + /// + /// Gets or sets the unique play session identifier. + /// + public string PlaySessionId { get; set; } = string.Empty; + + /// + /// Gets or sets the name of the pod that currently owns this session's lease. + /// + public string OwnerPod { get; set; } = string.Empty; + + /// + /// Gets or sets the UTC time at which the owning pod's lease expires. + /// + public DateTime LeaseExpiresUtc { get; set; } + + /// + /// Gets or sets the absolute path to the HLS manifest (.m3u8) file on shared storage. + /// + public string ManifestPath { get; set; } = string.Empty; + + /// + /// Gets or sets the path prefix for transcoded segment files on shared storage. + /// + public string SegmentPathPrefix { get; set; } = string.Empty; + + /// + /// Gets or sets the media source identifier associated with this session. + /// + public string MediaSourceId { get; set; } = string.Empty; + + /// + /// Gets or sets the zero-based index of the last segment that was fully written to durable storage. + /// + public int LastCompletedSegmentIndex { get; set; } + + /// + /// Gets or sets the last durable playback offset in ticks, used to resume playback after failover. + /// + public long LastDurablePlaybackOffset { get; set; } +} diff --git a/MediaBrowser.Controller/MediaEncoding/TranscodeStoreOptions.cs b/MediaBrowser.Controller/MediaEncoding/TranscodeStoreOptions.cs new file mode 100644 index 0000000000..23d457bd22 --- /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.MediaEncoding.Tests/Fakes/InMemoryTranscodeSessionStore.cs b/tests/Jellyfin.MediaEncoding.Tests/Fakes/InMemoryTranscodeSessionStore.cs new file mode 100644 index 0000000000..af3b89c350 --- /dev/null +++ b/tests/Jellyfin.MediaEncoding.Tests/Fakes/InMemoryTranscodeSessionStore.cs @@ -0,0 +1,169 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using MediaBrowser.Controller.MediaEncoding; + +namespace Jellyfin.MediaEncoding.Tests.Fakes; + +/// +/// Thread-safe, in-memory implementation of for use in unit tests. +/// +public sealed class InMemoryTranscodeSessionStore : ITranscodeSessionStore +{ + /// + /// The duration added to when a lease is renewed or first claimed. + /// + public static readonly TimeSpan DefaultLeaseDuration = TimeSpan.FromSeconds(30); + + private readonly Dictionary _sessions = new(StringComparer.OrdinalIgnoreCase); + private readonly Dictionary _liveStreams = 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) + { + // Another pod's lease is still valid – takeover not permitted. + return Task.FromResult(false); + } + + // Lease has expired – claim it atomically. + 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.Select(Clone).ToList(); + return Task.FromResult>(sessions); + } + } + + /// + public Task SetLiveStreamAsync(LiveStreamSession session, CancellationToken cancellationToken = default) + { + lock (_lock) + { + _liveStreams[MakeLiveStreamKey(session.LiveStreamId, session.SessionId)] = session; + if (!string.IsNullOrEmpty(session.PlaySessionId)) + { + _liveStreams[MakeLiveStreamKey(session.LiveStreamId, session.PlaySessionId)] = session; + } + } + + return Task.CompletedTask; + } + + /// + public Task TryGetLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default) + { + lock (_lock) + { + _liveStreams.TryGetValue(MakeLiveStreamKey(liveStreamId, sessionIdOrPlaySessionId), out var session); + return Task.FromResult(session); + } + } + + /// + public Task DeleteLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default) + { + lock (_lock) + { + if (_liveStreams.TryGetValue(MakeLiveStreamKey(liveStreamId, sessionIdOrPlaySessionId), out var session)) + { + _liveStreams.Remove(MakeLiveStreamKey(liveStreamId, session.SessionId)); + if (!string.IsNullOrEmpty(session.PlaySessionId)) + { + _liveStreams.Remove(MakeLiveStreamKey(liveStreamId, session.PlaySessionId)); + } + } + else + { + _liveStreams.Remove(MakeLiveStreamKey(liveStreamId, sessionIdOrPlaySessionId)); + } + } + + return Task.CompletedTask; + } + + private static string MakeLiveStreamKey(string liveStreamId, string sessionIdOrPlaySessionId) + => liveStreamId + "\x00" + sessionIdOrPlaySessionId; + + 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.MediaEncoding.Tests/Transcoding/TranscodeManagerTests.cs b/tests/Jellyfin.MediaEncoding.Tests/Transcoding/TranscodeManagerTests.cs new file mode 100644 index 0000000000..8ea94edb9b --- /dev/null +++ b/tests/Jellyfin.MediaEncoding.Tests/Transcoding/TranscodeManagerTests.cs @@ -0,0 +1,167 @@ +using System; +using System.Threading.Tasks; +using Jellyfin.MediaEncoding.Tests.Fakes; +using MediaBrowser.Controller.MediaEncoding; +using Xunit; + +namespace Jellyfin.MediaEncoding.Tests.Transcoding; + +/// +/// Unit tests for the HA session-store contract: lease expiry, double-claim prevention, +/// heartbeat renewal, and stale-session cleanup. +/// All tests exercise which implements the +/// interface that will be backed by Redis in Phase 5.2. +/// +public class TranscodeManagerTests +{ + private static TranscodeSession CreateSession( + string id, + string pod, + DateTime leaseExpiry, + int lastSegmentIndex = 0, + long lastOffset = 0L) + => new TranscodeSession + { + PlaySessionId = id, + OwnerPod = pod, + LeaseExpiresUtc = leaseExpiry, + ManifestPath = $"/transcode/{id}/manifest.m3u8", + SegmentPathPrefix = $"/transcode/{id}/segment", + MediaSourceId = $"media-source-{id}", + LastCompletedSegmentIndex = lastSegmentIndex, + LastDurablePlaybackOffset = lastOffset, + }; + + /// + /// Lease expiry: returns null + /// once has passed. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task TryGetAsync_AfterLeaseExpires_ReturnsNull() + { + var store = new InMemoryTranscodeSessionStore(); + var session = CreateSession("session-expired", "pod-a", DateTime.UtcNow.AddMilliseconds(-1)); + await store.SetAsync(session, TestContext.Current.CancellationToken); + + var result = await store.TryGetAsync("session-expired", TestContext.Current.CancellationToken); + + Assert.Null(result); + } + + /// + /// A session whose lease has not yet expired is returned correctly. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task TryGetAsync_WithinLease_ReturnsSession() + { + var store = new InMemoryTranscodeSessionStore(); + var session = CreateSession("session-live", "pod-a", DateTime.UtcNow.AddMinutes(5), lastSegmentIndex: 3, lastOffset: 18_000_000L); + await store.SetAsync(session, TestContext.Current.CancellationToken); + + var result = await store.TryGetAsync("session-live", TestContext.Current.CancellationToken); + + Assert.NotNull(result); + Assert.Equal("session-live", result.PlaySessionId); + Assert.Equal("pod-a", result.OwnerPod); + Assert.Equal(3, result.LastCompletedSegmentIndex); + Assert.Equal(18_000_000L, result.LastDurablePlaybackOffset); + } + + /// + /// Double-claim prevention: returns + /// false while the first pod's lease is still valid. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task TryTakeoverAsync_WhileLeaseValid_ReturnsFalse() + { + var store = new InMemoryTranscodeSessionStore(); + var session = CreateSession("session-valid", "pod-a", DateTime.UtcNow.AddMinutes(5)); + await store.SetAsync(session, TestContext.Current.CancellationToken); + + var firstAttempt = await store.TryTakeoverAsync("session-valid", "pod-b", TestContext.Current.CancellationToken); + var secondAttempt = await store.TryTakeoverAsync("session-valid", "pod-c", TestContext.Current.CancellationToken); + + Assert.False(firstAttempt); + Assert.False(secondAttempt); + } + + /// + /// After a lease expires, the first concurrent caller that invokes + /// wins; the second caller returns false. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task TryTakeoverAsync_AfterLeaseExpires_OnlyFirstClaimerSucceeds() + { + var store = new InMemoryTranscodeSessionStore(); + var session = CreateSession("session-stale", "pod-a", DateTime.UtcNow.AddMilliseconds(-1)); + await store.SetAsync(session, TestContext.Current.CancellationToken); + + // First pod wins; its takeover renews the lease atomically. + var firstTakeover = await store.TryTakeoverAsync("session-stale", "pod-b", TestContext.Current.CancellationToken); + + // Second pod is too late – pod-b already holds a fresh lease. + var secondTakeover = await store.TryTakeoverAsync("session-stale", "pod-c", TestContext.Current.CancellationToken); + + Assert.True(firstTakeover); + Assert.False(secondTakeover); + } + + /// + /// Heartbeat renewal: extends + /// beyond its original value. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task RenewLeaseAsync_ExtendsLeaseExpiry() + { + var store = new InMemoryTranscodeSessionStore(); + var originalExpiry = DateTime.UtcNow.AddSeconds(5); + var session = CreateSession("session-renew", "pod-a", originalExpiry, lastSegmentIndex: 2, lastOffset: 10_000_000L); + await store.SetAsync(session, TestContext.Current.CancellationToken); + + await store.RenewLeaseAsync("session-renew", TestContext.Current.CancellationToken); + + var renewed = await store.TryGetAsync("session-renew", TestContext.Current.CancellationToken); + Assert.NotNull(renewed); + Assert.True( + renewed.LeaseExpiresUtc > originalExpiry, + "Renewed lease expiry should be later than the original expiry."); + } + + /// + /// Stale-session cleanup: an expired session can be deleted without error, and a + /// subsequent returns null. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task DeleteAsync_ExpiredSession_CompletesWithoutError() + { + var store = new InMemoryTranscodeSessionStore(); + var session = CreateSession("session-delete", "pod-a", DateTime.UtcNow.AddMilliseconds(-1)); + await store.SetAsync(session, TestContext.Current.CancellationToken); + + var ex = await Record.ExceptionAsync(() => store.DeleteAsync("session-delete", TestContext.Current.CancellationToken)); + Assert.Null(ex); + + var result = await store.TryGetAsync("session-delete", TestContext.Current.CancellationToken); + Assert.Null(result); + } + + /// + /// Deleting a session that was never stored must complete without error. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task DeleteAsync_NonExistentSession_CompletesWithoutError() + { + var store = new InMemoryTranscodeSessionStore(); + + var ex = await Record.ExceptionAsync(() => store.DeleteAsync("nonexistent-session", TestContext.Current.CancellationToken)); + + Assert.Null(ex); + } +} 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 0000000000..086ce535e2 --- /dev/null +++ b/tests/Jellyfin.Server.Implementations.Tests/MediaEncoding/RedisTranscodeSessionStoreTests.cs @@ -0,0 +1,384 @@ +using System; +using System.Collections.Generic; +using System.Linq; +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, TestContext.Current.CancellationToken); + + var result = await store.TryGetAsync("session-1", TestContext.Current.CancellationToken); + + 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, TestContext.Current.CancellationToken); + + var result = await store.TryTakeoverAsync("session-2", "pod-b", TestContext.Current.CancellationToken); + + 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, TestContext.Current.CancellationToken); + + var result = await store.TryTakeoverAsync("session-3", "pod-b", TestContext.Current.CancellationToken); + + Assert.True(result); + + var updated = await store.TryGetAsync("session-3", TestContext.Current.CancellationToken); + 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, TestContext.Current.CancellationToken); + + 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, TestContext.Current.CancellationToken); + } + + var results = await Task.WhenAll(tasks); + + var successCount = 0; + foreach (var r in results) + { + if (r) + { + successCount++; + } + } + + Assert.Equal(1, successCount); + } + + /// + /// Verifies that stores a live stream + /// record that can be retrieved by session id. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task SetLiveStreamAsync_CanBeRetrievedBySessionId() + { + var store = new InMemoryTranscodeSessionStore(); + var liveStream = new LiveStreamSession + { + LiveStreamId = "stream-1", + SessionId = "session-a", + PlaySessionId = "play-session-a", + OwnerPod = "pod-a", + OpenedAtUtc = DateTime.UtcNow, + }; + + await store.SetLiveStreamAsync(liveStream, TestContext.Current.CancellationToken); + + var result = await store.TryGetLiveStreamAsync("stream-1", "session-a", TestContext.Current.CancellationToken); + Assert.NotNull(result); + Assert.Equal("session-a", result.SessionId); + Assert.Equal("pod-a", result.OwnerPod); + } + + /// + /// Verifies that stores a live stream + /// record that can be retrieved by play session id. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task SetLiveStreamAsync_CanBeRetrievedByPlaySessionId() + { + var store = new InMemoryTranscodeSessionStore(); + var liveStream = new LiveStreamSession + { + LiveStreamId = "stream-2", + SessionId = "session-b", + PlaySessionId = "play-session-b", + OwnerPod = "pod-a", + OpenedAtUtc = DateTime.UtcNow, + }; + + await store.SetLiveStreamAsync(liveStream, TestContext.Current.CancellationToken); + + var result = await store.TryGetLiveStreamAsync("stream-2", "play-session-b", TestContext.Current.CancellationToken); + Assert.NotNull(result); + Assert.Equal("session-b", result.SessionId); + } + + /// + /// Verifies that removes the live + /// stream record so that subsequent lookups by either session id or play session id return null. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task DeleteLiveStreamAsync_RemovesBothKeys() + { + var store = new InMemoryTranscodeSessionStore(); + var liveStream = new LiveStreamSession + { + LiveStreamId = "stream-3", + SessionId = "session-c", + PlaySessionId = "play-session-c", + OwnerPod = "pod-a", + OpenedAtUtc = DateTime.UtcNow, + }; + + await store.SetLiveStreamAsync(liveStream, TestContext.Current.CancellationToken); + await store.DeleteLiveStreamAsync("stream-3", "session-c", TestContext.Current.CancellationToken); + + var bySessionId = await store.TryGetLiveStreamAsync("stream-3", "session-c", TestContext.Current.CancellationToken); + var byPlaySessionId = await store.TryGetLiveStreamAsync("stream-3", "play-session-c", TestContext.Current.CancellationToken); + + Assert.Null(bySessionId); + Assert.Null(byPlaySessionId); + } + + /// + /// Verifies that returns null when + /// no matching record exists. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task TryGetLiveStreamAsync_WhenNotPresent_ReturnsNull() + { + var store = new InMemoryTranscodeSessionStore(); + + var result = await store.TryGetLiveStreamAsync("nonexistent-stream", "nonexistent-session", TestContext.Current.CancellationToken); + + Assert.Null(result); + } + + /// + /// 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 Dictionary _liveStreams = 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) + { + lock (_lock) + { + _liveStreams[MakeLiveStreamKey(session.LiveStreamId, session.SessionId)] = session; + if (!string.IsNullOrEmpty(session.PlaySessionId)) + { + _liveStreams[MakeLiveStreamKey(session.LiveStreamId, session.PlaySessionId)] = session; + } + } + + return Task.CompletedTask; + } + + /// + public Task TryGetLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default) + { + lock (_lock) + { + _liveStreams.TryGetValue(MakeLiveStreamKey(liveStreamId, sessionIdOrPlaySessionId), out var session); + return Task.FromResult(session); + } + } + + /// + public Task DeleteLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default) + { + lock (_lock) + { + if (_liveStreams.TryGetValue(MakeLiveStreamKey(liveStreamId, sessionIdOrPlaySessionId), out var session)) + { + _liveStreams.Remove(MakeLiveStreamKey(liveStreamId, session.SessionId)); + if (!string.IsNullOrEmpty(session.PlaySessionId)) + { + _liveStreams.Remove(MakeLiveStreamKey(liveStreamId, session.PlaySessionId)); + } + } + else + { + _liveStreams.Remove(MakeLiveStreamKey(liveStreamId, sessionIdOrPlaySessionId)); + } + } + + return Task.CompletedTask; + } + + private static string MakeLiveStreamKey(string liveStreamId, string sessionIdOrPlaySessionId) + => liveStreamId + "\x00" + sessionIdOrPlaySessionId; + + 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, + }; + } +}