From e3571eb2a43a9d43b2f2cab3333d78c7a526c748 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Fri, 11 Sep 2026 23:56:35 +1000 Subject: [PATCH] feat(session): record open live streams in the session store Live stream ownership is only tracked in process memory, so no replica can tell which pod holds a stream open once that pod is gone. - persist a LiveStreamSession record when the live stream mappings change - delete the record when the stream is closed - keep closing the stream when the store is unreachable --- .../Session/SessionManager.cs | 41 ++++++++++-- .../SessionManager/IdlePlaybackTests.cs | 4 +- .../SessionManager/LiveStreamHaRecordTests.cs | 66 +++++++++++++++++++ .../SessionManager/SessionManagerTests.cs | 10 ++- 4 files changed, 113 insertions(+), 8 deletions(-) create mode 100644 tests/Jellyfin.Server.Implementations.Tests/SessionManager/LiveStreamHaRecordTests.cs diff --git a/Emby.Server.Implementations/Session/SessionManager.cs b/Emby.Server.Implementations/Session/SessionManager.cs index 94215bed79..0a81e61870 100644 --- a/Emby.Server.Implementations/Session/SessionManager.cs +++ b/Emby.Server.Implementations/Session/SessionManager.cs @@ -28,6 +28,7 @@ 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; @@ -60,6 +61,7 @@ namespace Emby.Server.Implementations.Session private readonly IMediaSourceManager _mediaSourceManager; private readonly IServerApplicationHost _appHost; private readonly IDeviceManager _deviceManager; + private readonly ITranscodeSessionStore _transcodeSessionStore; private readonly CancellationTokenRegistration _shutdownCallback; private readonly ConcurrentDictionary _activeConnections = new(StringComparer.OrdinalIgnoreCase); @@ -89,6 +91,7 @@ namespace Emby.Server.Implementations.Session /// Instance of interface. /// Instance of interface. /// Instance of interface. + /// Instance of interface. public SessionManager( ILogger logger, IEventManager eventManager, @@ -102,7 +105,8 @@ namespace Emby.Server.Implementations.Session IServerApplicationHost appHost, IDeviceManager deviceManager, IMediaSourceManager mediaSourceManager, - IHostApplicationLifetime hostApplicationLifetime) + IHostApplicationLifetime hostApplicationLifetime, + ITranscodeSessionStore transcodeSessionStore) { _logger = logger; _eventManager = eventManager; @@ -116,6 +120,7 @@ namespace Emby.Server.Implementations.Session _appHost = appHost; _deviceManager = deviceManager; _mediaSourceManager = mediaSourceManager; + _transcodeSessionStore = transcodeSessionStore; _shutdownCallback = hostApplicationLifetime.ApplicationStopping.Register(OnApplicationStopping); _deviceManager.DeviceOptionsUpdated += OnDeviceManagerDeviceOptionsUpdated; @@ -350,6 +355,15 @@ 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); @@ -796,7 +810,7 @@ namespace Emby.Server.Implementations.Session if (!string.IsNullOrEmpty(info.LiveStreamId)) { - UpdateLiveStreamActiveSessionMappings(info.LiveStreamId, info.SessionId, info.PlaySessionId); + await UpdateLiveStreamActiveSessionMappings(info.LiveStreamId, info.SessionId, info.PlaySessionId).ConfigureAwait(false); } var eventArgs = new PlaybackStartEventArgs @@ -862,7 +876,7 @@ namespace Emby.Server.Implementations.Session return OnPlaybackProgress(info, false); } - private void UpdateLiveStreamActiveSessionMappings(string liveStreamId, string sessionId, string playSessionId) + private async Task UpdateLiveStreamActiveSessionMappings(string liveStreamId, string sessionId, string playSessionId) { var activeSessionMappings = _activeLiveStreamSessions.GetOrAdd(liveStreamId, _ => new ConcurrentDictionary()); @@ -886,6 +900,25 @@ 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); + } } /// @@ -931,7 +964,7 @@ namespace Emby.Server.Implementations.Session if (!string.IsNullOrEmpty(info.LiveStreamId)) { - UpdateLiveStreamActiveSessionMappings(info.LiveStreamId, info.SessionId, info.PlaySessionId); + await UpdateLiveStreamActiveSessionMappings(info.LiveStreamId, info.SessionId, info.PlaySessionId).ConfigureAwait(false); } var eventArgs = new PlaybackProgressEventArgs diff --git a/tests/Jellyfin.Server.Implementations.Tests/SessionManager/IdlePlaybackTests.cs b/tests/Jellyfin.Server.Implementations.Tests/SessionManager/IdlePlaybackTests.cs index 7722707cbe..6466c02dc0 100644 --- a/tests/Jellyfin.Server.Implementations.Tests/SessionManager/IdlePlaybackTests.cs +++ b/tests/Jellyfin.Server.Implementations.Tests/SessionManager/IdlePlaybackTests.cs @@ -8,6 +8,7 @@ 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; @@ -44,7 +45,8 @@ public class IdlePlaybackTests Mock.Of(), Mock.Of(), Mock.Of(), - Mock.Of()); + Mock.Of(), + new NullTranscodeSessionStore()); var session = await sessionManager.LogSessionActivity( "Test Client", "1.0.0", diff --git a/tests/Jellyfin.Server.Implementations.Tests/SessionManager/LiveStreamHaRecordTests.cs b/tests/Jellyfin.Server.Implementations.Tests/SessionManager/LiveStreamHaRecordTests.cs new file mode 100644 index 0000000000..4ea343afdc --- /dev/null +++ b/tests/Jellyfin.Server.Implementations.Tests/SessionManager/LiveStreamHaRecordTests.cs @@ -0,0 +1,66 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using MediaBrowser.Controller; +using MediaBrowser.Controller.Configuration; +using MediaBrowser.Controller.Devices; +using MediaBrowser.Controller.Drawing; +using MediaBrowser.Controller.Dto; +using MediaBrowser.Controller.Events; +using MediaBrowser.Controller.Library; +using MediaBrowser.Controller.MediaEncoding; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging.Abstractions; +using Moq; +using Xunit; + +namespace Jellyfin.Server.Implementations.Tests.SessionManager; + +public class LiveStreamHaRecordTests +{ + [Fact] + public async Task CloseLiveStreamIfNeededAsync_Should_DeleteDurableRecord() + { + var store = new Mock(); + var mediaSourceManager = new Mock(); + await using var sessionManager = CreateSessionManager(store.Object, mediaSourceManager.Object); + + await sessionManager.CloseLiveStreamIfNeededAsync("stream-1", "session-1"); + + store.Verify(s => s.DeleteLiveStreamAsync("stream-1", "session-1", It.IsAny()), Times.Once); + mediaSourceManager.Verify(m => m.CloseLiveStream("stream-1"), Times.Once); + } + + [Fact] + public async Task CloseLiveStreamIfNeededAsync_Should_CloseStream_WhenDurableStoreFails() + { + var store = new Mock(); + store.Setup(s => s.DeleteLiveStreamAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .ThrowsAsync(new InvalidOperationException("redis unreachable")); + var mediaSourceManager = new Mock(); + await using var sessionManager = CreateSessionManager(store.Object, mediaSourceManager.Object); + + await sessionManager.CloseLiveStreamIfNeededAsync("stream-2", "session-2"); + + mediaSourceManager.Verify(m => m.CloseLiveStream("stream-2"), Times.Once); + } + + private static Emby.Server.Implementations.Session.SessionManager CreateSessionManager( + ITranscodeSessionStore transcodeSessionStore, + IMediaSourceManager mediaSourceManager) + => new Emby.Server.Implementations.Session.SessionManager( + NullLogger.Instance, + Mock.Of(), + Mock.Of(), + Mock.Of(), + Mock.Of(), + Mock.Of(), + Mock.Of(), + Mock.Of(), + Mock.Of(), + Mock.Of(), + Mock.Of(), + mediaSourceManager, + Mock.Of(), + transcodeSessionStore); +} diff --git a/tests/Jellyfin.Server.Implementations.Tests/SessionManager/SessionManagerTests.cs b/tests/Jellyfin.Server.Implementations.Tests/SessionManager/SessionManagerTests.cs index f803c69af2..bb74ee9f9d 100644 --- a/tests/Jellyfin.Server.Implementations.Tests/SessionManager/SessionManagerTests.cs +++ b/tests/Jellyfin.Server.Implementations.Tests/SessionManager/SessionManagerTests.cs @@ -11,6 +11,7 @@ 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; @@ -41,7 +42,8 @@ public class SessionManagerTests Mock.Of(), Mock.Of(), Mock.Of(), - Mock.Of()); + Mock.Of(), + new NullTranscodeSessionStore()); await Assert.ThrowsAsync(exceptionType, () => sessionManager.GetAuthorizationToken( new User("test", "default", "default"), @@ -68,7 +70,8 @@ public class SessionManagerTests Mock.Of(), Mock.Of(), Mock.Of(), - Mock.Of()); + Mock.Of(), + new NullTranscodeSessionStore()); await Assert.ThrowsAsync(exceptionType, () => sessionManager.AuthenticateNewSessionInternal(authenticationRequest, false)); } @@ -238,7 +241,8 @@ public class SessionManagerTests Mock.Of(), Mock.Of(), Mock.Of(), - Mock.Of()); + Mock.Of(), + new NullTranscodeSessionStore()); } // All sessions are logged with the same client and device id on purpose, those values are taken