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

feat(session): record open live streams in the session store

Browse Source 
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
This commit is contained in:
Unkin Agent
2026-09-11 23:56:35 +10:00
parent 6ccabd4c72
commit e3571eb2a4
4 changed files with 113 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Emby.Server.Implementations/Session/SessionManager.cs
+37 -4
View File
@@ -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<string, SessionInfo> _activeConnections
= new(StringComparer.OrdinalIgnoreCase);
@@ -89,6 +91,7 @@ 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,
@@ -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<string, string>());
@@ -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);
}
}
/// <summary>
@@ -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
tests/Jellyfin.Server.Implementations.Tests/SessionManager/IdlePlaybackTests.cs
+3 -1
View File
@@ -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<IServerApplicationHost>(),
Mock.Of<IDeviceManager>(),
Mock.Of<IMediaSourceManager>(),
Mock.Of<IHostApplicationLifetime>());
Mock.Of<IHostApplicationLifetime>(),
new NullTranscodeSessionStore());
var session = await sessionManager.LogSessionActivity(
"Test Client",
"1.0.0",
tests/Jellyfin.Server.Implementations.Tests/SessionManager/LiveStreamHaRecordTests.cs
+66
View File
@@ -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<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
+7 -3
View File
@@ -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<IServerApplicationHost>(),
Mock.Of<IDeviceManager>(),
Mock.Of<IMediaSourceManager>(),
Mock.Of<IHostApplicationLifetime>());
Mock.Of<IHostApplicationLifetime>(),
new NullTranscodeSessionStore());
await Assert.ThrowsAsync(exceptionType, () => sessionManager.GetAuthorizationToken(
new User("test", "default", "default"),
@@ -68,7 +70,8 @@ public class SessionManagerTests
Mock.Of<IServerApplicationHost>(),
Mock.Of<IDeviceManager>(),
Mock.Of<IMediaSourceManager>(),
Mock.Of<IHostApplicationLifetime>());
Mock.Of<IHostApplicationLifetime>(),
new NullTranscodeSessionStore());
await Assert.ThrowsAsync(exceptionType, () => sessionManager.AuthenticateNewSessionInternal(authenticationRequest, false));
}
@@ -238,7 +241,8 @@ public class SessionManagerTests
Mock.Of<IServerApplicationHost>(),
Mock.Of<IDeviceManager>(),
Mock.Of<IMediaSourceManager>(),
Mock.Of<IHostApplicationLifetime>());
Mock.Of<IHostApplicationLifetime>(),
new NullTranscodeSessionStore());
}
// All sessions are logged with the same client and device id on purpose, those values are taken