feat(hls): register transcode sessions and renew their lease
A transcode that is not in the store is invisible to the other replicas, so the HLS entry points have to publish and hold the lease themselves. - register the play session in the store when ffmpeg starts - renew the lease on a background loop and delete the session when it ends - shorten segments and bound the playlist window when resuming a stored session - add RecoverySegmentLengthSeconds and RecoverySegmentBufferCount encoding options
This commit is contained in:
@@ -0,0 +1,259 @@
|
||||
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,
|
||||
};
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,223 @@
|
||||
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,
|
||||
};
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user