fix(session): tie directory ownership to the live connection

Ownership is claimed with a Lua check-and-set keyed on the instance holding the websocket, routing prefers a live controller over a local copy, the session list deduplicates by owner, removal is ownership-checked, undelivered routed messages surface, single-session lookups stop scanning the keyspace and directory writes leave the request path bounded by a timeout.
2026-09-24 23:50:01 +10:00
parent 6f362c33c9
commit 9e66708d87
17 changed files with 613 additions and 123 deletions
@@ -71,6 +71,9 @@ namespace Emby.Server.Implementations.Session
        private readonly ConcurrentDictionary<string, SessionInfo> _activeConnections
            = new(StringComparer.OrdinalIgnoreCase);

+        private readonly ConcurrentDictionary<string, long> _connectionEpochs = new(StringComparer.Ordinal);
+        private readonly ConcurrentDictionary<string, long> _lastDirectoryPublish = new(StringComparer.Ordinal);
+
        private readonly ConcurrentDictionary<string, ConcurrentDictionary<string, string>> _activeLiveStreamSessions
            = new(StringComparer.OrdinalIgnoreCase);

@@ -316,13 +319,13 @@ namespace Emby.Server.Implementations.Session
                    });
            }

-            await PublishToDirectoryAsync(session).ConfigureAwait(false);
+            QueueDirectoryPublish(session);

            return session;
        }

        /// <inheritdoc />
-        public void OnSessionControllerConnected(SessionInfo session)
+        public async Task OnSessionControllerConnected(SessionInfo session)
        {
            EventHelper.QueueEventIfNotNull(
                SessionControllerConnected,
@@ -333,9 +336,32 @@ namespace Emby.Server.Implementations.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);
+            // Ownership of the session belongs to whichever instance holds its connection, so this one
+            // claims it before the connection is used.
+            _lastDirectoryPublish[session.Id] = Environment.TickCount64;
+            await PublishToDirectoryAsync(session).ConfigureAwait(false);
+        }
+
+        // Keeps the directory write off the request path: the caller does not wait for Redis, and a
+        // session reporting playback every few seconds does not write on every report.
+        private void QueueDirectoryPublish(SessionInfo session)
+        {
+            if (!_directoryEnabled || string.IsNullOrEmpty(session.Id))
+            {
+                return;
+            }
+
+            var now = Environment.TickCount64;
+            var throttleMs = Math.Max(1000L, _sessionDirectoryOptions.RefreshIntervalSeconds * 500L);
+            var scheduled = _lastDirectoryPublish.AddOrUpdate(
+                session.Id,
+                now,
+                (_, last) => now - last >= throttleMs ? now : last);
+
+            if (scheduled == now)
+            {
+                _ = PublishToDirectoryAsync(session);
+            }
        }

        private async Task PublishToDirectoryAsync(SessionInfo session)
@@ -347,12 +373,16 @@ namespace Emby.Server.Implementations.Session

            try
            {
+                var connectedUtcTicks = GetConnectionEpoch(session);
+
                await _sessionDirectory.PublishAsync(
                    new SessionDirectoryEntry
                    {
                        OwnerPod = _podMessageBus.PodId,
+                        HoldsConnection = connectedUtcTicks > 0,
                        Session = ToSessionInfoDto(session)
-                    }).ConfigureAwait(false);
+                    },
+                    connectedUtcTicks).ConfigureAwait(false);
            }
            catch (Exception ex)
            {
@@ -360,14 +390,30 @@ namespace Emby.Server.Implementations.Session
            }
        }

+        // Ownership follows the connection, not the last request served: an instance without a live
+        // controller claims with epoch zero, which never displaces an instance that has one.
+        private long GetConnectionEpoch(SessionInfo session)
+        {
+            if (!session.SessionControllers.Any(i => i.IsSessionActive))
+            {
+                _connectionEpochs.TryRemove(session.Id, out _);
+                return 0;
+            }
+
+            return _connectionEpochs.GetOrAdd(session.Id, _ => DateTime.UtcNow.Ticks);
+        }
+
        private async ValueTask RemoveFromDirectoryAsync(SessionInfo session)
        {
+            _connectionEpochs.TryRemove(session.Id, out _);
+            _lastDirectoryPublish.TryRemove(session.Id, out _);
+
            if (!_directoryEnabled || string.IsNullOrEmpty(session.Id))
            {
                return;
            }

-            await _sessionDirectory.RemoveAsync(session.Id).ConfigureAwait(false);
+            await _sessionDirectory.RemoveAsync(session.Id, _podMessageBus.PodId).ConfigureAwait(false);
        }

        private async void RefreshSessionDirectory(object state)
@@ -402,10 +448,15 @@ namespace Emby.Server.Implementations.Session

        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 (!_directoryEnabled)
+            {
+                return null;
+            }

-            if (entry is null)
+            var entry = await _sessionDirectory.GetAsync(sessionId).ConfigureAwait(false);
+
+            if (entry?.Session is null
+                || string.Equals(entry.OwnerPod, _podMessageBus.PodId, StringComparison.Ordinal))
            {
                return null;
            }
@@ -428,18 +479,30 @@ namespace Emby.Server.Implementations.Session
                Capabilities = dto.Capabilities?.ToClientCapabilities()
            };

-            session.AddController(new RemoteSessionController(_podMessageBus, entry.OwnerPod, dto.Id, dto.IsActive, dto.SupportsMediaControl));
+            if (entry.HoldsConnection)
+            {
+                session.AddController(new RemoteSessionController(_podMessageBus, _logger, entry.OwnerPod, dto.Id, dto.SupportsMediaControl));
+            }

            return session;
        }

-        private async Task OnPodMessage(PodMessage message)
+        private Task OnPodMessage(PodMessage message)
        {
-            if (!string.Equals(message.Kind, RoutedSessionMessage.Kind, StringComparison.Ordinal))
+            switch (message.Kind)
            {
-                return;
+                case RoutedSessionMessage.Kind:
+                    return OnRoutedSessionMessage(message);
+                case RoutedAdditionalUserChange.Kind:
+                    OnRoutedAdditionalUserChange(message);
+                    return Task.CompletedTask;
+                default:
+                    return Task.CompletedTask;
            }
+        }

+        private async Task OnRoutedSessionMessage(PodMessage message)
+        {
            var routed = JsonSerializer.Deserialize<RoutedSessionMessage>(message.Payload, JsonDefaults.Options);
            if (routed is null)
            {
@@ -447,19 +510,46 @@ namespace Emby.Server.Implementations.Session
            }

            var session = Sessions.FirstOrDefault(i => string.Equals(i.Id, routed.SessionId, StringComparison.Ordinal));
-            if (session is null)
+            var controllers = session?.SessionControllers.Where(i => i.IsSessionActive).ToList();
+
+            if (controllers is null || controllers.Count == 0)
            {
-                _logger.LogDebug("Session {Session} was routed here but is no longer held by this instance.", routed.SessionId);
+                _logger.LogWarning(
+                    "A {MessageType} message for session {Session} was routed to this instance, which no longer holds its connection.",
+                    routed.MessageType,
+                    routed.SessionId);
                return;
            }

            using var data = JsonDocument.Parse(routed.Data);
-            foreach (var controller in session.SessionControllers)
+            foreach (var controller in controllers)
            {
                await controller.SendMessage(routed.MessageType, routed.MessageId, data.RootElement, CancellationToken.None).ConfigureAwait(false);
            }
        }

+        private void OnRoutedAdditionalUserChange(PodMessage message)
+        {
+            var routed = JsonSerializer.Deserialize<RoutedAdditionalUserChange>(message.Payload, JsonDefaults.Options);
+            var session = routed is null
+                ? null
+                : Sessions.FirstOrDefault(i => string.Equals(i.Id, routed.SessionId, StringComparison.Ordinal));
+
+            if (session is null)
+            {
+                return;
+            }
+
+            if (routed.Add)
+            {
+                AttachAdditionalUser(session, routed.UserId, _userManager.GetUserById(routed.UserId)?.Username);
+            }
+            else
+            {
+                DetachAdditionalUser(session, routed.UserId);
+            }
+        }
+
        /// <inheritdoc />
        public async Task CloseIfNeededAsync(SessionInfo session)
        {
@@ -1375,12 +1465,18 @@ namespace Emby.Server.Implementations.Session
            return session;
        }

-        // 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.
+        // A local SessionInfo without a live controller is a copy left behind by a request this instance
+        // happened to serve, not the connection: prefer the owner named by the directory over it.
        private async Task<SessionInfo> GetSessionToRemoteControl(string sessionId)
        {
-            var session = Sessions.FirstOrDefault(i => string.Equals(i.Id, sessionId, StringComparison.Ordinal))
-                ?? await GetRemoteSession(sessionId).ConfigureAwait(false);
+            var local = Sessions.FirstOrDefault(i => string.Equals(i.Id, sessionId, StringComparison.Ordinal));
+
+            if (local is not null && local.SessionControllers.Any(i => i.IsSessionActive))
+            {
+                return local;
+            }
+
+            var session = await GetRemoteSession(sessionId).ConfigureAwait(false) ?? local;

            if (session is null)
            {
@@ -1583,7 +1679,16 @@ namespace Emby.Server.Implementations.Session
        public async Task SendSyncPlayCommand(string sessionId, SendCommand command, CancellationToken cancellationToken)
        {
            CheckDisposed();
-            var session = GetSession(sessionId);
+
+            // SyncPlay group membership is instance-local, so a session listed by another instance is not
+            // reachable from here. It is skipped rather than reported as missing.
+            var session = GetSession(sessionId, false);
+            if (session is null)
+            {
+                _logger.LogDebug("SyncPlay command for session {Session} dropped; it is not held by this instance.", sessionId);
+                return;
+            }
+
            await SendMessageToSession(session, SessionMessageType.SyncPlayCommand, command, cancellationToken).ConfigureAwait(false);
        }

@@ -1591,7 +1696,14 @@ namespace Emby.Server.Implementations.Session
        public async Task SendSyncPlayGroupUpdate<T>(string sessionId, GroupUpdate<T> command, CancellationToken cancellationToken)
        {
            CheckDisposed();
-            var session = GetSession(sessionId);
+
+            var session = GetSession(sessionId, false);
+            if (session is null)
+            {
+                _logger.LogDebug("SyncPlay group update for session {Session} dropped; it is not held by this instance.", sessionId);
+                return;
+            }
+
            await SendMessageToSession(session, SessionMessageType.SyncPlayGroupUpdate, command, cancellationToken).ConfigureAwait(false);
        }

@@ -1764,17 +1876,18 @@ namespace Emby.Server.Implementations.Session
        /// <param name="controllingSessionId">The controlling session identifier.</param>
        /// <param name="sessionId">The session identifier.</param>
        /// <param name="userId">The user identifier.</param>
+        /// <returns>A task representing the operation.</returns>
        /// <exception cref="SecurityException">The controlling user is not allowed to attach the user to the session.</exception>
        /// <exception cref="ArgumentException">The requested user is already the primary user of the session.</exception>
-        public void AddAdditionalUser(string controllingSessionId, string sessionId, Guid userId)
+        public async Task AddAdditionalUser(string controllingSessionId, string sessionId, Guid userId)
        {
            CheckDisposed();

-            var session = GetSession(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);
                AssertCanAttachUser(controllingSession, userId);
            }
@@ -1784,18 +1897,16 @@ namespace Emby.Server.Implementations.Session
                throw new ArgumentException("The requested user is already the primary user of the session.");
            }

-            if (session.AdditionalUsers.All(i => !i.UserId.Equals(userId)))
-            {
-                var user = _userManager.GetUserById(userId)
-                    ?? throw new ArgumentException("The requested user does not exist.");
-                var newUser = new SessionUserInfo
-                {
-                    UserId = userId,
-                    UserName = user.Username
-                };
+            var user = _userManager.GetUserById(userId)
+                ?? throw new ArgumentException("The requested user does not exist.");

-                session.AdditionalUsers = [.. session.AdditionalUsers, newUser];
+            var local = GetSession(sessionId, false);
+            if (local is not null)
+            {
+                AttachAdditionalUser(local, userId, user.Username);
            }
+
+            await RouteAdditionalUserChange(sessionId, userId, true).ConfigureAwait(false);
        }

        /// <summary>
@@ -1804,17 +1915,18 @@ namespace Emby.Server.Implementations.Session
        /// <param name="controllingSessionId">The controlling session identifier.</param>
        /// <param name="sessionId">The session identifier.</param>
        /// <param name="userId">The user identifier.</param>
+        /// <returns>A task representing the operation.</returns>
        /// <exception cref="SecurityException">The controlling user is not allowed to control the session.</exception>
        /// <exception cref="ArgumentException">The requested user is already the primary user of the session.</exception>
-        public void RemoveAdditionalUser(string controllingSessionId, string sessionId, Guid userId)
+        public async Task RemoveAdditionalUser(string controllingSessionId, string sessionId, Guid userId)
        {
            CheckDisposed();

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

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

            if (session.UserId.Equals(userId))
@@ -1822,17 +1934,73 @@ namespace Emby.Server.Implementations.Session
                throw new ArgumentException("The requested user is already the primary user of the session.");
            }

-            var user = session.AdditionalUsers.FirstOrDefault(i => i.UserId.Equals(userId));
+            var local = GetSession(sessionId, false);
+            if (local is not null)
+            {
+                DetachAdditionalUser(local, userId);
+            }

-            if (user is not null)
+            await RouteAdditionalUserChange(sessionId, userId, false).ConfigureAwait(false);
+        }
+
+        private static void AttachAdditionalUser(SessionInfo session, Guid userId, string userName)
+        {
+            if (session.AdditionalUsers.All(i => !i.UserId.Equals(userId)))
+            {
+                session.AdditionalUsers = [.. session.AdditionalUsers, new SessionUserInfo { UserId = userId, UserName = userName }];
+            }
+        }
+
+        private static void DetachAdditionalUser(SessionInfo session, Guid userId)
+        {
+            var existing = session.AdditionalUsers.FirstOrDefault(i => i.UserId.Equals(userId));
+
+            if (existing is not null)
            {
                var list = session.AdditionalUsers.ToList();
-                list.Remove(user);
+                list.Remove(existing);

                session.AdditionalUsers = list.ToArray();
            }
        }

+        // The owner is the instance whose copy of the session is the one everyone else is shown, so the
+        // change has to be applied there as well as on whichever instance served the request.
+        private async Task RouteAdditionalUserChange(string sessionId, Guid userId, bool add)
+        {
+            if (!_directoryEnabled)
+            {
+                return;
+            }
+
+            var entry = await _sessionDirectory.GetAsync(sessionId).ConfigureAwait(false);
+
+            if (entry is null || string.Equals(entry.OwnerPod, _podMessageBus.PodId, StringComparison.Ordinal))
+            {
+                return;
+            }
+
+            var payload = new RoutedAdditionalUserChange
+            {
+                SessionId = sessionId,
+                UserId = userId,
+                Add = add
+            };
+
+            var delivered = await _podMessageBus.PublishAsync(
+                entry.OwnerPod,
+                new PodMessage
+                {
+                    Kind = RoutedAdditionalUserChange.Kind,
+                    Payload = JsonSerializer.Serialize(payload, JsonDefaults.Options)
+                }).ConfigureAwait(false);
+
+            if (delivered == 0)
+            {
+                _logger.LogWarning("Instance {OwnerPod} holds session {Session} but is not listening; the additional user change was not applied there.", entry.OwnerPod, sessionId);
+            }
+        }
+
        /// <summary>
        /// Authenticates the new session.
        /// </summary>
@@ -2227,8 +2395,12 @@ namespace Emby.Server.Implementations.Session
            CancellationToken cancellationToken)
        {
            var remote = await GetRemoteEntriesAsync(cancellationToken).ConfigureAwait(false);
+            var ownedElsewhere = remote.Select(entry => entry.Session.Id).ToHashSet(StringComparer.Ordinal);

+            // A session this instance only holds a copy of is reported by its owner, whose controllers are
+            // the ones that decide whether it is active and controllable.
            IEnumerable<SessionInfoDto> result = Sessions
+                .Where(i => !ownedElsewhere.Contains(i.Id))
                .Select(ToSessionInfoDto)
                .Concat(remote.Select(entry => entry.Session))
                .OrderByDescending(i => i.LastActivityDate);