unkin/jellyfin-ha-src
4
0
Fork 0
Code Issues 10 Pull Requests Packages Projects Releases Wiki Activity

fix: make transcode leases ownership-checked and cleanup-aware

Browse Source 
Cleanup never matched a live session because the controller registered empty
manifest and segment paths, renewal was a read-modify-write that could revert a
takeover, and the takeover script compared an ISO date to a number, so it errored.

- populate the session record's manifest and segment paths from the playlist path
- renew the lease via a Lua compare-and-set on the owning pod
- store the lease expiry as unix milliseconds so the scripts can compare it
- retain the session record past its lease so an orphan can still be taken over
- test the Redis store against a real Redis, including the renew-vs-takeover race
- drop the live stream record nothing ever read back
This commit is contained in:
Unkin Agent
2026-09-12 10:19:29 +10:00
parent d825f8ac81
commit baa16b6586
22 changed files with 644 additions and 1221 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Directory.Packages.props
+1
View File
@@ -86,6 +86,7 @@
<PackageVersion Include="System.Text.Json" Version="10.0.11" />
<PackageVersion Include="TagLibSharp" Version="2.3.0" />
<PackageVersion Include="Testcontainers.PostgreSql" Version="4.15.0" />
<PackageVersion Include="Testcontainers.Redis" Version="4.15.0" />
<PackageVersion Include="z440.atl.core" Version="7.16.0" />
<PackageVersion Include="TMDbLib" Version="3.0.0" />
<PackageVersion Include="UTF.Unknown" Version="2.7.0" />
Emby.Server.Implementations/MediaEncoding/RedisTranscodeSessionStore.cs
+67 -102
View File
@@ -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:";
/// <summary>
/// Lua script for atomic takeover: reads the stored session, checks whether the lease has
/// expired (comparing <c>LeaseExpiresUtc.Ticks</c> against the caller-supplied current ticks),
/// and if expired, updates the owner and expiry before returning 1; returns 0 otherwise.
/// 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.
/// </summary>
private const string TakeoverScript = @"
local raw = redis.call('GET', KEYS[1])
if not raw then return 0 end
local session = cjson.decode(raw)
local currentTicks = tonumber(ARGV[1])
if session['LeaseExpiresUtc'] > currentTicks then return 0 end
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";
/// <summary>
/// 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.
/// </summary>
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);
/// <inheritdoc />
public async Task SetAsync(TranscodeSession session, CancellationToken cancellationToken = default)
{
var key = GetKey(session.PlaySessionId);
var json = JsonSerializer.Serialize(session);
var leaseDurationMs = (long)_options.LeaseDurationSeconds * 1000;
await _db.StringSetAsync(key, json, TimeSpan.FromMilliseconds(leaseDurationMs)).ConfigureAwait(false);
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<TranscodeSession>(raw.ToString());
var session = JsonSerializer.Deserialize<TranscodeSession>(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";
}
/// <inheritdoc />
public async Task RenewLeaseAsync(string playSessionId, CancellationToken cancellationToken = default)
public async Task<bool> 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<TranscodeSession>(raw.ToString());
if (session is null)
{
return;
}
var leaseDurationMs = (long)_options.LeaseDurationSeconds * 1000;
session.LeaseExpiresUtc = DateTime.UtcNow.AddMilliseconds(leaseDurationMs);
var json = JsonSerializer.Serialize(session);
await _db.StringSetAsync(key, json, TimeSpan.FromMilliseconds(leaseDurationMs)).ConfigureAwait(false);
_logger.LogDebug("Renewed lease for transcode session {PlaySessionId}.", playSessionId);
return true;
}
/// <inheritdoc />
@@ -126,14 +144,10 @@ return 1";
/// <inheritdoc />
public async Task<bool> TryTakeoverAsync(string playSessionId, string claimingPod, CancellationToken cancellationToken = default)
{
var key = GetKey(playSessionId);
var leaseDurationMs = (long)_options.LeaseDurationSeconds * 1000;
var currentTicks = DateTime.UtcNow.Ticks;
var result = (long?)await _db.ScriptEvaluateAsync(
TakeoverScript,
keys: new RedisKey[] { key },
values: new RedisValue[] { currentTicks, claimingPod, leaseDurationMs }).ConfigureAwait(false);
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;
/// <inheritdoc />
public async Task<IEnumerable<TranscodeSession>> GetActiveSessionsAsync(CancellationToken cancellationToken = default)
{
@@ -181,7 +190,7 @@ return 1";
TranscodeSession? session;
try
{
session = JsonSerializer.Deserialize<TranscodeSession>(raw.ToString());
session = JsonSerializer.Deserialize<TranscodeSession>(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;
}
/// <inheritdoc />
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<DateTime>
{
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);
}
/// <inheritdoc />
public async Task<LiveStreamSession?> 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<LiveStreamSession>(raw.ToString());
}
/// <inheritdoc />
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<LiveStreamSession>(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<RedisKey>
{
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);
}
}
Emby.Server.Implementations/Session/SessionManager.cs
+4 -37
View File
@@ -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<string, SessionInfo> _activeConnections
= new(StringComparer.OrdinalIgnoreCase);
@@ -91,7 +89,6 @@ namespace Emby.Server.Implementations.Session
/// <param name="deviceManager">Instance of <see cref="IDeviceManager"/> interface.</param>
/// <param name="mediaSourceManager">Instance of <see cref="IMediaSourceManager"/> interface.</param>
/// <param name="hostApplicationLifetime">Instance of <see cref="IHostApplicationLifetime"/> interface.</param>
/// <param name="transcodeSessionStore">Instance of <see cref="ITranscodeSessionStore"/> interface.</param>
public SessionManager(
ILogger<SessionManager> 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<string, string>());
@@ -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);
}
}
/// <summary>
@@ -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
Jellyfin.Api/Controllers/DynamicHlsController.cs
+75 -31
View File
@@ -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;
/// <summary>
/// Initializes a new instance of the <see cref="DynamicHlsController"/> class.
@@ -77,6 +81,7 @@ public class DynamicHlsController : BaseJellyfinApiController
/// <param name="encodingHelper">Instance of <see cref="EncodingHelper"/>.</param>
/// <param name="dynamicHlsPlaylistGenerator">Instance of <see cref="IDynamicHlsPlaylistGenerator"/>.</param>
/// <param name="transcodeSessionStore">Instance of the <see cref="ITranscodeSessionStore"/> interface used to register and renew HLS transcoding session leases in the durable store.</param>
/// <param name="transcodeStoreOptions">The <see cref="TranscodeStoreOptions"/> holding the session lease duration.</param>
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> 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
}
}
/// <summary>
/// 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.
/// </summary>
/// <param name="isHaMode">Whether the session is being resumed from the durable store.</param>
/// <param name="requestedSegmentLength">The segment length requested by the streaming pipeline.</param>
/// <param name="encodingOptions">The encoding options holding the recovery segment length.</param>
/// <returns>The segment length in seconds.</returns>
internal static int GetEffectiveSegmentLength(bool isHaMode, int requestedSegmentLength, EncodingOptions encodingOptions)
=> isHaMode
? Math.Clamp(encodingOptions.RecoverySegmentLengthSeconds, 1, 6)
: requestedSegmentLength;
/// <summary>
/// Gets the ffmpeg <c>hls_list_size</c>. HA mode keeps a bounded rolling buffer; 0 (unbounded)
/// otherwise.
/// </summary>
/// <param name="isHaMode">Whether the session is being resumed from the durable store.</param>
/// <param name="encodingOptions">The encoding options holding the recovery buffer count.</param>
/// <returns>The number of segments to keep in the playlist, or 0 for all of them.</returns>
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)
/// <summary>
/// Builds the durable record for an HLS output owned by this instance. The manifest and segment
/// paths have to be real, otherwise <c>DeleteTranscodeFileTask</c> cannot tell which files
/// belong to a live session.
/// </summary>
/// <param name="playSessionId">The play session identifier.</param>
/// <param name="mediaSourceId">The media source identifier.</param>
/// <param name="playlistPath">The absolute path of the HLS playlist this instance is writing.</param>
/// <param name="leaseDuration">The initial lease duration.</param>
/// <returns>The session record to store.</returns>
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('.');
MediaBrowser.Controller/MediaEncoding/ITranscodeSessionStore.cs
+8 -33
View File
@@ -47,11 +47,17 @@ public interface ITranscodeSessionStore
/// <summary>
/// Renews the lease for an existing session, extending its
/// <see cref="TranscodeSession.LeaseExpiresUtc"/> by the store's configured lease duration.
/// The renewal is rejected when <paramref name="ownerPod"/> no longer owns the lease, so a
/// renewal in flight while another pod wins <see cref="TryTakeoverAsync"/> cannot revert the takeover.
/// </summary>
/// <param name="playSessionId">The play session identifier.</param>
/// <param name="ownerPod">The name of the pod that believes it owns the lease.</param>
/// <param name="cancellationToken">A cancellation token.</param>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
Task RenewLeaseAsync(string playSessionId, CancellationToken cancellationToken = default);
/// <returns>
/// <c>true</c> if the lease was renewed; <c>false</c> if the session no longer exists or is
/// owned by another pod.
/// </returns>
Task<bool> RenewLeaseAsync(string playSessionId, string ownerPod, CancellationToken cancellationToken = default);
/// <summary>
/// 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.
/// </returns>
Task<IEnumerable<TranscodeSession>> GetActiveSessionsAsync(CancellationToken cancellationToken = default);
/// <summary>
/// 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.
/// </summary>
/// <param name="session">The live stream session to store.</param>
/// <param name="cancellationToken">A cancellation token.</param>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
Task SetLiveStreamAsync(LiveStreamSession session, CancellationToken cancellationToken = default);
/// <summary>
/// Attempts to retrieve a live stream session by its live stream identifier and the
/// session or play-session identifier that owns it.
/// </summary>
/// <param name="liveStreamId">The live stream identifier.</param>
/// <param name="sessionIdOrPlaySessionId">The session identifier or play-session identifier.</param>
/// <param name="cancellationToken">A cancellation token.</param>
/// <returns>
/// The <see cref="LiveStreamSession"/> if it exists; otherwise <c>null</c>.
/// </returns>
Task<LiveStreamSession?> TryGetLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default);
/// <summary>
/// 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.
/// </summary>
/// <param name="liveStreamId">The live stream identifier.</param>
/// <param name="sessionIdOrPlaySessionId">The session identifier or play-session identifier.</param>
/// <param name="cancellationToken">A cancellation token.</param>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
Task DeleteLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default);
}
MediaBrowser.Controller/MediaEncoding/LiveStreamSession.cs
-36
View File
@@ -1,36 +0,0 @@
using System;
namespace MediaBrowser.Controller.MediaEncoding;
/// <summary>
/// Represents a durable record of an open live stream session, enabling HA pod recovery
/// when the owning pod crashes or is evicted.
/// </summary>
public sealed class LiveStreamSession
{
/// <summary>
/// Gets or sets the live stream identifier (e.g. a TV tuner channel token).
/// </summary>
public string LiveStreamId { get; set; } = string.Empty;
/// <summary>
/// Gets or sets the session identifier of the client that opened this live stream.
/// </summary>
public string SessionId { get; set; } = string.Empty;
/// <summary>
/// Gets or sets the play session identifier associated with this live stream,
/// or an empty string when the client did not supply one.
/// </summary>
public string PlaySessionId { get; set; } = string.Empty;
/// <summary>
/// Gets or sets the name of the pod that currently holds this live stream open.
/// </summary>
public string OwnerPod { get; set; } = string.Empty;
/// <summary>
/// Gets or sets the UTC time at which this record was created.
/// </summary>
public DateTime OpenedAtUtc { get; set; }
}
MediaBrowser.Controller/MediaEncoding/NullTranscodeSessionStore.cs
+2 -14
View File
@@ -24,8 +24,8 @@ public sealed class NullTranscodeSessionStore : ITranscodeSessionStore
=> Task.CompletedTask;
/// <inheritdoc />
public Task RenewLeaseAsync(string playSessionId, CancellationToken cancellationToken = default)
=> Task.CompletedTask;
public Task<bool> RenewLeaseAsync(string playSessionId, string ownerPod, CancellationToken cancellationToken = default)
=> Task.FromResult(true);
/// <inheritdoc />
public Task DeleteAsync(string playSessionId, CancellationToken cancellationToken = default)
@@ -34,16 +34,4 @@ public sealed class NullTranscodeSessionStore : ITranscodeSessionStore
/// <inheritdoc />
public Task<IEnumerable<TranscodeSession>> GetActiveSessionsAsync(CancellationToken cancellationToken = default)
=> Task.FromResult<IEnumerable<TranscodeSession>>(Array.Empty<TranscodeSession>());
/// <inheritdoc />
public Task SetLiveStreamAsync(LiveStreamSession session, CancellationToken cancellationToken = default)
=> Task.CompletedTask;
/// <inheritdoc />
public Task<LiveStreamSession?> TryGetLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default)
=> Task.FromResult<LiveStreamSession?>(null);
/// <inheritdoc />
public Task DeleteLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default)
=> Task.CompletedTask;
}
MediaBrowser.Controller/MediaEncoding/TranscodeSession.cs
+37
View File
@@ -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.
/// </summary>
public long LastDurablePlaybackOffset { get; set; }
/// <summary>
/// Creates a session record for an HLS output, deriving <see cref="ManifestPath"/> and
/// <see cref="SegmentPathPrefix"/> from the playlist path so that cleanup can recognise
/// every file the session owns.
/// </summary>
/// <param name="playSessionId">The play session identifier.</param>
/// <param name="mediaSourceId">The media source identifier.</param>
/// <param name="ownerPod">The name of the pod that owns the session.</param>
/// <param name="playlistPath">The absolute path of the HLS playlist (.m3u8) file.</param>
/// <param name="leaseDuration">The initial lease duration.</param>
/// <returns>The new <see cref="TranscodeSession"/>.</returns>
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,
};
/// <summary>
/// Gets the prefix every segment file of the HLS output at <paramref name="playlistPath"/> starts with.
/// Segments are written as <c>&lt;playlist path without extension&gt;&lt;index&gt;&lt;segment extension&gt;</c>.
/// </summary>
/// <param name="playlistPath">The absolute path of the HLS playlist (.m3u8) file.</param>
/// <returns>The segment path prefix.</returns>
public static string GetSegmentPathPrefix(string playlistPath)
=> Path.ChangeExtension(playlistPath, null) ?? playlistPath;
}
MediaBrowser.Controller/MediaEncoding/TranscodeStoreOptions.cs
+7
View File
@@ -16,4 +16,11 @@ public sealed class TranscodeStoreOptions
/// Gets or sets the duration in seconds for which a transcoding session lease is valid.
/// </summary>
public int LeaseDurationSeconds { get; set; } = 30;
/// <summary>
/// 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.
/// </summary>
public int SessionRetentionSeconds { get; set; } = 300;
}
README.md
+8 -7
View File
@@ -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
```
---
docs/FORK-DIFF.md
+11 -9
View File
@@ -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` |
tests/Jellyfin.Api.Tests/Controllers/DynamicHlsHaModeTests.cs
+105
View File
@@ -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
{
/// <summary>
/// Tests for the HA behaviour of <see cref="DynamicHlsController"/>: the session record it
/// registers in <see cref="ITranscodeSessionStore"/> and the ffmpeg segmenting options it
/// picks once a session is resumed on another pod.
/// </summary>
public class DynamicHlsHaModeTests
{
/// <summary>
/// The registered session has to name the files it owns, otherwise
/// <c>DeleteTranscodeFileTask</c> cannot recognise and protect them.
/// </summary>
[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);
}
/// <summary>
/// Every segment ffmpeg writes for the playlist has to start with the recorded prefix.
/// Segment paths are built as <c>&lt;playlist without extension&gt;&lt;index&gt;&lt;extension&gt;</c>.
/// </summary>
[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);
}
}
/// <summary>
/// HA mode shortens segments and bounds the rolling buffer; the configured values are
/// user-editable so they are clamped.
/// </summary>
[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));
}
/// <summary>
/// Outside HA mode the requested segment length is used unchanged and the playlist is unbounded.
/// </summary>
[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));
}
/// <summary>
/// HA mode keeps a bounded rolling buffer of segments for a takeover pod to serve.
/// </summary>
[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));
}
}
}
tests/Jellyfin.Api.Tests/Controllers/DynamicHlsHaTakeoverTests.cs
-259
View File
@@ -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
{
/// <summary>
/// Tests for HA recovery scenarios that will be wired into <see cref="DynamicHlsController"/>
/// in Phase 5.2. These tests verify the <see cref="ITranscodeSessionStore"/> contract that
/// the controller will rely on for missing-local-job recovery, claim racing, and cleanup guarding.
/// </summary>
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,
};
/// <summary>
/// Missing-local-job + durable-manifest-present: the store returns the session so
/// the controller can serve the existing manifest instead of returning an error.
/// </summary>
[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);
}
/// <summary>
/// Claim-race between two concurrent requesters: only one wins
/// <see cref="ITranscodeSessionStore.TryTakeoverAsync"/>.
/// The other receives <c>false</c>, indicating it should redirect (302) or wait.
/// </summary>
[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);
}
/// <summary>
/// Stale-manifest cleanup guard: a lease that has expired beyond the recovery window
/// causes the store to return <c>null</c>, signalling that cleanup may proceed safely.
/// </summary>
[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);
}
/// <summary>
/// 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.
/// </summary>
[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);
}
/// <summary>
/// Segment-length selection: when no entry exists in the store for the play-session
/// (HA mode inactive), the normal segment length should be used.
/// </summary>
[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);
}
/// <summary>
/// Minimal in-memory <see cref="ITranscodeSessionStore"/> used within this test class
/// to avoid a cross-project reference to Jellyfin.MediaEncoding.Tests.
/// </summary>
private sealed class HaTestSessionStore : ITranscodeSessionStore
{
private static readonly TimeSpan LeaseDuration = TimeSpan.FromSeconds(30);
private readonly Dictionary<string, TranscodeSession> _sessions =
new(StringComparer.OrdinalIgnoreCase);
private readonly Lock _lock = new();
public Task<TranscodeSession?> TryGetAsync(string playSessionId, CancellationToken cancellationToken = default)
{
lock (_lock)
{
if (_sessions.TryGetValue(playSessionId, out var s) && s.LeaseExpiresUtc > DateTime.UtcNow)
{
return Task.FromResult<TranscodeSession?>(Clone(s));
}
return Task.FromResult<TranscodeSession?>(null);
}
}
public Task<bool> 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<IEnumerable<TranscodeSession>> GetActiveSessionsAsync(CancellationToken cancellationToken = default)
{
lock (_lock)
{
var sessions = _sessions.Values
.Where(s => s.LeaseExpiresUtc > DateTime.UtcNow)
.Select(Clone)
.ToList();
return Task.FromResult<IEnumerable<TranscodeSession>>(sessions);
}
}
public Task SetLiveStreamAsync(LiveStreamSession session, CancellationToken cancellationToken = default)
=> Task.CompletedTask;
public Task<LiveStreamSession?> TryGetLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default)
=> Task.FromResult<LiveStreamSession?>(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,
};
}
}
}
tests/Jellyfin.Api.Tests/Controllers/DynamicHlsSessionRegistrationTests.cs
-223
View File
@@ -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
{
/// <summary>
/// Tests for HLS session registration, lease renewal, and cleanup behaviour wired into
/// <see cref="Jellyfin.Api.Controllers.DynamicHlsController"/> in Phase 5.2.2a.
/// These tests verify the <see cref="ITranscodeSessionStore"/> contract used by the controller.
/// </summary>
public class DynamicHlsSessionRegistrationTests
{
private static TranscodeSession CreateSession(string id, string pod)
=> new TranscodeSession
{
PlaySessionId = id,
OwnerPod = pod,
LeaseExpiresUtc = DateTime.UtcNow.AddSeconds(30),
ManifestPath = string.Empty,
SegmentPathPrefix = string.Empty,
MediaSourceId = "media-source-1",
LastCompletedSegmentIndex = 0,
LastDurablePlaybackOffset = 0L,
};
/// <summary>
/// After registering a session via <see cref="ITranscodeSessionStore.SetAsync"/>,
/// <see cref="ITranscodeSessionStore.TryGetAsync"/> must return a non-null result with
/// matching <see cref="TranscodeSession.PlaySessionId"/> and <see cref="TranscodeSession.OwnerPod"/>.
/// </summary>
[Fact]
[Trait("Category", "UnitTest")]
public async Task SessionRegistration_AfterStreamStart_StoreContainsSession()
{
var store = new InMemoryTranscodeSessionStore();
var session = CreateSession("session-reg-1", "pod-a");
await store.SetAsync(session, 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);
}
/// <summary>
/// After calling <see cref="ITranscodeSessionStore.DeleteAsync"/>,
/// <see cref="ITranscodeSessionStore.TryGetAsync"/> must return <c>null</c>.
/// </summary>
[Fact]
[Trait("Category", "UnitTest")]
public async Task SessionCleanup_AfterStreamEnd_StoreReturnsNull()
{
var store = new InMemoryTranscodeSessionStore();
var session = CreateSession("session-cleanup-1", "pod-b");
await store.SetAsync(session, 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);
}
/// <summary>
/// After a session's initial lease window would have expired, calling
/// <see cref="ITranscodeSessionStore.RenewLeaseAsync"/> must extend the lease so that
/// <see cref="ITranscodeSessionStore.TryGetAsync"/> still returns the session as active.
/// </summary>
[Fact]
[Trait("Category", "UnitTest")]
public async Task LeaseRenewal_ExtendsBeyondInitialExpiry()
{
var store = new InMemoryTranscodeSessionStore();
// Create the session with a lease that has already expired.
var session = new TranscodeSession
{
PlaySessionId = "session-renewal-1",
OwnerPod = "pod-c",
LeaseExpiresUtc = DateTime.UtcNow.AddMilliseconds(-1),
ManifestPath = string.Empty,
SegmentPathPrefix = string.Empty,
MediaSourceId = "media-source-1",
LastCompletedSegmentIndex = 0,
LastDurablePlaybackOffset = 0L,
};
await store.SetAsync(session, 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);
}
/// <summary>
/// Minimal thread-safe in-memory implementation of <see cref="ITranscodeSessionStore"/>
/// used within this test class to avoid a cross-project reference.
/// </summary>
private sealed class InMemoryTranscodeSessionStore : ITranscodeSessionStore
{
private static readonly TimeSpan DefaultLeaseDuration = TimeSpan.FromSeconds(30);
private readonly Dictionary<string, TranscodeSession> _sessions =
new(StringComparer.OrdinalIgnoreCase);
private readonly Lock _lock = new();
public Task<TranscodeSession?> TryGetAsync(string playSessionId, CancellationToken cancellationToken = default)
{
lock (_lock)
{
if (_sessions.TryGetValue(playSessionId, out var session) && session.LeaseExpiresUtc > DateTime.UtcNow)
{
return Task.FromResult<TranscodeSession?>(Clone(session));
}
return Task.FromResult<TranscodeSession?>(null);
}
}
public Task<bool> TryTakeoverAsync(string playSessionId, string claimingPod, CancellationToken cancellationToken = default)
{
lock (_lock)
{
if (!_sessions.TryGetValue(playSessionId, out var session))
{
return Task.FromResult(false);
}
if (session.LeaseExpiresUtc > DateTime.UtcNow)
{
return Task.FromResult(false);
}
session.OwnerPod = claimingPod;
session.LeaseExpiresUtc = DateTime.UtcNow.Add(DefaultLeaseDuration);
return Task.FromResult(true);
}
}
public Task SetAsync(TranscodeSession session, CancellationToken cancellationToken = default)
{
lock (_lock)
{
_sessions[session.PlaySessionId] = session;
}
return Task.CompletedTask;
}
public Task RenewLeaseAsync(string playSessionId, CancellationToken cancellationToken = default)
{
lock (_lock)
{
if (_sessions.TryGetValue(playSessionId, out var session))
{
session.LeaseExpiresUtc = DateTime.UtcNow.Add(DefaultLeaseDuration);
}
}
return Task.CompletedTask;
}
public Task DeleteAsync(string playSessionId, CancellationToken cancellationToken = default)
{
lock (_lock)
{
_sessions.Remove(playSessionId);
}
return Task.CompletedTask;
}
public Task<IEnumerable<TranscodeSession>> GetActiveSessionsAsync(CancellationToken cancellationToken = default)
{
lock (_lock)
{
var sessions = _sessions.Values
.Where(s => s.LeaseExpiresUtc > DateTime.UtcNow)
.Select(Clone)
.ToList();
return Task.FromResult<IEnumerable<TranscodeSession>>(sessions);
}
}
public Task SetLiveStreamAsync(LiveStreamSession session, CancellationToken cancellationToken = default)
=> Task.CompletedTask;
public Task<LiveStreamSession?> TryGetLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default)
=> Task.FromResult<LiveStreamSession?>(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,
};
}
}
}
tests/Jellyfin.MediaEncoding.Tests/Fakes/InMemoryTranscodeSessionStore.cs
+7 -56
View File
@@ -18,7 +18,6 @@ public sealed class InMemoryTranscodeSessionStore : ITranscodeSessionStore
public static readonly TimeSpan DefaultLeaseDuration = TimeSpan.FromSeconds(30);
private readonly Dictionary<string, TranscodeSession> _sessions = new(StringComparer.OrdinalIgnoreCase);
private readonly Dictionary<string, LiveStreamSession> _liveStreams = new(StringComparer.OrdinalIgnoreCase);
private readonly Lock _lock = new();
/// <inheritdoc />
@@ -70,17 +69,19 @@ public sealed class InMemoryTranscodeSessionStore : ITranscodeSessionStore
}
/// <inheritdoc />
public Task RenewLeaseAsync(string playSessionId, CancellationToken cancellationToken = default)
public Task<bool> 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);
}
}
/// <inheritdoc />
@@ -104,56 +105,6 @@ public sealed class InMemoryTranscodeSessionStore : ITranscodeSessionStore
}
}
/// <inheritdoc />
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;
}
/// <inheritdoc />
public Task<LiveStreamSession?> TryGetLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default)
{
lock (_lock)
{
_liveStreams.TryGetValue(MakeLiveStreamKey(liveStreamId, sessionIdOrPlaySessionId), out var session);
return Task.FromResult(session);
}
}
/// <inheritdoc />
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
{
tests/Jellyfin.MediaEncoding.Tests/Transcoding/TranscodeManagerTests.cs → tests/Jellyfin.MediaEncoding.Tests/Transcoding/InMemoryTranscodeSessionStoreTests.cs
+25 -6
View File
@@ -7,12 +7,12 @@ using Xunit;
namespace Jellyfin.MediaEncoding.Tests.Transcoding;
/// <summary>
/// Unit tests for the HA session-store contract: lease expiry, double-claim prevention,
/// heartbeat renewal, and stale-session cleanup.
/// All tests exercise <see cref="InMemoryTranscodeSessionStore"/> which implements the
/// <see cref="ITranscodeSessionStore"/> interface that will be backed by Redis in Phase 5.2.
/// Unit tests for the <see cref="ITranscodeSessionStore"/> contract — lease expiry, double-claim
/// prevention, ownership-checked renewal and stale-session cleanup — against
/// <see cref="InMemoryTranscodeSessionStore"/>, the reference implementation. The Redis-backed
/// implementation is covered by <c>RedisTranscodeSessionStoreTests</c> against a real Redis.
/// </summary>
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.");
}
/// <summary>
/// A renewal from a pod that no longer owns the lease must fail and must not revert ownership.
/// </summary>
[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);
}
/// <summary>
/// Stale-session cleanup: an expired session can be deleted without error, and a
/// subsequent <see cref="ITranscodeSessionStore.TryGetAsync"/> returns <c>null</c>.
tests/Jellyfin.Server.Implementations.Tests/Jellyfin.Server.Implementations.Tests.csproj
+2
View File
@@ -18,6 +18,8 @@
<PackageReference Include="AutoFixture.AutoMoq" />
<PackageReference Include="Microsoft.NET.Test.Sdk" />
<PackageReference Include="Moq" />
<PackageReference Include="StackExchange.Redis" />
<PackageReference Include="Testcontainers.Redis" />
<PackageReference Include="xunit.v3" />
<PackageReference Include="xunit.runner.visualstudio">
<PrivateAssets>all</PrivateAssets>
tests/Jellyfin.Server.Implementations.Tests/MediaEncoding/RedisTranscodeSessionStoreTests.cs
+212 -311
View File
@@ -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;
/// <summary>
/// Tests for transcode session store contract behavior, using <see cref="InMemoryTranscodeSessionStore"/>
/// as a reference implementation (no real Redis required).
/// Integration tests for <see cref="RedisTranscodeSessionStore"/> and its Lua scripts against a
/// real Redis container.
/// </summary>
public class RedisTranscodeSessionStoreTests
[Trait("Category", "RequiresDocker")]
public sealed class RedisTranscodeSessionStoreTests : IAsyncLifetime
{
private readonly RedisContainer _container;
private IConnectionMultiplexer? _redis;
/// <summary>
/// Verifies that <see cref="ITranscodeSessionStore.TryGetAsync"/> returns <c>null</c> after
/// a session's lease has expired.
/// Initializes a new instance of the <see cref="RedisTranscodeSessionStoreTests"/> class.
/// </summary>
[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();
}
/// <summary>
/// Verifies that <see cref="ITranscodeSessionStore.TryTakeoverAsync"/> returns <c>false</c>
/// when the session's lease is still valid.
/// Starts the Redis container before any tests in the class run.
/// </summary>
/// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
public async ValueTask InitializeAsync()
{
await _container.StartAsync().ConfigureAwait(false);
_redis = await ConnectionMultiplexer.ConnectAsync(_container.GetConnectionString()).ConfigureAwait(false);
}
/// <summary>
/// Stops and removes the Redis container after all tests in the class have run.
/// </summary>
/// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
public async ValueTask DisposeAsync()
{
if (_redis is not null)
{
await _redis.DisposeAsync().ConfigureAwait(false);
}
await _container.DisposeAsync().ConfigureAwait(false);
}
/// <summary>
/// A stored session round-trips through Redis with the paths cleanup relies on intact.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[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);
}
/// <summary>
/// The owning pod can extend its own lease.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[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);
}
/// <summary>
/// A pod that does not own the lease cannot renew it, and its attempt leaves the owner alone.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[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);
}
/// <summary>
/// Renewal of a session that is gone fails instead of recreating it.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[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));
}
/// <summary>
/// A valid lease blocks takeover.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[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);
}
/// <summary>
/// Verifies that <see cref="ITranscodeSessionStore.TryTakeoverAsync"/> returns <c>true</c>
/// 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.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[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);
}
/// <summary>
/// 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.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[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<bool>[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));
}
/// <summary>
/// Verifies that <see cref="ITranscodeSessionStore.SetLiveStreamAsync"/> 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.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[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);
}
/// <summary>
/// Verifies that <see cref="ITranscodeSessionStore.SetLiveStreamAsync"/> 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.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[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);
}
}
/// <summary>
/// Verifies that <see cref="ITranscodeSessionStore.DeleteLiveStreamAsync"/> 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.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[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));
}
/// <summary>
/// Verifies that <see cref="ITranscodeSessionStore.TryGetLiveStreamAsync"/> returns null when
/// no matching record exists.
/// </summary>
[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);
}
/// <summary>
/// Thread-safe, in-memory implementation of <see cref="ITranscodeSessionStore"/> used within
/// this test class to avoid a cross-project reference to Jellyfin.MediaEncoding.Tests.
/// </summary>
private sealed class InMemoryTranscodeSessionStore : ITranscodeSessionStore
{
private static readonly TimeSpan DefaultLeaseDuration = TimeSpan.FromSeconds(30);
private readonly Dictionary<string, TranscodeSession> _sessions = new(StringComparer.OrdinalIgnoreCase);
private readonly Dictionary<string, LiveStreamSession> _liveStreams = new(StringComparer.OrdinalIgnoreCase);
private readonly Lock _lock = new();
/// <inheritdoc />
public Task<TranscodeSession?> 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<TranscodeSession?>(Clone(session));
}
return Task.FromResult<TranscodeSession?>(null);
}
}
/// <inheritdoc />
public Task<bool> TryTakeoverAsync(string playSessionId, string claimingPod, CancellationToken cancellationToken = default)
{
lock (_lock)
{
if (!_sessions.TryGetValue(playSessionId, out var session))
{
return Task.FromResult(false);
}
if (session.LeaseExpiresUtc > DateTime.UtcNow)
{
return Task.FromResult(false);
}
session.OwnerPod = claimingPod;
session.LeaseExpiresUtc = DateTime.UtcNow.Add(DefaultLeaseDuration);
return Task.FromResult(true);
}
}
/// <inheritdoc />
public Task SetAsync(TranscodeSession session, CancellationToken cancellationToken = default)
{
lock (_lock)
{
_sessions[session.PlaySessionId] = session;
}
return Task.CompletedTask;
}
/// <inheritdoc />
public Task RenewLeaseAsync(string playSessionId, CancellationToken cancellationToken = default)
{
lock (_lock)
{
if (_sessions.TryGetValue(playSessionId, out var session))
{
session.LeaseExpiresUtc = DateTime.UtcNow.Add(DefaultLeaseDuration);
}
}
return Task.CompletedTask;
}
/// <inheritdoc />
public Task DeleteAsync(string playSessionId, CancellationToken cancellationToken = default)
{
lock (_lock)
{
_sessions.Remove(playSessionId);
}
return Task.CompletedTask;
}
/// <inheritdoc />
public Task<IEnumerable<TranscodeSession>> GetActiveSessionsAsync(CancellationToken cancellationToken = default)
{
lock (_lock)
{
var sessions = _sessions.Values
.Where(s => s.LeaseExpiresUtc > DateTime.UtcNow)
.Select(Clone)
.ToList();
return Task.FromResult<IEnumerable<TranscodeSession>>(sessions);
}
}
/// <inheritdoc />
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;
}
/// <inheritdoc />
public Task<LiveStreamSession?> TryGetLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default)
{
lock (_lock)
{
_liveStreams.TryGetValue(MakeLiveStreamKey(liveStreamId, sessionIdOrPlaySessionId), out var session);
return Task.FromResult(session);
}
}
/// <inheritdoc />
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<RedisTranscodeSessionStore>.Instance);
}
tests/Jellyfin.Server.Implementations.Tests/ScheduledTasks/DeleteTranscodeFileTaskTests.cs
+69 -21
View File
@@ -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;
/// <summary>
/// Tests for lease-aware cleanup behavior expected of <c>DeleteTranscodeFileTask</c> once
/// it is made HA-aware in Phase 5.2.
/// <para>
/// The current <c>DeleteTranscodeFileTask</c> implementation uses file-age only and does not
/// check <see cref="ITranscodeSessionStore"/>, which creates a data-loss risk on shared NFS
/// storage. These tests document the correct contract by exercising the store directly.
/// </para>
/// Tests for the lease-aware cleanup behaviour of <c>DeleteTranscodeFileTask</c>: 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.
/// </summary>
public class DeleteTranscodeFileTaskTests
{
@@ -313,6 +310,64 @@ public class DeleteTranscodeFileTaskTests
Assert.Empty(deletedFiles);
}
/// <summary>
/// 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.
/// </summary>
[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<string>();
var fileSystemMock = new Mock<IFileSystem>();
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<FileSystemMetadata>()))
.Returns(DateTime.UtcNow.AddDays(-2));
fileSystemMock
.Setup(fs => fs.DeleteFile(It.IsAny<string>()))
.Callback<string>(deletedFiles.Add);
fileSystemMock
.Setup(fs => fs.GetDirectories(It.IsAny<string>(), It.IsAny<bool>()))
.Returns(Enumerable.Empty<FileSystemMetadata>());
var localizationMock = new Mock<MediaBrowser.Model.Globalization.ILocalizationManager>();
localizationMock
.Setup(l => l.GetLocalizedString(It.IsAny<string>()))
.Returns<string>(key => key);
var task = new Emby.Server.Implementations.ScheduledTasks.Tasks.DeleteTranscodeFileTask(
new Mock<Microsoft.Extensions.Logging.ILogger<Emby.Server.Implementations.ScheduledTasks.Tasks.DeleteTranscodeFileTask>>().Object,
fileSystemMock.Object,
CreateConfigMock(TranscodePath).Object,
localizationMock.Object,
store);
await task.ExecuteAsync(new Progress<double>(), CancellationToken.None);
Assert.Equal(new[] { orphanedFile }, deletedFiles);
}
/// <summary>
/// Minimal in-memory <see cref="ITranscodeSessionStore"/> 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<bool> 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<LiveStreamSession?> TryGetLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default)
=> Task.FromResult<LiveStreamSession?>(null);
public Task DeleteLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default)
=> Task.CompletedTask;
private static TranscodeSession Clone(TranscodeSession source)
=> new TranscodeSession
{
tests/Jellyfin.Server.Implementations.Tests/SessionManager/IdlePlaybackTests.cs
+1 -3
View File
@@ -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<IServerApplicationHost>(),
Mock.Of<IDeviceManager>(),
Mock.Of<IMediaSourceManager>(),
Mock.Of<IHostApplicationLifetime>(),
new NullTranscodeSessionStore());
Mock.Of<IHostApplicationLifetime>());
var session = await sessionManager.LogSessionActivity(
"Test Client",
"1.0.0",
tests/Jellyfin.Server.Implementations.Tests/SessionManager/LiveStreamHaRecordTests.cs
-66
View File
@@ -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<ITranscodeSessionStore>();
var mediaSourceManager = new Mock<IMediaSourceManager>();
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<CancellationToken>()), Times.Once);
mediaSourceManager.Verify(m => m.CloseLiveStream("stream-1"), Times.Once);
}
[Fact]
public async Task CloseLiveStreamIfNeededAsync_Should_CloseStream_WhenDurableStoreFails()
{
var store = new Mock<ITranscodeSessionStore>();
store.Setup(s => s.DeleteLiveStreamAsync(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<CancellationToken>()))
.ThrowsAsync(new InvalidOperationException("redis unreachable"));
var mediaSourceManager = new Mock<IMediaSourceManager>();
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<Emby.Server.Implementations.Session.SessionManager>.Instance,
Mock.Of<IEventManager>(),
Mock.Of<IUserDataManager>(),
Mock.Of<IServerConfigurationManager>(),
Mock.Of<ILibraryManager>(),
Mock.Of<IUserManager>(),
Mock.Of<IMusicManager>(),
Mock.Of<IDtoService>(),
Mock.Of<IImageProcessor>(),
Mock.Of<IServerApplicationHost>(),
Mock.Of<IDeviceManager>(),
mediaSourceManager,
Mock.Of<IHostApplicationLifetime>(),
transcodeSessionStore);
}
tests/Jellyfin.Server.Implementations.Tests/SessionManager/SessionManagerTests.cs
+3 -7
View File
@@ -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<IServerApplicationHost>(),
Mock.Of<IDeviceManager>(),
Mock.Of<IMediaSourceManager>(),
Mock.Of<IHostApplicationLifetime>(),
new NullTranscodeSessionStore());
Mock.Of<IHostApplicationLifetime>());
await Assert.ThrowsAsync(exceptionType, () => sessionManager.GetAuthorizationToken(
new User("test", "default", "default"),
@@ -70,8 +68,7 @@ public class SessionManagerTests
Mock.Of<IServerApplicationHost>(),
Mock.Of<IDeviceManager>(),
Mock.Of<IMediaSourceManager>(),
Mock.Of<IHostApplicationLifetime>(),
new NullTranscodeSessionStore());
Mock.Of<IHostApplicationLifetime>());
await Assert.ThrowsAsync(exceptionType, () => sessionManager.AuthenticateNewSessionInternal(authenticationRequest, false));
}
@@ -241,8 +238,7 @@ public class SessionManagerTests
Mock.Of<IServerApplicationHost>(),
Mock.Of<IDeviceManager>(),
Mock.Of<IMediaSourceManager>(),
Mock.Of<IHostApplicationLifetime>(),
new NullTranscodeSessionStore());
Mock.Of<IHostApplicationLifetime>());
}
// All sessions are logged with the same client and device id on purpose, those values are taken