diff --git a/Directory.Packages.props b/Directory.Packages.props index 1732f575a5..28376ff47d 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -86,6 +86,7 @@ + diff --git a/Emby.Server.Implementations/MediaEncoding/RedisTranscodeSessionStore.cs b/Emby.Server.Implementations/MediaEncoding/RedisTranscodeSessionStore.cs index 6ac5c01205..1f75b59f47 100644 --- a/Emby.Server.Implementations/MediaEncoding/RedisTranscodeSessionStore.cs +++ b/Emby.Server.Implementations/MediaEncoding/RedisTranscodeSessionStore.cs @@ -2,6 +2,7 @@ using System; using System.Collections.Generic; using System.Linq; using System.Text.Json; +using System.Text.Json.Serialization; using System.Threading; using System.Threading.Tasks; using MediaBrowser.Controller.MediaEncoding; @@ -18,26 +19,42 @@ namespace Emby.Server.Implementations.MediaEncoding; 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. + /// expired (comparing the stored expiry against the caller-supplied current time) and, if it + /// has, claims it for the calling pod 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 +if tonumber(session['LeaseExpiresUtc']) > tonumber(ARGV[1]) 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) +session['LeaseExpiresUtc'] = tonumber(ARGV[1]) + tonumber(ARGV[3]) +redis.call('SET', KEYS[1], cjson.encode(session), 'PX', tonumber(ARGV[4])) return 1"; + /// + /// Lua script for atomic, ownership-checked renewal: extends the lease only while the calling + /// pod still owns it, so a renewal racing a successful takeover cannot revert the new owner. + /// + private const string RenewScript = @" +local raw = redis.call('GET', KEYS[1]) +if not raw then return 0 end +local session = cjson.decode(raw) +if session['OwnerPod'] ~= ARGV[2] then return 0 end +session['LeaseExpiresUtc'] = tonumber(ARGV[1]) + tonumber(ARGV[3]) +redis.call('SET', KEYS[1], cjson.encode(session), 'PX', tonumber(ARGV[4])) +return 1"; + + // The lease expiry is serialized as unix milliseconds because the Lua scripts compare it + // numerically; an ISO-8601 string cannot be compared against a number in Lua. + private static readonly JsonSerializerOptions _jsonOptions = new JsonSerializerOptions + { + Converters = { new UnixMillisecondsDateTimeConverter() } + }; + private readonly IConnectionMultiplexer _redis; private readonly IDatabase _db; private readonly TranscodeStoreOptions _options; @@ -60,13 +77,16 @@ return 1"; _logger = logger; } + private long LeaseDurationMs => (long)_options.LeaseDurationSeconds * 1000; + + private long RetentionMs => Math.Max((long)_options.SessionRetentionSeconds * 1000, LeaseDurationMs); + /// 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); + var json = JsonSerializer.Serialize(session, _jsonOptions); + await _db.StringSetAsync(key, json, TimeSpan.FromMilliseconds(RetentionMs)).ConfigureAwait(false); _logger.LogDebug("Set transcode session {PlaySessionId} in Redis.", session.PlaySessionId); } @@ -80,10 +100,10 @@ return 1"; return null; } - var session = JsonSerializer.Deserialize(raw.ToString()); + var session = JsonSerializer.Deserialize(raw.ToString(), _jsonOptions); - // 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. + // The record outlives the lease so that an orphaned session can still be taken over, + // so the lease has to be checked explicitly here. if (session is null || session.LeaseExpiresUtc <= DateTime.UtcNow) { return null; @@ -93,26 +113,24 @@ return 1"; } /// - public async Task RenewLeaseAsync(string playSessionId, CancellationToken cancellationToken = default) + public async Task RenewLeaseAsync(string playSessionId, string ownerPod, CancellationToken cancellationToken = default) { - var key = GetKey(playSessionId); - var raw = await _db.StringGetAsync(key).ConfigureAwait(false); - if (!raw.HasValue) + var result = (long?)await _db.ScriptEvaluateAsync( + RenewScript, + keys: new RedisKey[] { GetKey(playSessionId) }, + values: new RedisValue[] { UnixMillisecondsNow(), ownerPod, LeaseDurationMs, RetentionMs }).ConfigureAwait(false); + + if (result != 1) { - return; + _logger.LogWarning( + "Pod {OwnerPod} no longer owns transcode session {PlaySessionId}; lease not renewed.", + ownerPod, + playSessionId); + return false; } - 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); + return true; } /// @@ -126,14 +144,10 @@ return 1"; /// 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); + keys: new RedisKey[] { GetKey(playSessionId) }, + values: new RedisValue[] { UnixMillisecondsNow(), claimingPod, LeaseDurationMs, RetentionMs }).ConfigureAwait(false); var succeeded = result == 1; if (succeeded) @@ -144,11 +158,6 @@ return 1"; 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) { @@ -181,7 +190,7 @@ return 1"; TranscodeSession? session; try { - session = JsonSerializer.Deserialize(raw.ToString()); + session = JsonSerializer.Deserialize(raw.ToString(), _jsonOptions); } catch (Exception ex) { @@ -189,7 +198,7 @@ return 1"; continue; } - if (session is not null) + if (session is not null && session.LeaseExpiresUtc > DateTime.UtcNow) { sessions.Add(session); } @@ -199,72 +208,28 @@ return 1"; 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); + private static string GetKey(string playSessionId) => KeyPrefix + playSessionId; - // Also index by play session id so the caller can look up by either key. - if (!string.IsNullOrEmpty(session.PlaySessionId)) + private static long UnixMillisecondsNow() => DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); + + private sealed class UnixMillisecondsDateTimeConverter : JsonConverter + { + public override DateTime Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options) { - var playKey = GetLiveStreamKey(session.LiveStreamId, session.PlaySessionId); - await _db.StringSetAsync(playKey, json, TimeSpan.FromMilliseconds(leaseDurationMs)).ConfigureAwait(false); + var milliseconds = reader.TryGetInt64(out var value) ? value : (long)reader.GetDouble(); + return DateTimeOffset.FromUnixTimeMilliseconds(milliseconds).UtcDateTime; } - _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) + public override void Write(Utf8JsonWriter writer, DateTime value, JsonSerializerOptions options) { - 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) + var utc = value.Kind switch { - // 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) - }; + DateTimeKind.Utc => value, + DateTimeKind.Local => value.ToUniversalTime(), + _ => DateTime.SpecifyKind(value, DateTimeKind.Utc) + }; - 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; - } + writer.WriteNumberValue(new DateTimeOffset(utc, TimeSpan.Zero).ToUnixTimeMilliseconds()); } - - // Fallback: delete just the key that was supplied. - await _db.KeyDeleteAsync(key).ConfigureAwait(false); } } diff --git a/Emby.Server.Implementations/Session/SessionManager.cs b/Emby.Server.Implementations/Session/SessionManager.cs index 0a81e61870..94215bed79 100644 --- a/Emby.Server.Implementations/Session/SessionManager.cs +++ b/Emby.Server.Implementations/Session/SessionManager.cs @@ -28,7 +28,6 @@ using MediaBrowser.Controller.Events; using MediaBrowser.Controller.Events.Authentication; using MediaBrowser.Controller.Events.Session; using MediaBrowser.Controller.Library; -using MediaBrowser.Controller.MediaEncoding; using MediaBrowser.Controller.Net; using MediaBrowser.Controller.Session; using MediaBrowser.Model.Dto; @@ -61,7 +60,6 @@ namespace Emby.Server.Implementations.Session private readonly IMediaSourceManager _mediaSourceManager; private readonly IServerApplicationHost _appHost; private readonly IDeviceManager _deviceManager; - private readonly ITranscodeSessionStore _transcodeSessionStore; private readonly CancellationTokenRegistration _shutdownCallback; private readonly ConcurrentDictionary _activeConnections = new(StringComparer.OrdinalIgnoreCase); @@ -91,7 +89,6 @@ namespace Emby.Server.Implementations.Session /// Instance of interface. /// Instance of interface. /// Instance of interface. - /// Instance of interface. public SessionManager( ILogger logger, IEventManager eventManager, @@ -105,8 +102,7 @@ namespace Emby.Server.Implementations.Session IServerApplicationHost appHost, IDeviceManager deviceManager, IMediaSourceManager mediaSourceManager, - IHostApplicationLifetime hostApplicationLifetime, - ITranscodeSessionStore transcodeSessionStore) + IHostApplicationLifetime hostApplicationLifetime) { _logger = logger; _eventManager = eventManager; @@ -120,7 +116,6 @@ namespace Emby.Server.Implementations.Session _appHost = appHost; _deviceManager = deviceManager; _mediaSourceManager = mediaSourceManager; - _transcodeSessionStore = transcodeSessionStore; _shutdownCallback = hostApplicationLifetime.ApplicationStopping.Register(OnApplicationStopping); _deviceManager.DeviceOptionsUpdated += OnDeviceManagerDeviceOptionsUpdated; @@ -355,15 +350,6 @@ namespace Emby.Server.Implementations.Session if (liveStreamNeedsToBeClosed) { - try - { - await _transcodeSessionStore.DeleteLiveStreamAsync(liveStreamId, sessionIdOrPlaySessionId).ConfigureAwait(false); - } - catch (Exception ex) - { - _logger.LogWarning(ex, "Failed to delete live stream session {LiveStreamId}/{SessionId} from durable store.", liveStreamId, sessionIdOrPlaySessionId); - } - try { await _mediaSourceManager.CloseLiveStream(liveStreamId).ConfigureAwait(false); @@ -810,7 +796,7 @@ namespace Emby.Server.Implementations.Session if (!string.IsNullOrEmpty(info.LiveStreamId)) { - await UpdateLiveStreamActiveSessionMappings(info.LiveStreamId, info.SessionId, info.PlaySessionId).ConfigureAwait(false); + UpdateLiveStreamActiveSessionMappings(info.LiveStreamId, info.SessionId, info.PlaySessionId); } var eventArgs = new PlaybackStartEventArgs @@ -876,7 +862,7 @@ namespace Emby.Server.Implementations.Session return OnPlaybackProgress(info, false); } - private async Task UpdateLiveStreamActiveSessionMappings(string liveStreamId, string sessionId, string playSessionId) + private void UpdateLiveStreamActiveSessionMappings(string liveStreamId, string sessionId, string playSessionId) { var activeSessionMappings = _activeLiveStreamSessions.GetOrAdd(liveStreamId, _ => new ConcurrentDictionary()); @@ -900,25 +886,6 @@ namespace Emby.Server.Implementations.Session activeSessionMappings[sessionId] = string.Empty; } } - - // Persist to the durable store so a takeover pod can discover open live streams. - var liveStreamSession = new LiveStreamSession - { - LiveStreamId = liveStreamId, - SessionId = sessionId, - PlaySessionId = playSessionId ?? string.Empty, - OwnerPod = Environment.GetEnvironmentVariable("JELLYFIN_INSTANCE_ID") ?? Environment.MachineName, - OpenedAtUtc = DateTime.UtcNow, - }; - - try - { - await _transcodeSessionStore.SetLiveStreamAsync(liveStreamSession).ConfigureAwait(false); - } - catch (Exception ex) - { - _logger.LogWarning(ex, "Failed to persist live stream session {LiveStreamId}/{SessionId} to durable store.", liveStreamId, sessionId); - } } /// @@ -964,7 +931,7 @@ namespace Emby.Server.Implementations.Session if (!string.IsNullOrEmpty(info.LiveStreamId)) { - await UpdateLiveStreamActiveSessionMappings(info.LiveStreamId, info.SessionId, info.PlaySessionId).ConfigureAwait(false); + UpdateLiveStreamActiveSessionMappings(info.LiveStreamId, info.SessionId, info.PlaySessionId); } var eventArgs = new PlaybackProgressEventArgs diff --git a/Jellyfin.Api/Controllers/DynamicHlsController.cs b/Jellyfin.Api/Controllers/DynamicHlsController.cs index bce703e34d..291ca5b0ed 100644 --- a/Jellyfin.Api/Controllers/DynamicHlsController.cs +++ b/Jellyfin.Api/Controllers/DynamicHlsController.cs @@ -29,6 +29,7 @@ using Microsoft.AspNetCore.Authorization; using Microsoft.AspNetCore.Http; using Microsoft.AspNetCore.Mvc; using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; namespace Jellyfin.Api.Controllers; @@ -44,6 +45,8 @@ public class DynamicHlsController : BaseJellyfinApiController private const EncoderPreset DefaultEventEncoderPreset = EncoderPreset.superfast; private const TranscodingJobType TranscodingJobType = MediaBrowser.Controller.MediaEncoding.TranscodingJobType.Hls; + private static readonly string _podIdentity = Environment.GetEnvironmentVariable("JELLYFIN_INSTANCE_ID") ?? Environment.MachineName; + private readonly Version _minFFmpegFlacInMp4 = new Version(6, 0); private readonly Version _minFFmpegX265BframeInFmp4 = new Version(7, 0, 1); private readonly Version _minFFmpegHlsSegmentOptions = new Version(5, 0); @@ -61,6 +64,7 @@ public class DynamicHlsController : BaseJellyfinApiController private readonly DynamicHlsHelper _dynamicHlsHelper; private readonly EncodingOptions _encodingOptions; private readonly ITranscodeSessionStore _transcodeSessionStore; + private readonly TranscodeStoreOptions _transcodeStoreOptions; /// /// Initializes a new instance of the class. @@ -77,6 +81,7 @@ public class DynamicHlsController : BaseJellyfinApiController /// Instance of . /// Instance of . /// Instance of the interface used to register and renew HLS transcoding session leases in the durable store. + /// The holding the session lease duration. public DynamicHlsController( ILibraryManager libraryManager, IUserManager userManager, @@ -89,7 +94,8 @@ public class DynamicHlsController : BaseJellyfinApiController DynamicHlsHelper dynamicHlsHelper, EncodingHelper encodingHelper, IDynamicHlsPlaylistGenerator dynamicHlsPlaylistGenerator, - ITranscodeSessionStore transcodeSessionStore) + ITranscodeSessionStore transcodeSessionStore, + IOptions transcodeStoreOptions) { _libraryManager = libraryManager; _userManager = userManager; @@ -103,6 +109,7 @@ public class DynamicHlsController : BaseJellyfinApiController _encodingHelper = encodingHelper; _dynamicHlsPlaylistGenerator = dynamicHlsPlaylistGenerator; _transcodeSessionStore = transcodeSessionStore; + _transcodeStoreOptions = transcodeStoreOptions.Value; _encodingOptions = serverConfigurationManager.GetEncodingOptions(); } @@ -324,6 +331,7 @@ public class DynamicHlsController : BaseJellyfinApiController await RegisterTranscodeSessionAsync( playSessionId ?? string.Empty, mediaSourceId ?? string.Empty, + playlistPath, cancellationToken) .ConfigureAwait(false); StartLeaseRenewal(playSessionId ?? string.Empty, cancellationToken); @@ -1536,6 +1544,7 @@ public class DynamicHlsController : BaseJellyfinApiController await RegisterTranscodeSessionAsync( streamingRequest.PlaySessionId ?? string.Empty, streamingRequest.MediaSourceId ?? string.Empty, + playlistPath, cancellationToken) .ConfigureAwait(false); StartLeaseRenewal(streamingRequest.PlaySessionId ?? string.Empty, cancellationToken); @@ -1563,6 +1572,31 @@ public class DynamicHlsController : BaseJellyfinApiController } } + /// + /// Gets the segment length ffmpeg is told to use. HA mode shortens segments so a takeover pod + /// has to re-encode less; the configured value is user-editable so it is clamped. + /// + /// Whether the session is being resumed from the durable store. + /// The segment length requested by the streaming pipeline. + /// The encoding options holding the recovery segment length. + /// The segment length in seconds. + internal static int GetEffectiveSegmentLength(bool isHaMode, int requestedSegmentLength, EncodingOptions encodingOptions) + => isHaMode + ? Math.Clamp(encodingOptions.RecoverySegmentLengthSeconds, 1, 6) + : requestedSegmentLength; + + /// + /// Gets the ffmpeg hls_list_size. HA mode keeps a bounded rolling buffer; 0 (unbounded) + /// otherwise. + /// + /// Whether the session is being resumed from the durable store. + /// The encoding options holding the recovery buffer count. + /// The number of segments to keep in the playlist, or 0 for all of them. + internal static int GetHlsListSize(bool isHaMode, EncodingOptions encodingOptions) + => isHaMode + ? Math.Clamp(encodingOptions.RecoverySegmentBufferCount, 2, 10) + : 0; + internal static async Task WaitForActiveTranscodingRequests(TranscodingJob? job, CancellationToken cancellationToken) { while (job?.ActiveRequestCount > 0) @@ -1596,22 +1630,28 @@ public class DynamicHlsController : BaseJellyfinApiController } } - private async Task RegisterTranscodeSessionAsync(string playSessionId, string mediaSourceId, CancellationToken cancellationToken) + /// + /// Builds the durable record for an HLS output owned by this instance. The manifest and segment + /// paths have to be real, otherwise DeleteTranscodeFileTask cannot tell which files + /// belong to a live session. + /// + /// The play session identifier. + /// The media source identifier. + /// The absolute path of the HLS playlist this instance is writing. + /// The initial lease duration. + /// The session record to store. + internal static TranscodeSession CreateSessionRecord(string playSessionId, string mediaSourceId, string playlistPath, TimeSpan leaseDuration) + => TranscodeSession.CreateForPlaylist(playSessionId, mediaSourceId, _podIdentity, playlistPath, leaseDuration); + + private async Task RegisterTranscodeSessionAsync(string playSessionId, string mediaSourceId, string playlistPath, 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, - }; + var session = CreateSessionRecord( + playSessionId, + mediaSourceId, + playlistPath, + TimeSpan.FromSeconds(_transcodeStoreOptions.LeaseDurationSeconds)); await _transcodeSessionStore.SetAsync(session, cancellationToken).ConfigureAwait(false); } catch (Exception ex) @@ -1622,16 +1662,19 @@ public class DynamicHlsController : BaseJellyfinApiController private void StartLeaseRenewal(string playSessionId, CancellationToken cancellationToken) { + var renewalInterval = TimeSpan.FromSeconds(Math.Max(1, _transcodeStoreOptions.LeaseDurationSeconds / 3)); + _ = Task.Run( async () => { + var ownsLease = true; try { while (!cancellationToken.IsCancellationRequested) { try { - await Task.Delay(TimeSpan.FromSeconds(10), cancellationToken).ConfigureAwait(false); + await Task.Delay(renewalInterval, cancellationToken).ConfigureAwait(false); } catch (OperationCanceledException) { @@ -1640,7 +1683,12 @@ public class DynamicHlsController : BaseJellyfinApiController try { - await _transcodeSessionStore.RenewLeaseAsync(playSessionId, cancellationToken).ConfigureAwait(false); + if (!await _transcodeSessionStore.RenewLeaseAsync(playSessionId, _podIdentity, cancellationToken).ConfigureAwait(false)) + { + // Another pod owns the session now; its record must not be touched. + ownsLease = false; + break; + } } catch (Exception ex) { @@ -1654,13 +1702,16 @@ public class DynamicHlsController : BaseJellyfinApiController } finally { - try + if (ownsLease) { - await _transcodeSessionStore.DeleteAsync(playSessionId, CancellationToken.None).ConfigureAwait(false); - } - catch (Exception ex) - { - _logger.LogWarning(ex, "Failed to delete HLS session {PlaySessionId} from durable store.", playSessionId); + 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); + } } } }, @@ -1701,15 +1752,8 @@ public class DynamicHlsController : BaseJellyfinApiController var outputExtension = EncodingHelper.GetSegmentFileExtension(state.Request.SegmentContainer); var outputTsArg = outputPrefix + "%d" + outputExtension; - // In HA mode, use shorter segments and a bounded rolling buffer for faster failover recovery. - // state.SegmentLength is already validated by the streaming pipeline; RecoverySegmentLengthSeconds - // comes from EncodingOptions (user-editable config) so it is clamped here. - var effectiveSegmentLength = isHaMode - ? Math.Clamp(_encodingOptions.RecoverySegmentLengthSeconds, 1, 6) - : state.SegmentLength; - var hlsListSize = isHaMode - ? Math.Clamp(_encodingOptions.RecoverySegmentBufferCount, 2, 10) - : 0; + var effectiveSegmentLength = GetEffectiveSegmentLength(isHaMode, state.SegmentLength, _encodingOptions); + var hlsListSize = GetHlsListSize(isHaMode, _encodingOptions); var segmentFormat = string.Empty; var segmentContainer = outputExtension.TrimStart('.'); diff --git a/MediaBrowser.Controller/MediaEncoding/ITranscodeSessionStore.cs b/MediaBrowser.Controller/MediaEncoding/ITranscodeSessionStore.cs index e72d327cd1..a69167928d 100644 --- a/MediaBrowser.Controller/MediaEncoding/ITranscodeSessionStore.cs +++ b/MediaBrowser.Controller/MediaEncoding/ITranscodeSessionStore.cs @@ -47,11 +47,17 @@ public interface ITranscodeSessionStore /// /// Renews the lease for an existing session, extending its /// by the store's configured lease duration. + /// The renewal is rejected when no longer owns the lease, so a + /// renewal in flight while another pod wins cannot revert the takeover. /// /// The play session identifier. + /// The name of the pod that believes it owns the lease. /// A cancellation token. - /// A representing the asynchronous operation. - Task RenewLeaseAsync(string playSessionId, CancellationToken cancellationToken = default); + /// + /// true if the lease was renewed; false if the session no longer exists or is + /// owned by another pod. + /// + Task RenewLeaseAsync(string playSessionId, string ownerPod, CancellationToken cancellationToken = default); /// /// Removes a transcoding session from the store. @@ -70,35 +76,4 @@ public interface ITranscodeSessionStore /// 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 deleted file mode 100644 index 6549d4adb1..0000000000 --- a/MediaBrowser.Controller/MediaEncoding/LiveStreamSession.cs +++ /dev/null @@ -1,36 +0,0 @@ -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 index a92ab8098d..e87f07a2c7 100644 --- a/MediaBrowser.Controller/MediaEncoding/NullTranscodeSessionStore.cs +++ b/MediaBrowser.Controller/MediaEncoding/NullTranscodeSessionStore.cs @@ -24,8 +24,8 @@ public sealed class NullTranscodeSessionStore : ITranscodeSessionStore => Task.CompletedTask; /// - public Task RenewLeaseAsync(string playSessionId, CancellationToken cancellationToken = default) - => Task.CompletedTask; + public Task RenewLeaseAsync(string playSessionId, string ownerPod, CancellationToken cancellationToken = default) + => Task.FromResult(true); /// public Task DeleteAsync(string playSessionId, CancellationToken cancellationToken = default) @@ -34,16 +34,4 @@ public sealed class NullTranscodeSessionStore : ITranscodeSessionStore /// 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 index 690f36e7cd..232e2006a3 100644 --- a/MediaBrowser.Controller/MediaEncoding/TranscodeSession.cs +++ b/MediaBrowser.Controller/MediaEncoding/TranscodeSession.cs @@ -1,4 +1,5 @@ using System; +using System.IO; namespace MediaBrowser.Controller.MediaEncoding; @@ -46,4 +47,40 @@ public sealed class TranscodeSession /// Gets or sets the last durable playback offset in ticks, used to resume playback after failover. /// public long LastDurablePlaybackOffset { get; set; } + + /// + /// Creates a session record for an HLS output, deriving and + /// from the playlist path so that cleanup can recognise + /// every file the session owns. + /// + /// The play session identifier. + /// The media source identifier. + /// The name of the pod that owns the session. + /// The absolute path of the HLS playlist (.m3u8) file. + /// The initial lease duration. + /// The new . + public static TranscodeSession CreateForPlaylist( + string playSessionId, + string mediaSourceId, + string ownerPod, + string playlistPath, + TimeSpan leaseDuration) + => new TranscodeSession + { + PlaySessionId = playSessionId, + OwnerPod = ownerPod, + LeaseExpiresUtc = DateTime.UtcNow.Add(leaseDuration), + ManifestPath = playlistPath, + SegmentPathPrefix = GetSegmentPathPrefix(playlistPath), + MediaSourceId = mediaSourceId, + }; + + /// + /// Gets the prefix every segment file of the HLS output at starts with. + /// Segments are written as <playlist path without extension><index><segment extension>. + /// + /// The absolute path of the HLS playlist (.m3u8) file. + /// The segment path prefix. + public static string GetSegmentPathPrefix(string playlistPath) + => Path.ChangeExtension(playlistPath, null) ?? playlistPath; } diff --git a/MediaBrowser.Controller/MediaEncoding/TranscodeStoreOptions.cs b/MediaBrowser.Controller/MediaEncoding/TranscodeStoreOptions.cs index 23d457bd22..65a7b31348 100644 --- a/MediaBrowser.Controller/MediaEncoding/TranscodeStoreOptions.cs +++ b/MediaBrowser.Controller/MediaEncoding/TranscodeStoreOptions.cs @@ -16,4 +16,11 @@ public sealed class TranscodeStoreOptions /// Gets or sets the duration in seconds for which a transcoding session lease is valid. /// public int LeaseDurationSeconds { get; set; } = 30; + + /// + /// Gets or sets how long in seconds a session record is retained after its lease was last renewed. + /// The record must outlive the lease, otherwise an orphaned session is gone before another pod + /// can take it over. + /// + public int SessionRetentionSeconds { get; set; } = 300; } diff --git a/README.md b/README.md index 5c02b0665b..26f34c97ad 100644 --- a/README.md +++ b/README.md @@ -18,7 +18,6 @@ This fork adds a thin HA layer on top of unmodified Jellyfin core: - **`RedisTranscodeSessionStore`** — a Redis-backed implementation using atomic Lua takeover scripts and TTL-based lease expiry - **`NullTranscodeSessionStore`** — a no-op fallback so single-instance deployments work with zero configuration change - **Lease-aware `DeleteTranscodeFileTask`** — coordinates cleanup across replicas so a restarting pod doesn't delete segments another pod is actively streaming -- **`SessionManager` HA recovery** — safe takeover of live HLS streams when a pod takes over after lease expiry - **PostgreSQL database provider** — alternative to SQLite for shared-database HA setups (experimental, under `src/Jellyfin.Database/Jellyfin.Database.Providers.PostgreSQL`) --- @@ -52,7 +51,7 @@ This fork adds a thin HA layer on top of unmodified Jellyfin core: **How takeover works:** 1. Pod A starts an HLS transcode and writes a `TranscodeSession` to Redis with a 30-second lease. -2. Pod A renews the lease every `LeaseDurationSeconds / 2` seconds. +2. Pod A renews the lease every `LeaseDurationSeconds / 3` seconds; the renewal is rejected if Pod A no longer owns it. 3. If Pod A dies, the lease expires in Redis after 30 seconds. 4. Pod B receives a client request for the same play session, calls `TryTakeoverAsync`, and atomically claims ownership via a Lua script. 5. Pod B resumes FFmpeg from the last durable segment index. The client sees a brief stutter, not an error. @@ -109,6 +108,7 @@ When `RedisConnectionString` is set, `RedisTranscodeSessionStore` is registered |-----|---------|-------------| | `Jellyfin:TranscodeStore:RedisConnectionString` | _(empty)_ | StackExchange.Redis connection string. Empty = single-instance mode. | | `Jellyfin:TranscodeStore:LeaseDurationSeconds` | `30` | How long a pod's transcode lease is valid before another pod may take over. | +| `Jellyfin:TranscodeStore:SessionRetentionSeconds` | `300` | How long an unrenewed session record is kept so another pod can still take it over. | ### Redis connection string examples @@ -465,8 +465,8 @@ dotnet test Jellyfin.sln \ The transcode session store and HA recovery tests live in: -- `tests/Jellyfin.Server.Implementations.Tests/MediaEncoding/RedisTranscodeSessionStoreTests.cs` -- `tests/Jellyfin.MediaEncoding.Tests/Fakes/InMemoryTranscodeSessionStore.cs` +- `tests/Jellyfin.Server.Implementations.Tests/MediaEncoding/RedisTranscodeSessionStoreTests.cs` (real Redis, `Category=RequiresDocker`) +- `tests/Jellyfin.MediaEncoding.Tests/Transcoding/InMemoryTranscodeSessionStoreTests.cs` ```bash dotnet test tests/Jellyfin.Server.Implementations.Tests \ @@ -502,9 +502,10 @@ src/Jellyfin.Database/ tests/ Jellyfin.Server.Implementations.Tests/MediaEncoding/ - RedisTranscodeSessionStoreTests.cs - Jellyfin.MediaEncoding.Tests/Fakes/ - InMemoryTranscodeSessionStore.cs + RedisTranscodeSessionStoreTests.cs ← real Redis via Testcontainers + Jellyfin.MediaEncoding.Tests/ + Fakes/InMemoryTranscodeSessionStore.cs + Transcoding/InMemoryTranscodeSessionStoreTests.cs ``` --- diff --git a/docs/FORK-DIFF.md b/docs/FORK-DIFF.md index 034579029a..0483ed27ff 100644 --- a/docs/FORK-DIFF.md +++ b/docs/FORK-DIFF.md @@ -32,21 +32,24 @@ auth or plugin logic is rewritten. | File | Purpose | |------|---------| -| `MediaBrowser.Controller/MediaEncoding/ITranscodeSessionStore.cs` | DI contract for durable transcode and live stream session tracking | +| `MediaBrowser.Controller/MediaEncoding/ITranscodeSessionStore.cs` | DI contract for durable transcode session tracking | | `MediaBrowser.Controller/MediaEncoding/TranscodeSession.cs` | Session record: owning pod, lease expiry, manifest and segment paths, last durable segment | -| `MediaBrowser.Controller/MediaEncoding/LiveStreamSession.cs` | Record of an open live stream and the pod holding it | -| `MediaBrowser.Controller/MediaEncoding/TranscodeStoreOptions.cs` | `RedisConnectionString` and `LeaseDurationSeconds` | +| `MediaBrowser.Controller/MediaEncoding/TranscodeStoreOptions.cs` | `RedisConnectionString`, `LeaseDurationSeconds` and `SessionRetentionSeconds` | | `MediaBrowser.Controller/MediaEncoding/NullTranscodeSessionStore.cs` | No-op store used when no Redis connection is configured | -| `Emby.Server.Implementations/MediaEncoding/RedisTranscodeSessionStore.cs` | Redis store; sessions under `jellyfin:transcode:{playSessionId}`, live streams under `jellyfin:livestream:{liveStreamId}:{sessionId}`, key TTL mirrors the lease | +| `Emby.Server.Implementations/MediaEncoding/RedisTranscodeSessionStore.cs` | Redis store; sessions under `jellyfin:transcode:{playSessionId}`, key TTL is the retention window so an orphaned session outlives its lease | -Lease takeover runs as a single Lua script, so concurrent pods cannot both claim -an expired lease: +Lease takeover and renewal each run as a single Lua script, so concurrent pods cannot both +claim an expired lease and a renewal cannot revert a takeover. The expiry is stored as unix +milliseconds so the script can compare it: ```lua local raw = redis.call('GET', KEYS[1]) if not raw then return 0 end local session = cjson.decode(raw) -if session['LeaseExpiresUtc'] > tonumber(ARGV[1]) then return 0 end +-- takeover: only an expired lease may be claimed +if tonumber(session['LeaseExpiresUtc']) > tonumber(ARGV[1]) then return 0 end +-- renewal: only the pod that still owns the lease may extend it +-- if session['OwnerPod'] ~= ARGV[2] then return 0 end session['OwnerPod'] = ARGV[2] -- update expiry and SET with PX in the same script return 1 @@ -84,9 +87,8 @@ optionally uploading a pre-migration copy of the SQLite file to S3. | File | Change | |------|--------| | `Jellyfin.Server/CoreAppHost.cs` (+46) | Registers the Redis or null transcode store and scan-leader lease from startup config | -| `Jellyfin.Api/Controllers/DynamicHlsController.cs` (+133/-9) | Registers the play session, runs lease renewal, and shortens segments when resuming a stored session | +| `Jellyfin.Api/Controllers/DynamicHlsController.cs` | Registers the play session with its manifest and segment paths, renews the lease under this pod's identity, and shortens segments when resuming a stored session | | `Emby.Server.Implementations/ScheduledTasks/Tasks/DeleteTranscodeFileTask.cs` (+52/-6) | Keeps files belonging to an active session in the store | -| `Emby.Server.Implementations/Session/SessionManager.cs` (+37/-4) | Persists and deletes the durable live stream record | | `Emby.Server.Implementations/ScheduledTasks/ScheduledTaskWorker.cs` (+28/-1) | Skips timer-driven gated tasks without the leader lease | | `Emby.Server.Implementations/ScheduledTasks/TaskManager.cs` (+12/-2) | Passes the lease and options to each worker | | `Jellyfin.Server.Implementations/Extensions/ServiceCollectionExtensions.cs` (+53) | Registers the PostgreSQL provider and a pooled `NpgsqlDataSource` | diff --git a/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsHaModeTests.cs b/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsHaModeTests.cs new file mode 100644 index 0000000000..dac7be4bb9 --- /dev/null +++ b/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsHaModeTests.cs @@ -0,0 +1,105 @@ +using System; +using System.Globalization; +using System.IO; +using Jellyfin.Api.Controllers; +using MediaBrowser.Controller.MediaEncoding; +using MediaBrowser.Model.Configuration; +using Xunit; + +namespace Jellyfin.Api.Tests.Controllers +{ + /// + /// Tests for the HA behaviour of : the session record it + /// registers in and the ffmpeg segmenting options it + /// picks once a session is resumed on another pod. + /// + public class DynamicHlsHaModeTests + { + /// + /// The registered session has to name the files it owns, otherwise + /// DeleteTranscodeFileTask cannot recognise and protect them. + /// + [Fact] + [Trait("Category", "UnitTest")] + public void CreateSessionRecord_PopulatesPathsFromPlaylistPath() + { + const string PlaylistPath = "/transcodes/9e1c6f.m3u8"; + + var session = DynamicHlsController.CreateSessionRecord("play-1", "media-1", PlaylistPath, TimeSpan.FromSeconds(30)); + + Assert.Equal(PlaylistPath, session.ManifestPath); + Assert.Equal("/transcodes/9e1c6f", session.SegmentPathPrefix); + Assert.Equal("play-1", session.PlaySessionId); + Assert.Equal("media-1", session.MediaSourceId); + Assert.NotEmpty(session.OwnerPod); + Assert.True(session.LeaseExpiresUtc > DateTime.UtcNow); + } + + /// + /// Every segment ffmpeg writes for the playlist has to start with the recorded prefix. + /// Segment paths are built as <playlist without extension><index><extension>. + /// + [Theory] + [InlineData(".ts")] + [InlineData(".mp4")] + [Trait("Category", "UnitTest")] + public void GetSegmentPathPrefix_CoversEverySegmentPath(string segmentExtension) + { + const string PlaylistPath = "/transcodes/9e1c6f.m3u8"; + var prefix = TranscodeSession.GetSegmentPathPrefix(PlaylistPath); + + for (var index = 0; index < 5; index++) + { + var segmentPath = Path.Combine( + Path.GetDirectoryName(PlaylistPath)!, + Path.GetFileNameWithoutExtension(PlaylistPath) + index.ToString(CultureInfo.InvariantCulture) + segmentExtension); + + Assert.StartsWith(prefix, segmentPath, StringComparison.Ordinal); + } + } + + /// + /// HA mode shortens segments and bounds the rolling buffer; the configured values are + /// user-editable so they are clamped. + /// + [Theory] + [InlineData(2, 2)] + [InlineData(0, 1)] + [InlineData(60, 6)] + [Trait("Category", "UnitTest")] + public void GetEffectiveSegmentLength_InHaMode_ClampsConfiguredRecoveryLength(int configured, int expected) + { + var options = new EncodingOptions { RecoverySegmentLengthSeconds = configured }; + + Assert.Equal(expected, DynamicHlsController.GetEffectiveSegmentLength(true, 6, options)); + } + + /// + /// Outside HA mode the requested segment length is used unchanged and the playlist is unbounded. + /// + [Fact] + [Trait("Category", "UnitTest")] + public void GetSegmentingOptions_OutsideHaMode_UsesRequestedValues() + { + var options = new EncodingOptions { RecoverySegmentLengthSeconds = 2, RecoverySegmentBufferCount = 5 }; + + Assert.Equal(6, DynamicHlsController.GetEffectiveSegmentLength(false, 6, options)); + Assert.Equal(0, DynamicHlsController.GetHlsListSize(false, options)); + } + + /// + /// HA mode keeps a bounded rolling buffer of segments for a takeover pod to serve. + /// + [Theory] + [InlineData(5, 5)] + [InlineData(0, 2)] + [InlineData(100, 10)] + [Trait("Category", "UnitTest")] + public void GetHlsListSize_InHaMode_ClampsConfiguredBufferCount(int configured, int expected) + { + var options = new EncodingOptions { RecoverySegmentBufferCount = configured }; + + Assert.Equal(expected, DynamicHlsController.GetHlsListSize(true, options)); + } + } +} diff --git a/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsHaTakeoverTests.cs b/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsHaTakeoverTests.cs deleted file mode 100644 index bdbca25d13..0000000000 --- a/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsHaTakeoverTests.cs +++ /dev/null @@ -1,259 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Linq; -using System.Threading; -using System.Threading.Tasks; -using Jellyfin.Api.Controllers; -using MediaBrowser.Controller.MediaEncoding; -using Xunit; - -namespace Jellyfin.Api.Tests.Controllers -{ - /// - /// Tests for HA recovery scenarios that will be wired into - /// in Phase 5.2. These tests verify the contract that - /// the controller will rely on for missing-local-job recovery, claim racing, and cleanup guarding. - /// - public class DynamicHlsHaTakeoverTests - { - private static TranscodeSession CreateSession(string id, string pod, DateTime leaseExpiry) - => new TranscodeSession - { - PlaySessionId = id, - OwnerPod = pod, - LeaseExpiresUtc = leaseExpiry, - ManifestPath = $"/transcode/{id}/manifest.m3u8", - SegmentPathPrefix = $"/transcode/{id}/segment", - MediaSourceId = "media-source-1", - LastCompletedSegmentIndex = 3, - LastDurablePlaybackOffset = 18_000_000L, - }; - - /// - /// Missing-local-job + durable-manifest-present: the store returns the session so - /// the controller can serve the existing manifest instead of returning an error. - /// - [Fact] - [Trait("Category", "UnitTest")] - public async Task DurableManifestPresent_WithLiveSession_StoreReturnsSession() - { - var store = new HaTestSessionStore(); - var session = CreateSession("ha-session-1", "pod-a", DateTime.UtcNow.AddMinutes(5)); - await store.SetAsync(session, TestContext.Current.CancellationToken); - - // Simulate controller recovery: look up the session in the durable store. - var recovered = await store.TryGetAsync("ha-session-1", TestContext.Current.CancellationToken); - - Assert.NotNull(recovered); - Assert.Equal("/transcode/ha-session-1/manifest.m3u8", recovered.ManifestPath); - } - - /// - /// Claim-race between two concurrent requesters: only one wins - /// . - /// The other receives false, indicating it should redirect (302) or wait. - /// - [Fact] - [Trait("Category", "UnitTest")] - public async Task ClaimRace_TwoConcurrentRequesters_OnlyOneWinsTakeover() - { - var store = new HaTestSessionStore(); - - // Original pod crashed – lease is expired. - var session = CreateSession("ha-session-2", "pod-a", DateTime.UtcNow.AddMilliseconds(-1)); - await store.SetAsync(session, TestContext.Current.CancellationToken); - - // Two pods simultaneously attempt to claim the orphaned session. - var task1 = store.TryTakeoverAsync("ha-session-2", "pod-b", TestContext.Current.CancellationToken); - var task2 = store.TryTakeoverAsync("ha-session-2", "pod-c", TestContext.Current.CancellationToken); - var results = await Task.WhenAll(task1, task2); - - // Exactly one pod must win. - var wins = Array.FindAll(results, r => r); - Assert.Single(wins); - } - - /// - /// Stale-manifest cleanup guard: a lease that has expired beyond the recovery window - /// causes the store to return null, signalling that cleanup may proceed safely. - /// - [Fact] - [Trait("Category", "UnitTest")] - public async Task StaleManifestCleanupGuard_ExpiredBeyondRecoveryWindow_StoreReturnsNull() - { - var store = new HaTestSessionStore(); - - // Lease expired hours ago – well beyond any recovery window. - var session = CreateSession("ha-session-3", "pod-a", DateTime.UtcNow.AddHours(-2)); - await store.SetAsync(session, TestContext.Current.CancellationToken); - - // Controller or cleanup task checks the store before deleting files. - var liveSession = await store.TryGetAsync("ha-session-3", TestContext.Current.CancellationToken); - - // Store returns null → cleanup may proceed without risking data loss. - Assert.Null(liveSession); - } - - /// - /// Segment-length selection: when the play-session has an active entry in the store - /// (HA mode is active), the recovery segment length should be preferred over the normal one. - /// - [Fact] - [Trait("Category", "UnitTest")] - public async Task SegmentLength_UsesRecoveryValue_WhenHaModeIsActive() - { - const int normalSegmentLength = 6; - const int recoverySegmentLength = 2; - - var store = new HaTestSessionStore(); - var session = CreateSession("ha-session-4", "pod-a", DateTime.UtcNow.AddMinutes(5)); - await store.SetAsync(session, TestContext.Current.CancellationToken); - - // Simulate the controller's HA-mode check: if the session is in the store, HA mode is active. - var existingSession = await store.TryGetAsync("ha-session-4", TestContext.Current.CancellationToken); - var isHaMode = existingSession is not null; - - var effectiveSegmentLength = isHaMode ? recoverySegmentLength : normalSegmentLength; - - Assert.True(isHaMode, "Session should be found in the store, activating HA mode."); - Assert.Equal(recoverySegmentLength, effectiveSegmentLength); - } - - /// - /// Segment-length selection: when no entry exists in the store for the play-session - /// (HA mode inactive), the normal segment length should be used. - /// - [Fact] - [Trait("Category", "UnitTest")] - public async Task SegmentLength_UsesNormalValue_WhenHaModeIsInactive() - { - const int normalSegmentLength = 6; - const int recoverySegmentLength = 2; - - var store = new HaTestSessionStore(); - - // No session registered – HA mode is inactive. - var existingSession = await store.TryGetAsync("nonexistent-session", TestContext.Current.CancellationToken); - var isHaMode = existingSession is not null; - - var effectiveSegmentLength = isHaMode ? recoverySegmentLength : normalSegmentLength; - - Assert.False(isHaMode, "No session in the store means HA mode should be inactive."); - Assert.Equal(normalSegmentLength, effectiveSegmentLength); - } - - /// - /// Minimal in-memory used within this test class - /// to avoid a cross-project reference to Jellyfin.MediaEncoding.Tests. - /// - private sealed class HaTestSessionStore : ITranscodeSessionStore - { - private static readonly TimeSpan LeaseDuration = TimeSpan.FromSeconds(30); - - private readonly Dictionary _sessions = - new(StringComparer.OrdinalIgnoreCase); - - private readonly Lock _lock = new(); - - public Task TryGetAsync(string playSessionId, CancellationToken cancellationToken = default) - { - lock (_lock) - { - if (_sessions.TryGetValue(playSessionId, out var s) && s.LeaseExpiresUtc > DateTime.UtcNow) - { - return Task.FromResult(Clone(s)); - } - - return Task.FromResult(null); - } - } - - public Task TryTakeoverAsync(string playSessionId, string claimingPod, CancellationToken cancellationToken = default) - { - lock (_lock) - { - if (!_sessions.TryGetValue(playSessionId, out var s)) - { - return Task.FromResult(false); - } - - if (s.LeaseExpiresUtc > DateTime.UtcNow) - { - return Task.FromResult(false); - } - - s.OwnerPod = claimingPod; - s.LeaseExpiresUtc = DateTime.UtcNow.Add(LeaseDuration); - return Task.FromResult(true); - } - } - - public Task SetAsync(TranscodeSession session, CancellationToken cancellationToken = default) - { - lock (_lock) - { - _sessions[session.PlaySessionId] = session; - } - - return Task.CompletedTask; - } - - public Task RenewLeaseAsync(string playSessionId, CancellationToken cancellationToken = default) - { - lock (_lock) - { - if (_sessions.TryGetValue(playSessionId, out var s)) - { - s.LeaseExpiresUtc = DateTime.UtcNow.Add(LeaseDuration); - } - } - - return Task.CompletedTask; - } - - public Task DeleteAsync(string playSessionId, CancellationToken cancellationToken = default) - { - lock (_lock) - { - _sessions.Remove(playSessionId); - } - - return Task.CompletedTask; - } - - public Task> GetActiveSessionsAsync(CancellationToken cancellationToken = default) - { - lock (_lock) - { - var sessions = _sessions.Values - .Where(s => s.LeaseExpiresUtc > DateTime.UtcNow) - .Select(Clone) - .ToList(); - return Task.FromResult>(sessions); - } - } - - public Task SetLiveStreamAsync(LiveStreamSession session, CancellationToken cancellationToken = default) - => Task.CompletedTask; - - public Task TryGetLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default) - => Task.FromResult(null); - - public Task DeleteLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default) - => Task.CompletedTask; - - private static TranscodeSession Clone(TranscodeSession source) - => new TranscodeSession - { - PlaySessionId = source.PlaySessionId, - OwnerPod = source.OwnerPod, - LeaseExpiresUtc = source.LeaseExpiresUtc, - ManifestPath = source.ManifestPath, - SegmentPathPrefix = source.SegmentPathPrefix, - MediaSourceId = source.MediaSourceId, - LastCompletedSegmentIndex = source.LastCompletedSegmentIndex, - LastDurablePlaybackOffset = source.LastDurablePlaybackOffset, - }; - } - } -} diff --git a/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsSessionRegistrationTests.cs b/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsSessionRegistrationTests.cs deleted file mode 100644 index cb09c00ff5..0000000000 --- a/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsSessionRegistrationTests.cs +++ /dev/null @@ -1,223 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Linq; -using System.Threading; -using System.Threading.Tasks; -using MediaBrowser.Controller.MediaEncoding; -using Xunit; - -namespace Jellyfin.Api.Tests.Controllers -{ - /// - /// Tests for HLS session registration, lease renewal, and cleanup behaviour wired into - /// in Phase 5.2.2a. - /// These tests verify the contract used by the controller. - /// - public class DynamicHlsSessionRegistrationTests - { - private static TranscodeSession CreateSession(string id, string pod) - => new TranscodeSession - { - PlaySessionId = id, - OwnerPod = pod, - LeaseExpiresUtc = DateTime.UtcNow.AddSeconds(30), - ManifestPath = string.Empty, - SegmentPathPrefix = string.Empty, - MediaSourceId = "media-source-1", - LastCompletedSegmentIndex = 0, - LastDurablePlaybackOffset = 0L, - }; - - /// - /// After registering a session via , - /// must return a non-null result with - /// matching and . - /// - [Fact] - [Trait("Category", "UnitTest")] - public async Task SessionRegistration_AfterStreamStart_StoreContainsSession() - { - var store = new InMemoryTranscodeSessionStore(); - var session = CreateSession("session-reg-1", "pod-a"); - - await store.SetAsync(session, TestContext.Current.CancellationToken); - - var retrieved = await store.TryGetAsync("session-reg-1", TestContext.Current.CancellationToken); - - Assert.NotNull(retrieved); - Assert.Equal("session-reg-1", retrieved.PlaySessionId); - Assert.Equal("pod-a", retrieved.OwnerPod); - } - - /// - /// After calling , - /// must return null. - /// - [Fact] - [Trait("Category", "UnitTest")] - public async Task SessionCleanup_AfterStreamEnd_StoreReturnsNull() - { - var store = new InMemoryTranscodeSessionStore(); - var session = CreateSession("session-cleanup-1", "pod-b"); - - await store.SetAsync(session, TestContext.Current.CancellationToken); - await store.DeleteAsync("session-cleanup-1", TestContext.Current.CancellationToken); - - var retrieved = await store.TryGetAsync("session-cleanup-1", TestContext.Current.CancellationToken); - - Assert.Null(retrieved); - } - - /// - /// After a session's initial lease window would have expired, calling - /// must extend the lease so that - /// still returns the session as active. - /// - [Fact] - [Trait("Category", "UnitTest")] - public async Task LeaseRenewal_ExtendsBeyondInitialExpiry() - { - var store = new InMemoryTranscodeSessionStore(); - - // Create the session with a lease that has already expired. - var session = new TranscodeSession - { - PlaySessionId = "session-renewal-1", - OwnerPod = "pod-c", - LeaseExpiresUtc = DateTime.UtcNow.AddMilliseconds(-1), - ManifestPath = string.Empty, - SegmentPathPrefix = string.Empty, - MediaSourceId = "media-source-1", - LastCompletedSegmentIndex = 0, - LastDurablePlaybackOffset = 0L, - }; - await store.SetAsync(session, TestContext.Current.CancellationToken); - - // Verify the session is not accessible because the lease has expired. - Assert.Null(await store.TryGetAsync("session-renewal-1", TestContext.Current.CancellationToken)); - - // Renew the lease. - await store.RenewLeaseAsync("session-renewal-1", TestContext.Current.CancellationToken); - - // After renewal the session must be accessible again. - var renewed = await store.TryGetAsync("session-renewal-1", TestContext.Current.CancellationToken); - Assert.NotNull(renewed); - Assert.Equal("session-renewal-1", renewed.PlaySessionId); - Assert.True(renewed.LeaseExpiresUtc > DateTime.UtcNow); - } - - /// - /// Minimal thread-safe in-memory implementation of - /// used within this test class to avoid a cross-project reference. - /// - private sealed class InMemoryTranscodeSessionStore : ITranscodeSessionStore - { - private static readonly TimeSpan DefaultLeaseDuration = TimeSpan.FromSeconds(30); - - private readonly Dictionary _sessions = - new(StringComparer.OrdinalIgnoreCase); - - private readonly Lock _lock = new(); - - public Task TryGetAsync(string playSessionId, CancellationToken cancellationToken = default) - { - lock (_lock) - { - if (_sessions.TryGetValue(playSessionId, out var session) && session.LeaseExpiresUtc > DateTime.UtcNow) - { - return Task.FromResult(Clone(session)); - } - - return Task.FromResult(null); - } - } - - public Task TryTakeoverAsync(string playSessionId, string claimingPod, CancellationToken cancellationToken = default) - { - lock (_lock) - { - if (!_sessions.TryGetValue(playSessionId, out var session)) - { - return Task.FromResult(false); - } - - if (session.LeaseExpiresUtc > DateTime.UtcNow) - { - return Task.FromResult(false); - } - - session.OwnerPod = claimingPod; - session.LeaseExpiresUtc = DateTime.UtcNow.Add(DefaultLeaseDuration); - return Task.FromResult(true); - } - } - - public Task SetAsync(TranscodeSession session, CancellationToken cancellationToken = default) - { - lock (_lock) - { - _sessions[session.PlaySessionId] = session; - } - - return Task.CompletedTask; - } - - public Task RenewLeaseAsync(string playSessionId, CancellationToken cancellationToken = default) - { - lock (_lock) - { - if (_sessions.TryGetValue(playSessionId, out var session)) - { - session.LeaseExpiresUtc = DateTime.UtcNow.Add(DefaultLeaseDuration); - } - } - - return Task.CompletedTask; - } - - public Task DeleteAsync(string playSessionId, CancellationToken cancellationToken = default) - { - lock (_lock) - { - _sessions.Remove(playSessionId); - } - - return Task.CompletedTask; - } - - public Task> GetActiveSessionsAsync(CancellationToken cancellationToken = default) - { - lock (_lock) - { - var sessions = _sessions.Values - .Where(s => s.LeaseExpiresUtc > DateTime.UtcNow) - .Select(Clone) - .ToList(); - return Task.FromResult>(sessions); - } - } - - public Task SetLiveStreamAsync(LiveStreamSession session, CancellationToken cancellationToken = default) - => Task.CompletedTask; - - public Task TryGetLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default) - => Task.FromResult(null); - - public Task DeleteLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default) - => Task.CompletedTask; - - private static TranscodeSession Clone(TranscodeSession source) - => new TranscodeSession - { - PlaySessionId = source.PlaySessionId, - OwnerPod = source.OwnerPod, - LeaseExpiresUtc = source.LeaseExpiresUtc, - ManifestPath = source.ManifestPath, - SegmentPathPrefix = source.SegmentPathPrefix, - MediaSourceId = source.MediaSourceId, - LastCompletedSegmentIndex = source.LastCompletedSegmentIndex, - LastDurablePlaybackOffset = source.LastDurablePlaybackOffset, - }; - } - } -} diff --git a/tests/Jellyfin.MediaEncoding.Tests/Fakes/InMemoryTranscodeSessionStore.cs b/tests/Jellyfin.MediaEncoding.Tests/Fakes/InMemoryTranscodeSessionStore.cs index af3b89c350..e0dfb60db5 100644 --- a/tests/Jellyfin.MediaEncoding.Tests/Fakes/InMemoryTranscodeSessionStore.cs +++ b/tests/Jellyfin.MediaEncoding.Tests/Fakes/InMemoryTranscodeSessionStore.cs @@ -18,7 +18,6 @@ public sealed class InMemoryTranscodeSessionStore : ITranscodeSessionStore 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(); /// @@ -70,17 +69,19 @@ public sealed class InMemoryTranscodeSessionStore : ITranscodeSessionStore } /// - public Task RenewLeaseAsync(string playSessionId, CancellationToken cancellationToken = default) + public Task RenewLeaseAsync(string playSessionId, string ownerPod, CancellationToken cancellationToken = default) { lock (_lock) { - if (_sessions.TryGetValue(playSessionId, out var session)) + if (!_sessions.TryGetValue(playSessionId, out var session) + || !string.Equals(session.OwnerPod, ownerPod, StringComparison.Ordinal)) { - session.LeaseExpiresUtc = DateTime.UtcNow.Add(DefaultLeaseDuration); + return Task.FromResult(false); } - } - return Task.CompletedTask; + session.LeaseExpiresUtc = DateTime.UtcNow.Add(DefaultLeaseDuration); + return Task.FromResult(true); + } } /// @@ -104,56 +105,6 @@ public sealed class InMemoryTranscodeSessionStore : ITranscodeSessionStore } } - /// - 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 { diff --git a/tests/Jellyfin.MediaEncoding.Tests/Transcoding/TranscodeManagerTests.cs b/tests/Jellyfin.MediaEncoding.Tests/Transcoding/InMemoryTranscodeSessionStoreTests.cs similarity index 81% rename from tests/Jellyfin.MediaEncoding.Tests/Transcoding/TranscodeManagerTests.cs rename to tests/Jellyfin.MediaEncoding.Tests/Transcoding/InMemoryTranscodeSessionStoreTests.cs index 8ea94edb9b..b84944cb23 100644 --- a/tests/Jellyfin.MediaEncoding.Tests/Transcoding/TranscodeManagerTests.cs +++ b/tests/Jellyfin.MediaEncoding.Tests/Transcoding/InMemoryTranscodeSessionStoreTests.cs @@ -7,12 +7,12 @@ 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. +/// Unit tests for the contract — lease expiry, double-claim +/// prevention, ownership-checked renewal and stale-session cleanup — against +/// , the reference implementation. The Redis-backed +/// implementation is covered by RedisTranscodeSessionStoreTests against a real Redis. /// -public class TranscodeManagerTests +public class InMemoryTranscodeSessionStoreTests { private static TranscodeSession CreateSession( string id, @@ -123,7 +123,7 @@ public class TranscodeManagerTests 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); + Assert.True(await store.RenewLeaseAsync("session-renew", "pod-a", TestContext.Current.CancellationToken)); var renewed = await store.TryGetAsync("session-renew", TestContext.Current.CancellationToken); Assert.NotNull(renewed); @@ -132,6 +132,25 @@ public class TranscodeManagerTests "Renewed lease expiry should be later than the original expiry."); } + /// + /// A renewal from a pod that no longer owns the lease must fail and must not revert ownership. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task RenewLeaseAsync_ByNonOwner_ReturnsFalse() + { + var store = new InMemoryTranscodeSessionStore(); + var session = CreateSession("session-renew-other", "pod-a", DateTime.UtcNow.AddMilliseconds(-1)); + await store.SetAsync(session, TestContext.Current.CancellationToken); + + Assert.True(await store.TryTakeoverAsync("session-renew-other", "pod-b", TestContext.Current.CancellationToken)); + Assert.False(await store.RenewLeaseAsync("session-renew-other", "pod-a", TestContext.Current.CancellationToken)); + + var current = await store.TryGetAsync("session-renew-other", TestContext.Current.CancellationToken); + Assert.NotNull(current); + Assert.Equal("pod-b", current.OwnerPod); + } + /// /// Stale-session cleanup: an expired session can be deleted without error, and a /// subsequent returns null. diff --git a/tests/Jellyfin.Server.Implementations.Tests/Jellyfin.Server.Implementations.Tests.csproj b/tests/Jellyfin.Server.Implementations.Tests/Jellyfin.Server.Implementations.Tests.csproj index 29de52a2ba..f8d35c39af 100644 --- a/tests/Jellyfin.Server.Implementations.Tests/Jellyfin.Server.Implementations.Tests.csproj +++ b/tests/Jellyfin.Server.Implementations.Tests/Jellyfin.Server.Implementations.Tests.csproj @@ -18,6 +18,8 @@ + + all diff --git a/tests/Jellyfin.Server.Implementations.Tests/MediaEncoding/RedisTranscodeSessionStoreTests.cs b/tests/Jellyfin.Server.Implementations.Tests/MediaEncoding/RedisTranscodeSessionStoreTests.cs index 086ce535e2..d3218113ef 100644 --- a/tests/Jellyfin.Server.Implementations.Tests/MediaEncoding/RedisTranscodeSessionStoreTests.cs +++ b/tests/Jellyfin.Server.Implementations.Tests/MediaEncoding/RedisTranscodeSessionStoreTests.cs @@ -1,384 +1,285 @@ using System; -using System.Collections.Generic; +using System.Globalization; using System.Linq; -using System.Threading; using System.Threading.Tasks; +using Emby.Server.Implementations.MediaEncoding; using MediaBrowser.Controller.MediaEncoding; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using StackExchange.Redis; +using Testcontainers.Redis; using Xunit; namespace Jellyfin.Server.Implementations.Tests.MediaEncoding; /// -/// Tests for transcode session store contract behavior, using -/// as a reference implementation (no real Redis required). +/// Integration tests for and its Lua scripts against a +/// real Redis container. /// -public class RedisTranscodeSessionStoreTests +[Trait("Category", "RequiresDocker")] +public sealed class RedisTranscodeSessionStoreTests : IAsyncLifetime { + private readonly RedisContainer _container; + private IConnectionMultiplexer? _redis; + /// - /// Verifies that returns null after - /// a session's lease has expired. + /// Initializes a new instance of the class. /// - [Fact] - [Trait("Category", "UnitTest")] - public async Task TryGetAsync_AfterLeaseExpires_ReturnsNull() + public RedisTranscodeSessionStoreTests() { - 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); + _container = new RedisBuilder("redis:7-alpine").Build(); } /// - /// Verifies that returns false - /// when the session's lease is still valid. + /// Starts the Redis container before any tests in the class run. /// + /// A representing the asynchronous operation. + public async ValueTask InitializeAsync() + { + await _container.StartAsync().ConfigureAwait(false); + _redis = await ConnectionMultiplexer.ConnectAsync(_container.GetConnectionString()).ConfigureAwait(false); + } + + /// + /// Stops and removes the Redis container after all tests in the class have run. + /// + /// A representing the asynchronous operation. + public async ValueTask DisposeAsync() + { + if (_redis is not null) + { + await _redis.DisposeAsync().ConfigureAwait(false); + } + + await _container.DisposeAsync().ConfigureAwait(false); + } + + /// + /// A stored session round-trips through Redis with the paths cleanup relies on intact. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task SetAsync_RoundTripsSession() + { + var store = CreateStore(); + var id = NewSessionId(); + var session = TranscodeSession.CreateForPlaylist(id, "media-1", "pod-a", "/transcodes/abc.m3u8", TimeSpan.FromSeconds(30)); + + await store.SetAsync(session, TestContext.Current.CancellationToken); + + var stored = await store.TryGetAsync(id, TestContext.Current.CancellationToken); + + Assert.NotNull(stored); + Assert.Equal("pod-a", stored.OwnerPod); + Assert.Equal("media-1", stored.MediaSourceId); + Assert.Equal("/transcodes/abc.m3u8", stored.ManifestPath); + Assert.Equal("/transcodes/abc", stored.SegmentPathPrefix); + Assert.True(stored.LeaseExpiresUtc > DateTime.UtcNow); + } + + /// + /// The owning pod can extend its own lease. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task RenewLeaseAsync_ByOwner_ExtendsLease() + { + var store = CreateStore(leaseSeconds: 4); + var id = NewSessionId(); + await store.SetAsync(NewSession(id, "pod-a", TimeSpan.FromSeconds(4)), TestContext.Current.CancellationToken); + + var before = await store.TryGetAsync(id, TestContext.Current.CancellationToken); + await Task.Delay(TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + Assert.True(await store.RenewLeaseAsync(id, "pod-a", TestContext.Current.CancellationToken)); + + var after = await store.TryGetAsync(id, TestContext.Current.CancellationToken); + Assert.NotNull(before); + Assert.NotNull(after); + Assert.True(after.LeaseExpiresUtc > before.LeaseExpiresUtc); + Assert.Equal("pod-a", after.OwnerPod); + } + + /// + /// A pod that does not own the lease cannot renew it, and its attempt leaves the owner alone. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task RenewLeaseAsync_ByNonOwner_ReturnsFalse() + { + var store = CreateStore(); + var id = NewSessionId(); + await store.SetAsync(NewSession(id, "pod-a", TimeSpan.FromSeconds(30)), TestContext.Current.CancellationToken); + + Assert.False(await store.RenewLeaseAsync(id, "pod-b", TestContext.Current.CancellationToken)); + + var stored = await store.TryGetAsync(id, TestContext.Current.CancellationToken); + Assert.NotNull(stored); + Assert.Equal("pod-a", stored.OwnerPod); + } + + /// + /// Renewal of a session that is gone fails instead of recreating it. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task RenewLeaseAsync_AfterDelete_ReturnsFalse() + { + var store = CreateStore(); + var id = NewSessionId(); + await store.SetAsync(NewSession(id, "pod-a", TimeSpan.FromSeconds(30)), TestContext.Current.CancellationToken); + await store.DeleteAsync(id, TestContext.Current.CancellationToken); + + Assert.False(await store.RenewLeaseAsync(id, "pod-a", TestContext.Current.CancellationToken)); + Assert.Null(await store.TryGetAsync(id, TestContext.Current.CancellationToken)); + } + + /// + /// A valid lease blocks takeover. + /// + /// A representing the asynchronous operation. [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), - }; + var store = CreateStore(); + var id = NewSessionId(); + await store.SetAsync(NewSession(id, "pod-a", TimeSpan.FromSeconds(30)), TestContext.Current.CancellationToken); - await store.SetAsync(session, TestContext.Current.CancellationToken); + Assert.False(await store.TryTakeoverAsync(id, "pod-b", TestContext.Current.CancellationToken)); - var result = await store.TryTakeoverAsync("session-2", "pod-b", TestContext.Current.CancellationToken); - - Assert.False(result); + var stored = await store.TryGetAsync(id, TestContext.Current.CancellationToken); + Assert.NotNull(stored); + Assert.Equal("pod-a", stored.OwnerPod); } /// - /// Verifies that returns true - /// and updates the owner when the session's lease has expired. + /// Once the lease expires the session record is still retained, so another pod can claim it. /// + /// A representing the asynchronous operation. [Fact] - [Trait("Category", "UnitTest")] - public async Task TryTakeoverAsync_AfterLeaseExpires_ReturnsTrue_AndUpdatesOwner() + public async Task TryTakeoverAsync_AfterLeaseExpires_TransfersOwnership() { - var store = new InMemoryTranscodeSessionStore(); - var session = new TranscodeSession - { - PlaySessionId = "session-3", - OwnerPod = "pod-a", - LeaseExpiresUtc = DateTime.UtcNow.AddSeconds(-1), - }; + var store = CreateStore(leaseSeconds: 1); + var id = NewSessionId(); + await store.SetAsync(NewSession(id, "pod-a", TimeSpan.FromSeconds(1)), TestContext.Current.CancellationToken); - await store.SetAsync(session, TestContext.Current.CancellationToken); + await Task.Delay(TimeSpan.FromMilliseconds(1200), TestContext.Current.CancellationToken); + Assert.Null(await store.TryGetAsync(id, TestContext.Current.CancellationToken)); - var result = await store.TryTakeoverAsync("session-3", "pod-b", TestContext.Current.CancellationToken); + Assert.True(await store.TryTakeoverAsync(id, "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); + var stored = await store.TryGetAsync(id, TestContext.Current.CancellationToken); + Assert.NotNull(stored); + Assert.Equal("pod-b", stored.OwnerPod); + Assert.Equal("/transcodes/" + id + ".m3u8", stored.ManifestPath); } /// - /// Verifies that when multiple pods concurrently attempt to take over an expired session, - /// exactly one succeeds. + /// Only one of several pods racing for an expired lease wins it. /// + /// A representing the asynchronous operation. [Fact] - [Trait("Category", "UnitTest")] - public async Task ConcurrentTryTakeover_OnlyOneWins() + public async Task TryTakeoverAsync_ConcurrentClaims_OnlyOneWins() { - var store = new InMemoryTranscodeSessionStore(); - var session = new TranscodeSession - { - PlaySessionId = "session-4", - OwnerPod = "pod-a", - LeaseExpiresUtc = DateTime.UtcNow.AddSeconds(-1), - }; + var store = CreateStore(leaseSeconds: 1); + var id = NewSessionId(); + await store.SetAsync(NewSession(id, "pod-a", TimeSpan.FromSeconds(1)), TestContext.Current.CancellationToken); + await Task.Delay(TimeSpan.FromMilliseconds(1200), TestContext.Current.CancellationToken); - await store.SetAsync(session, TestContext.Current.CancellationToken); + var claims = Enumerable.Range(0, 10) + .Select(i => store.TryTakeoverAsync(id, "pod-" + i.ToString(CultureInfo.InvariantCulture), TestContext.Current.CancellationToken)) + .ToList(); - 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(claims); - var results = await Task.WhenAll(tasks); - - var successCount = 0; - foreach (var r in results) - { - if (r) - { - successCount++; - } - } - - Assert.Equal(1, successCount); + Assert.Equal(1, results.Count(won => won)); } /// - /// Verifies that stores a live stream - /// record that can be retrieved by session id. + /// The race behind the non-atomic renewal: after another pod wins the takeover, a renewal from + /// the previous owner must fail rather than restore its own ownership and lease. /// + /// A representing the asynchronous operation. [Fact] - [Trait("Category", "UnitTest")] - public async Task SetLiveStreamAsync_CanBeRetrievedBySessionId() + public async Task RenewLeaseAsync_AfterLosingTakeover_DoesNotRevertOwner() { - var store = new InMemoryTranscodeSessionStore(); - var liveStream = new LiveStreamSession - { - LiveStreamId = "stream-1", - SessionId = "session-a", - PlaySessionId = "play-session-a", - OwnerPod = "pod-a", - OpenedAtUtc = DateTime.UtcNow, - }; + var store = CreateStore(leaseSeconds: 1); + var id = NewSessionId(); + await store.SetAsync(NewSession(id, "pod-a", TimeSpan.FromSeconds(1)), TestContext.Current.CancellationToken); + await Task.Delay(TimeSpan.FromMilliseconds(1200), TestContext.Current.CancellationToken); - await store.SetLiveStreamAsync(liveStream, TestContext.Current.CancellationToken); + Assert.True(await store.TryTakeoverAsync(id, "pod-b", TestContext.Current.CancellationToken)); + Assert.False(await store.RenewLeaseAsync(id, "pod-a", 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); + var stored = await store.TryGetAsync(id, TestContext.Current.CancellationToken); + Assert.NotNull(stored); + Assert.Equal("pod-b", stored.OwnerPod); } /// - /// Verifies that stores a live stream - /// record that can be retrieved by play session id. + /// Renewal and takeover racing at the moment the lease expires always leave exactly one owner: + /// the claiming pod when the takeover wins, the original pod when the renewal got in first. /// + /// A representing the asynchronous operation. [Fact] - [Trait("Category", "UnitTest")] - public async Task SetLiveStreamAsync_CanBeRetrievedByPlaySessionId() + public async Task RenewLeaseAsync_RacingTakeover_LeavesSingleConsistentOwner() { - var store = new InMemoryTranscodeSessionStore(); - var liveStream = new LiveStreamSession + var store = CreateStore(leaseSeconds: 1); + + for (var i = 0; i < 5; i++) { - LiveStreamId = "stream-2", - SessionId = "session-b", - PlaySessionId = "play-session-b", - OwnerPod = "pod-a", - OpenedAtUtc = DateTime.UtcNow, - }; + var id = NewSessionId(); + await store.SetAsync(NewSession(id, "pod-a", TimeSpan.FromSeconds(1)), TestContext.Current.CancellationToken); - await store.SetLiveStreamAsync(liveStream, TestContext.Current.CancellationToken); + // Both calls are issued at the expiry boundary so their order is genuinely undefined. + await Task.Delay(TimeSpan.FromMilliseconds(1000), TestContext.Current.CancellationToken); + var renewal = store.RenewLeaseAsync(id, "pod-a", TestContext.Current.CancellationToken); + var takeover = store.TryTakeoverAsync(id, "pod-b", 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); + var renewed = await renewal; + var tookOver = await takeover; + + var stored = await store.TryGetAsync(id, TestContext.Current.CancellationToken); + Assert.NotNull(stored); + Assert.False(renewed && tookOver, "A renewal and a takeover must not both succeed for the same lease."); + Assert.Equal(tookOver ? "pod-b" : "pod-a", stored.OwnerPod); + } } /// - /// Verifies that removes the live - /// stream record so that subsequent lookups by either session id or play session id return null. + /// Cleanup reads the active sessions, so a session whose lease has lapsed must not be reported + /// as active even while its record is retained for takeover. /// + /// A representing the asynchronous operation. [Fact] - [Trait("Category", "UnitTest")] - public async Task DeleteLiveStreamAsync_RemovesBothKeys() + public async Task GetActiveSessionsAsync_ExcludesExpiredLeases() { - var store = new InMemoryTranscodeSessionStore(); - var liveStream = new LiveStreamSession - { - LiveStreamId = "stream-3", - SessionId = "session-c", - PlaySessionId = "play-session-c", - OwnerPod = "pod-a", - OpenedAtUtc = DateTime.UtcNow, - }; + var store = CreateStore(leaseSeconds: 1); + var expiredId = NewSessionId(); + await store.SetAsync(NewSession(expiredId, "pod-a", TimeSpan.FromSeconds(1)), TestContext.Current.CancellationToken); + await Task.Delay(TimeSpan.FromMilliseconds(1200), TestContext.Current.CancellationToken); - await store.SetLiveStreamAsync(liveStream, TestContext.Current.CancellationToken); - await store.DeleteLiveStreamAsync("stream-3", "session-c", TestContext.Current.CancellationToken); + var liveId = NewSessionId(); + await store.SetAsync(NewSession(liveId, "pod-a", TimeSpan.FromSeconds(30)), 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); + var active = (await store.GetActiveSessionsAsync(TestContext.Current.CancellationToken)).ToList(); - Assert.Null(bySessionId); - Assert.Null(byPlaySessionId); + Assert.Contains(active, s => string.Equals(s.PlaySessionId, liveId, StringComparison.Ordinal)); + Assert.DoesNotContain(active, s => string.Equals(s.PlaySessionId, expiredId, StringComparison.Ordinal)); } - /// - /// Verifies that returns null when - /// no matching record exists. - /// - [Fact] - [Trait("Category", "UnitTest")] - public async Task TryGetLiveStreamAsync_WhenNotPresent_ReturnsNull() - { - var store = new InMemoryTranscodeSessionStore(); + private static string NewSessionId() => "play-" + Guid.NewGuid().ToString("N"); - var result = await store.TryGetLiveStreamAsync("nonexistent-stream", "nonexistent-session", TestContext.Current.CancellationToken); + private static TranscodeSession NewSession(string id, string pod, TimeSpan leaseDuration) + => TranscodeSession.CreateForPlaylist(id, "media-" + id, pod, "/transcodes/" + id + ".m3u8", leaseDuration); - 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) + private RedisTranscodeSessionStore CreateStore(int leaseSeconds = 30, int retentionSeconds = 300) + => new RedisTranscodeSessionStore( + _redis!, + Options.Create(new TranscodeStoreOptions { - 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, - }; - } + LeaseDurationSeconds = leaseSeconds, + SessionRetentionSeconds = retentionSeconds + }), + NullLogger.Instance); } diff --git a/tests/Jellyfin.Server.Implementations.Tests/ScheduledTasks/DeleteTranscodeFileTaskTests.cs b/tests/Jellyfin.Server.Implementations.Tests/ScheduledTasks/DeleteTranscodeFileTaskTests.cs index 0abf2f5e3d..30b3485883 100644 --- a/tests/Jellyfin.Server.Implementations.Tests/ScheduledTasks/DeleteTranscodeFileTaskTests.cs +++ b/tests/Jellyfin.Server.Implementations.Tests/ScheduledTasks/DeleteTranscodeFileTaskTests.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.IO; using System.Linq; using System.Threading; using System.Threading.Tasks; @@ -13,13 +14,9 @@ using Xunit; namespace Jellyfin.Server.Implementations.Tests.ScheduledTasks; /// -/// Tests for lease-aware cleanup behavior expected of DeleteTranscodeFileTask once -/// it is made HA-aware in Phase 5.2. -/// -/// The current DeleteTranscodeFileTask implementation uses file-age only and does not -/// check , which creates a data-loss risk on shared NFS -/// storage. These tests document the correct contract by exercising the store directly. -/// +/// Tests for the lease-aware cleanup behaviour of DeleteTranscodeFileTask: files belonging +/// to a session whose lease is still live must survive a cleanup pass, and a store failure must +/// abort the pass rather than risk deleting files in use. /// public class DeleteTranscodeFileTaskTests { @@ -313,6 +310,64 @@ public class DeleteTranscodeFileTaskTests Assert.Empty(deletedFiles); } + /// + /// The files of a live session, named exactly as the HLS pipeline writes them and recorded by + /// the same production factory the controller uses, survive a cleanup pass while an unrelated + /// stale file from a finished session is deleted. + /// + [Fact] + [Trait("Category", "UnitTest")] + public async Task ExecuteAsync_WithLiveSessionRegisteredByProduction_KeepsItsFilesAndDeletesTheRest() + { + const string TranscodePath = "/transcode"; + var playlistPath = Path.Combine(TranscodePath, "9e1c6f.m3u8"); + var session = TranscodeSession.CreateForPlaylist("play-1", "media-1", "pod-a", playlistPath, TimeSpan.FromMinutes(5)); + + var sessionFiles = new[] + { + playlistPath, + TranscodeSession.GetSegmentPathPrefix(playlistPath) + "0.ts", + TranscodeSession.GetSegmentPathPrefix(playlistPath) + "1.ts", + TranscodeSession.GetSegmentPathPrefix(playlistPath) + "-1.mp4", + }; + + var orphanedFile = Path.Combine(TranscodePath, "abandoned.m3u8"); + + var store = new CleanupTestSessionStore(); + await store.SetAsync(session, TestContext.Current.CancellationToken); + + var deletedFiles = new List(); + var fileSystemMock = new Mock(); + fileSystemMock + .Setup(fs => fs.GetFiles(TranscodePath, true)) + .Returns(sessionFiles.Append(orphanedFile).Select(path => new FileSystemMetadata { FullName = path, IsDirectory = false })); + fileSystemMock + .Setup(fs => fs.GetLastWriteTimeUtc(It.IsAny())) + .Returns(DateTime.UtcNow.AddDays(-2)); + fileSystemMock + .Setup(fs => fs.DeleteFile(It.IsAny())) + .Callback(deletedFiles.Add); + fileSystemMock + .Setup(fs => fs.GetDirectories(It.IsAny(), It.IsAny())) + .Returns(Enumerable.Empty()); + + var localizationMock = new Mock(); + localizationMock + .Setup(l => l.GetLocalizedString(It.IsAny())) + .Returns(key => key); + + var task = new Emby.Server.Implementations.ScheduledTasks.Tasks.DeleteTranscodeFileTask( + new Mock>().Object, + fileSystemMock.Object, + CreateConfigMock(TranscodePath).Object, + localizationMock.Object, + store); + + await task.ExecuteAsync(new Progress(), CancellationToken.None); + + Assert.Equal(new[] { orphanedFile }, deletedFiles); + } + /// /// Minimal in-memory used within this test class /// to avoid a cross-project reference to Jellyfin.MediaEncoding.Tests. @@ -369,17 +424,19 @@ public class DeleteTranscodeFileTaskTests return Task.CompletedTask; } - public Task RenewLeaseAsync(string playSessionId, CancellationToken cancellationToken = default) + public Task RenewLeaseAsync(string playSessionId, string ownerPod, CancellationToken cancellationToken = default) { lock (_lock) { - if (_sessions.TryGetValue(playSessionId, out var s)) + if (!_sessions.TryGetValue(playSessionId, out var s) + || !string.Equals(s.OwnerPod, ownerPod, StringComparison.Ordinal)) { - s.LeaseExpiresUtc = DateTime.UtcNow.Add(LeaseDuration); + return Task.FromResult(false); } - } - return Task.CompletedTask; + s.LeaseExpiresUtc = DateTime.UtcNow.Add(LeaseDuration); + return Task.FromResult(true); + } } public Task DeleteAsync(string playSessionId, CancellationToken cancellationToken = default) @@ -404,15 +461,6 @@ public class DeleteTranscodeFileTaskTests } } - public Task SetLiveStreamAsync(LiveStreamSession session, CancellationToken cancellationToken = default) - => Task.CompletedTask; - - public Task TryGetLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default) - => Task.FromResult(null); - - public Task DeleteLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default) - => Task.CompletedTask; - private static TranscodeSession Clone(TranscodeSession source) => new TranscodeSession { diff --git a/tests/Jellyfin.Server.Implementations.Tests/SessionManager/IdlePlaybackTests.cs b/tests/Jellyfin.Server.Implementations.Tests/SessionManager/IdlePlaybackTests.cs index 6466c02dc0..7722707cbe 100644 --- a/tests/Jellyfin.Server.Implementations.Tests/SessionManager/IdlePlaybackTests.cs +++ b/tests/Jellyfin.Server.Implementations.Tests/SessionManager/IdlePlaybackTests.cs @@ -8,7 +8,6 @@ using MediaBrowser.Controller.Drawing; using MediaBrowser.Controller.Dto; using MediaBrowser.Controller.Events; using MediaBrowser.Controller.Library; -using MediaBrowser.Controller.MediaEncoding; using MediaBrowser.Controller.Session; using MediaBrowser.Model.Dto; using MediaBrowser.Model.Session; @@ -45,8 +44,7 @@ public class IdlePlaybackTests Mock.Of(), Mock.Of(), Mock.Of(), - Mock.Of(), - new NullTranscodeSessionStore()); + Mock.Of()); var session = await sessionManager.LogSessionActivity( "Test Client", "1.0.0", diff --git a/tests/Jellyfin.Server.Implementations.Tests/SessionManager/LiveStreamHaRecordTests.cs b/tests/Jellyfin.Server.Implementations.Tests/SessionManager/LiveStreamHaRecordTests.cs deleted file mode 100644 index 4ea343afdc..0000000000 --- a/tests/Jellyfin.Server.Implementations.Tests/SessionManager/LiveStreamHaRecordTests.cs +++ /dev/null @@ -1,66 +0,0 @@ -using System; -using System.Threading; -using System.Threading.Tasks; -using MediaBrowser.Controller; -using MediaBrowser.Controller.Configuration; -using MediaBrowser.Controller.Devices; -using MediaBrowser.Controller.Drawing; -using MediaBrowser.Controller.Dto; -using MediaBrowser.Controller.Events; -using MediaBrowser.Controller.Library; -using MediaBrowser.Controller.MediaEncoding; -using Microsoft.Extensions.Hosting; -using Microsoft.Extensions.Logging.Abstractions; -using Moq; -using Xunit; - -namespace Jellyfin.Server.Implementations.Tests.SessionManager; - -public class LiveStreamHaRecordTests -{ - [Fact] - public async Task CloseLiveStreamIfNeededAsync_Should_DeleteDurableRecord() - { - var store = new Mock(); - var mediaSourceManager = new Mock(); - await using var sessionManager = CreateSessionManager(store.Object, mediaSourceManager.Object); - - await sessionManager.CloseLiveStreamIfNeededAsync("stream-1", "session-1"); - - store.Verify(s => s.DeleteLiveStreamAsync("stream-1", "session-1", It.IsAny()), Times.Once); - mediaSourceManager.Verify(m => m.CloseLiveStream("stream-1"), Times.Once); - } - - [Fact] - public async Task CloseLiveStreamIfNeededAsync_Should_CloseStream_WhenDurableStoreFails() - { - var store = new Mock(); - store.Setup(s => s.DeleteLiveStreamAsync(It.IsAny(), It.IsAny(), It.IsAny())) - .ThrowsAsync(new InvalidOperationException("redis unreachable")); - var mediaSourceManager = new Mock(); - await using var sessionManager = CreateSessionManager(store.Object, mediaSourceManager.Object); - - await sessionManager.CloseLiveStreamIfNeededAsync("stream-2", "session-2"); - - mediaSourceManager.Verify(m => m.CloseLiveStream("stream-2"), Times.Once); - } - - private static Emby.Server.Implementations.Session.SessionManager CreateSessionManager( - ITranscodeSessionStore transcodeSessionStore, - IMediaSourceManager mediaSourceManager) - => new Emby.Server.Implementations.Session.SessionManager( - NullLogger.Instance, - Mock.Of(), - Mock.Of(), - Mock.Of(), - Mock.Of(), - Mock.Of(), - Mock.Of(), - Mock.Of(), - Mock.Of(), - Mock.Of(), - Mock.Of(), - mediaSourceManager, - Mock.Of(), - transcodeSessionStore); -} diff --git a/tests/Jellyfin.Server.Implementations.Tests/SessionManager/SessionManagerTests.cs b/tests/Jellyfin.Server.Implementations.Tests/SessionManager/SessionManagerTests.cs index bb74ee9f9d..f803c69af2 100644 --- a/tests/Jellyfin.Server.Implementations.Tests/SessionManager/SessionManagerTests.cs +++ b/tests/Jellyfin.Server.Implementations.Tests/SessionManager/SessionManagerTests.cs @@ -11,7 +11,6 @@ using MediaBrowser.Controller.Drawing; using MediaBrowser.Controller.Dto; using MediaBrowser.Controller.Events; using MediaBrowser.Controller.Library; -using MediaBrowser.Controller.MediaEncoding; using MediaBrowser.Controller.Net; using MediaBrowser.Controller.Session; using MediaBrowser.Model.Session; @@ -42,8 +41,7 @@ public class SessionManagerTests Mock.Of(), Mock.Of(), Mock.Of(), - Mock.Of(), - new NullTranscodeSessionStore()); + Mock.Of()); await Assert.ThrowsAsync(exceptionType, () => sessionManager.GetAuthorizationToken( new User("test", "default", "default"), @@ -70,8 +68,7 @@ public class SessionManagerTests Mock.Of(), Mock.Of(), Mock.Of(), - Mock.Of(), - new NullTranscodeSessionStore()); + Mock.Of()); await Assert.ThrowsAsync(exceptionType, () => sessionManager.AuthenticateNewSessionInternal(authenticationRequest, false)); } @@ -241,8 +238,7 @@ public class SessionManagerTests Mock.Of(), Mock.Of(), Mock.Of(), - Mock.Of(), - new NullTranscodeSessionStore()); + Mock.Of()); } // All sessions are logged with the same client and device id on purpose, those values are taken