feat(session): share the session directory between instances

Publish every session to valkey with the instance holding it, and route a command for a non-local session to its owner over a per-instance pub/sub channel. Entries expire, so a dead instance leaves the directory.
2026-09-24 22:48:48 +10:00
parent 6691b785c3
commit 00d0765152
24 changed files with 1203 additions and 30 deletions
@@ -5,6 +5,7 @@ using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Globalization;
 using System.Linq;
+using System.Text.Json;
 using System.Threading;
 using System.Threading.Tasks;
 using Jellyfin.Data;
@@ -15,6 +16,7 @@ using Jellyfin.Database.Implementations.Entities;
 using Jellyfin.Database.Implementations.Entities.Security;
 using Jellyfin.Database.Implementations.Enums;
 using Jellyfin.Extensions;
+using Jellyfin.Extensions.Json;
 using MediaBrowser.Common.Events;
 using MediaBrowser.Common.Extensions;
 using MediaBrowser.Controller;
@@ -39,6 +41,7 @@ using MediaBrowser.Model.SyncPlay;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
 using Episode = MediaBrowser.Controller.Entities.TV.Episode;

 namespace Emby.Server.Implementations.Session
@@ -60,6 +63,9 @@ namespace Emby.Server.Implementations.Session
        private readonly IMediaSourceManager _mediaSourceManager;
        private readonly IServerApplicationHost _appHost;
        private readonly IDeviceManager _deviceManager;
+        private readonly ISessionDirectory _sessionDirectory;
+        private readonly IPodMessageBus _podMessageBus;
+        private readonly SessionDirectoryOptions _sessionDirectoryOptions;
        private readonly CancellationTokenRegistration _shutdownCallback;
        private readonly ConcurrentDictionary<string, SessionInfo> _activeConnections
            = new(StringComparer.OrdinalIgnoreCase);
@@ -69,6 +75,7 @@ namespace Emby.Server.Implementations.Session

        private Timer _idleTimer;
        private Timer _inactiveTimer;
+        private Timer _directoryTimer;

        private DtoOptions _itemInfoDtoOptions;
        private bool _disposed;
@@ -89,6 +96,9 @@ 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="sessionDirectory">Instance of <see cref="ISessionDirectory"/> interface.</param>
+        /// <param name="podMessageBus">Instance of <see cref="IPodMessageBus"/> interface.</param>
+        /// <param name="sessionDirectoryOptions">The session directory options.</param>
        public SessionManager(
            ILogger<SessionManager> logger,
            IEventManager eventManager,
@@ -102,7 +112,10 @@ namespace Emby.Server.Implementations.Session
            IServerApplicationHost appHost,
            IDeviceManager deviceManager,
            IMediaSourceManager mediaSourceManager,
-            IHostApplicationLifetime hostApplicationLifetime)
+            IHostApplicationLifetime hostApplicationLifetime,
+            ISessionDirectory sessionDirectory,
+            IPodMessageBus podMessageBus,
+            IOptions<SessionDirectoryOptions> sessionDirectoryOptions)
        {
            _logger = logger;
            _eventManager = eventManager;
@@ -116,9 +129,19 @@ namespace Emby.Server.Implementations.Session
            _appHost = appHost;
            _deviceManager = deviceManager;
            _mediaSourceManager = mediaSourceManager;
+            _sessionDirectory = sessionDirectory;
+            _podMessageBus = podMessageBus;
+            _sessionDirectoryOptions = sessionDirectoryOptions.Value;
            _shutdownCallback = hostApplicationLifetime.ApplicationStopping.Register(OnApplicationStopping);

            _deviceManager.DeviceOptionsUpdated += OnDeviceManagerDeviceOptionsUpdated;
+
+            if (_sessionDirectory is not NullSessionDirectory)
+            {
+                _podMessageBus.Subscribe(OnPodMessage);
+                var interval = TimeSpan.FromSeconds(Math.Max(1, _sessionDirectoryOptions.RefreshIntervalSeconds));
+                _directoryTimer = new Timer(RefreshSessionDirectory, null, interval, interval);
+            }
        }

        /// <summary>
@@ -218,6 +241,8 @@ namespace Emby.Server.Implementations.Session

            _eventManager.Publish(new SessionEndedEventArgs(info));

+            await RemoveFromDirectoryAsync(info).ConfigureAwait(false);
+
            await info.DisposeAsync().ConfigureAwait(false);
        }

@@ -288,6 +313,8 @@ namespace Emby.Server.Implementations.Session
                    });
            }

+            await PublishToDirectoryAsync(session).ConfigureAwait(false);
+
            return session;
        }

@@ -302,6 +329,134 @@ namespace Emby.Server.Implementations.Session
                    SessionInfo = session
                },
                _logger);
+
+            // The session only becomes remote-controllable once it has a connection, so the other
+            // instances have to be told again now that it does.
+            _ = PublishToDirectoryAsync(session);
+        }
+
+        private bool DirectoryEnabled => _sessionDirectory is not NullSessionDirectory;
+
+        private async Task PublishToDirectoryAsync(SessionInfo session)
+        {
+            if (!DirectoryEnabled || string.IsNullOrEmpty(session.Id))
+            {
+                return;
+            }
+
+            try
+            {
+                await _sessionDirectory.PublishAsync(
+                    new SessionDirectoryEntry
+                    {
+                        OwnerPod = _podMessageBus.PodId,
+                        Session = ToSessionInfoDto(session)
+                    }).ConfigureAwait(false);
+            }
+            catch (Exception ex)
+            {
+                _logger.LogDebug(ex, "Error publishing session {Session} to the directory.", session.Id);
+            }
+        }
+
+        private async ValueTask RemoveFromDirectoryAsync(SessionInfo session)
+        {
+            if (!DirectoryEnabled || string.IsNullOrEmpty(session.Id))
+            {
+                return;
+            }
+
+            await _sessionDirectory.RemoveAsync(session.Id).ConfigureAwait(false);
+        }
+
+        private async void RefreshSessionDirectory(object state)
+        {
+            try
+            {
+                foreach (var session in _activeConnections.Values)
+                {
+                    await PublishToDirectoryAsync(session).ConfigureAwait(false);
+                }
+            }
+            catch (Exception ex)
+            {
+                _logger.LogDebug(ex, "Error refreshing the session directory.");
+            }
+        }
+
+        private async Task<IReadOnlyList<SessionDirectoryEntry>> GetRemoteEntriesAsync(CancellationToken cancellationToken)
+        {
+            if (!DirectoryEnabled)
+            {
+                return Array.Empty<SessionDirectoryEntry>();
+            }
+
+            var entries = await _sessionDirectory.GetAllAsync(cancellationToken).ConfigureAwait(false);
+
+            return entries
+                .Where(entry => entry.Session is not null
+                    && !string.Equals(entry.OwnerPod, _podMessageBus.PodId, StringComparison.Ordinal))
+                .ToList();
+        }
+
+        private async Task<SessionInfo> GetRemoteSession(string sessionId)
+        {
+            var entries = await GetRemoteEntriesAsync(CancellationToken.None).ConfigureAwait(false);
+            var entry = entries.FirstOrDefault(i => string.Equals(i.Session.Id, sessionId, StringComparison.Ordinal));
+
+            if (entry is null)
+            {
+                return null;
+            }
+
+            var dto = entry.Session;
+            var session = new SessionInfo(this, _logger)
+            {
+                Id = dto.Id,
+                UserId = dto.UserId,
+                UserName = dto.UserName,
+                Client = dto.Client,
+                DeviceId = dto.DeviceId,
+                DeviceName = dto.DeviceName,
+                DeviceType = dto.DeviceType,
+                ApplicationVersion = dto.ApplicationVersion,
+                RemoteEndPoint = dto.RemoteEndPoint,
+                LastActivityDate = dto.LastActivityDate,
+                ServerId = dto.ServerId,
+                AdditionalUsers = dto.AdditionalUsers ?? [],
+                Capabilities = dto.Capabilities?.ToClientCapabilities()
+            };
+
+            session.AddController(new RemoteSessionController(_podMessageBus, entry.OwnerPod, dto.Id, dto.IsActive, dto.SupportsMediaControl));
+
+            return session;
+        }
+
+        private async Task OnPodMessage(PodMessage message)
+        {
+            if (!string.Equals(message.Kind, RoutedSessionMessage.Kind, StringComparison.Ordinal))
+            {
+                return;
+            }
+
+            var routed = JsonSerializer.Deserialize<RoutedSessionMessage>(message.Payload, JsonDefaults.Options);
+            if (routed is null)
+            {
+                return;
+            }
+
+            var session = Sessions.FirstOrDefault(i => string.Equals(i.Id, routed.SessionId, StringComparison.Ordinal));
+            if (session is null)
+            {
+                _logger.LogDebug("Session {Session} was routed here but is no longer held by this instance.", routed.SessionId);
+                return;
+            }
+
+            using var data = JsonDocument.Parse(routed.Data);
+            foreach (var controller in session.SessionControllers)
+            {
+                await controller.SendMessage(routed.MessageType, routed.MessageId, data.RootElement, CancellationToken.None).ConfigureAwait(false);
+            }
        }

        /// <inheritdoc />
@@ -1219,10 +1374,12 @@ namespace Emby.Server.Implementations.Session
            return session;
        }

-        private SessionInfo GetSessionToRemoteControl(string sessionId)
+        // A session held by another instance is reachable too: the returned SessionInfo carries a
+        // controller that forwards to its owner instead of writing to a local connection.
+        private async Task<SessionInfo> GetSessionToRemoteControl(string sessionId)
        {
-            // Accept either device id or session id
-            var session = Sessions.FirstOrDefault(i => string.Equals(i.Id, sessionId, StringComparison.Ordinal));
+            var session = Sessions.FirstOrDefault(i => string.Equals(i.Id, sessionId, StringComparison.Ordinal))
+                ?? await GetRemoteSession(sessionId).ConfigureAwait(false);

            if (session is null)
            {
@@ -1291,19 +1448,19 @@ namespace Emby.Server.Implementations.Session
        }

        /// <inheritdoc />
-        public Task SendGeneralCommand(string controllingSessionId, string sessionId, GeneralCommand command, CancellationToken cancellationToken)
+        public async Task SendGeneralCommand(string controllingSessionId, string sessionId, GeneralCommand command, CancellationToken cancellationToken)
        {
            CheckDisposed();

-            var session = GetSessionToRemoteControl(sessionId);
+            var session = await GetSessionToRemoteControl(sessionId).ConfigureAwait(false);

            if (!string.IsNullOrEmpty(controllingSessionId))
            {
-                var controllingSession = GetSession(controllingSessionId);
+                var controllingSession = await GetSessionToRemoteControl(controllingSessionId).ConfigureAwait(false);
                AssertCanControl(session, controllingSession);
            }

-            return SendMessageToSession(session, SessionMessageType.GeneralCommand, command, cancellationToken);
+            await SendMessageToSession(session, SessionMessageType.GeneralCommand, command, cancellationToken).ConfigureAwait(false);
        }

        private static async Task SendMessageToSession<T>(SessionInfo session, SessionMessageType name, T data, CancellationToken cancellationToken)
@@ -1340,7 +1497,7 @@ namespace Emby.Server.Implementations.Session
        {
            CheckDisposed();

-            var session = GetSessionToRemoteControl(sessionId);
+            var session = await GetSessionToRemoteControl(sessionId).ConfigureAwait(false);

            var user = session.UserId.IsEmpty() ? null : _userManager.GetUserById(session.UserId);

@@ -1410,7 +1567,7 @@ namespace Emby.Server.Implementations.Session

            if (!string.IsNullOrEmpty(controllingSessionId))
            {
-                var controllingSession = GetSession(controllingSessionId);
+                var controllingSession = await GetSessionToRemoteControl(controllingSessionId).ConfigureAwait(false);
                AssertCanControl(session, controllingSession);
                if (!controllingSession.UserId.IsEmpty())
                {
@@ -1521,15 +1678,15 @@ namespace Emby.Server.Implementations.Session
        }

        /// <inheritdoc />
-        public Task SendPlaystateCommand(string controllingSessionId, string sessionId, PlaystateRequest command, CancellationToken cancellationToken)
+        public async Task SendPlaystateCommand(string controllingSessionId, string sessionId, PlaystateRequest command, CancellationToken cancellationToken)
        {
            CheckDisposed();

-            var session = GetSessionToRemoteControl(sessionId);
+            var session = await GetSessionToRemoteControl(sessionId).ConfigureAwait(false);

            if (!string.IsNullOrEmpty(controllingSessionId))
            {
-                var controllingSession = GetSession(controllingSessionId);
+                var controllingSession = await GetSessionToRemoteControl(controllingSessionId).ConfigureAwait(false);
                AssertCanControl(session, controllingSession);
                if (!controllingSession.UserId.IsEmpty())
                {
@@ -1537,7 +1694,7 @@ namespace Emby.Server.Implementations.Session
                }
            }

-            return SendMessageToSession(session, SessionMessageType.Playstate, command, cancellationToken);
+            await SendMessageToSession(session, SessionMessageType.Playstate, command, cancellationToken).ConfigureAwait(false);
        }

        private void AssertCanControl(SessionInfo session, SessionInfo controllingSession)
@@ -2060,14 +2217,21 @@ namespace Emby.Server.Implementations.Session
        }

        /// <inheritdoc/>
-        public IReadOnlyList<SessionInfoDto> GetSessions(
+        public async Task<IReadOnlyList<SessionInfoDto>> GetSessions(
            Guid userId,
            string deviceId,
            int? activeWithinSeconds,
            Guid? controllableUserToCheck,
-            bool isApiKey)
+            bool isApiKey,
+            CancellationToken cancellationToken)
        {
-            var result = Sessions;
+            var remote = await GetRemoteEntriesAsync(cancellationToken).ConfigureAwait(false);
+
+            IEnumerable<SessionInfoDto> result = Sessions
+                .Select(ToSessionInfoDto)
+                .Concat(remote.Select(entry => entry.Session))
+                .OrderByDescending(i => i.LastActivityDate);
+
            if (!string.IsNullOrEmpty(deviceId))
            {
                result = result.Where(i => string.Equals(i.DeviceId, deviceId, StringComparison.OrdinalIgnoreCase));
@@ -2115,7 +2279,7 @@ namespace Emby.Server.Implementations.Session
                if (!userCanControlOthers)
                {
                    // User cannot control other user's sessions, validate user id.
-                    result = result.Where(i => i.UserId.IsEmpty() || i.ContainsUser(userId));
+                    result = result.Where(i => i.UserId.IsEmpty() || ContainsUser(i, userId));
                }

                result = result.Where(i =>
@@ -2136,7 +2300,7 @@ namespace Emby.Server.Implementations.Session
            else if (!userIsAdmin)
            {
                // Request isn't from administrator, limit to "own" sessions.
-                result = result.Where(i => i.UserId.IsEmpty() || i.ContainsUser(userId));
+                result = result.Where(i => i.UserId.IsEmpty() || ContainsUser(i, userId));
            }

            if (!userIsAdmin)
@@ -2159,7 +2323,18 @@ namespace Emby.Server.Implementations.Session
                result = result.Where(i => i.LastActivityDate >= minActiveDate);
            }

-            return result.Select(ToSessionInfoDto).ToList();
+            return result.ToList();
+        }
+
+        private static bool ContainsUser(SessionInfoDto session, Guid userId)
+        {
+            if (session.UserId.Equals(userId))
+            {
+                return true;
+            }
+
+            return session.AdditionalUsers is not null
+                && session.AdditionalUsers.Any(i => i.UserId.Equals(userId));
        }

        /// <inheritdoc />
@@ -2234,6 +2409,12 @@ namespace Emby.Server.Implementations.Session
                _inactiveTimer = null;
            }

+            if (_directoryTimer is not null)
+            {
+                await _directoryTimer.DisposeAsync().ConfigureAwait(false);
+                _directoryTimer = null;
+            }
+
            await _shutdownCallback.DisposeAsync().ConfigureAwait(false);

            _deviceManager.DeviceOptionsUpdated -= OnDeviceManagerDeviceOptionsUpdated;
@@ -2257,6 +2438,7 @@ namespace Emby.Server.Implementations.Session
            // Close open websockets to allow Kestrel to shut down cleanly
            foreach (var session in _activeConnections.Values)
            {
+                await RemoveFromDirectoryAsync(session).ConfigureAwait(false);
                await session.DisposeAsync().ConfigureAwait(false);
            }