From 9e66708d874e00ab6560028cb56093ee258f8d63 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Thu, 24 Sep 2026 23:50:01 +1000 Subject: [PATCH] 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. --- .../Session/RedisPodMessageBus.cs | 11 +- .../Session/RedisSessionDirectory.cs | 114 ++++++-- .../Session/RemoteSessionController.cs | 27 +- .../Session/SessionManager.cs | 256 +++++++++++++++--- .../Session/SessionWebSocketListener.cs | 6 +- Jellyfin.Api/Controllers/SessionController.cs | 8 +- .../Session/IPodMessageBus.cs | 8 +- .../Session/ISessionDirectory.cs | 24 +- .../Session/ISessionManager.cs | 9 +- .../Session/NullPodMessageBus.cs | 6 +- .../Session/NullSessionDirectory.cs | 10 +- .../Session/RoutedAdditionalUserChange.cs | 30 ++ .../Session/SessionDirectoryEntry.cs | 6 + .../Session/SessionDirectoryOptions.cs | 5 + README.md | 1 + .../SessionManager/SessionManagerTests.cs | 6 +- .../SessionDirectoryReplicaTests.cs | 209 ++++++++++++-- 17 files changed, 613 insertions(+), 123 deletions(-) create mode 100644 MediaBrowser.Controller/Session/RoutedAdditionalUserChange.cs diff --git a/Emby.Server.Implementations/Session/RedisPodMessageBus.cs b/Emby.Server.Implementations/Session/RedisPodMessageBus.cs index bad3d80ecd..955be0454d 100644 --- a/Emby.Server.Implementations/Session/RedisPodMessageBus.cs +++ b/Emby.Server.Implementations/Session/RedisPodMessageBus.cs @@ -1,5 +1,6 @@ using System; using System.Text.Json; +using System.Threading; using System.Threading.Tasks; using Jellyfin.Extensions.Json; using MediaBrowser.Controller.Session; @@ -16,6 +17,8 @@ public sealed class RedisPodMessageBus : IPodMessageBus { private const string ChannelPrefix = "jellyfin:pod:"; + private static readonly TimeSpan _publishTimeout = TimeSpan.FromSeconds(5); + private static readonly JsonSerializerOptions _jsonOptions = JsonDefaults.Options; private readonly ISubscriber _subscriber; @@ -39,7 +42,7 @@ public sealed class RedisPodMessageBus : IPodMessageBus public string PodId { get; } /// - public void Publish(string targetPod, PodMessage message) + public async Task PublishAsync(string targetPod, PodMessage message, CancellationToken cancellationToken = default) { ArgumentException.ThrowIfNullOrEmpty(targetPod); ArgumentNullException.ThrowIfNull(message); @@ -48,14 +51,14 @@ public sealed class RedisPodMessageBus : IPodMessageBus try { - _subscriber.Publish( + return await _subscriber.PublishAsync( RedisChannel.Literal(ChannelPrefix + targetPod), - JsonSerializer.Serialize(message, _jsonOptions), - CommandFlags.FireAndForget); + JsonSerializer.Serialize(message, _jsonOptions)).WaitAsync(_publishTimeout, cancellationToken).ConfigureAwait(false); } catch (Exception ex) { _logger.LogWarning(ex, "Failed to send a {Kind} message to {TargetPod}.", message.Kind, targetPod); + return 0; } } diff --git a/Emby.Server.Implementations/Session/RedisSessionDirectory.cs b/Emby.Server.Implementations/Session/RedisSessionDirectory.cs index 20980983a6..14be5e0162 100644 --- a/Emby.Server.Implementations/Session/RedisSessionDirectory.cs +++ b/Emby.Server.Implementations/Session/RedisSessionDirectory.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.Globalization; using System.Linq; using System.Text.Json; using System.Threading; @@ -13,12 +14,47 @@ using StackExchange.Redis; namespace Emby.Server.Implementations.Session; /// -/// A Redis-backed . Each entry is a key with an expiry, so the sessions of -/// an instance that stops refreshing them disappear on their own. +/// A Redis-backed . A session is owned by the instance holding its +/// connection: ownership is claimed through a check-and-set, so an instance that only served a request +/// for the session cannot take it from the instance the device is actually connected to. Each entry is a +/// key with an expiry, so the sessions of an instance that stops refreshing them disappear on their own. /// public sealed class RedisSessionDirectory : ISessionDirectory { private const string KeyPrefix = "jellyfin:session:"; + private const string OwnerKeyPrefix = "jellyfin:sessionowner:"; + + /// + /// Lua script for an atomic ownership claim. The owner key holds pod|connectedTicks, where + /// the ticks are zero for an instance that holds no connection. A claim by another instance is + /// refused unless its connection is newer than the recorded one, so the instance holding the live + /// connection keeps ownership however many requests the others serve. + /// + private const string ClaimScript = @" +local current = redis.call('GET', KEYS[1]) +if current then + local separator = string.find(current, '|', 1, true) + local owner = string.sub(current, 1, separator - 1) + local connected = tonumber(string.sub(current, separator + 1)) + if owner ~= ARGV[1] and connected > 0 and tonumber(ARGV[2]) <= connected then + return 0 + end +end +redis.call('SET', KEYS[1], ARGV[1] .. '|' .. ARGV[2], 'PX', ARGV[4]) +redis.call('SET', KEYS[2], ARGV[3], 'PX', ARGV[4]) +return 1"; + + /// + /// Lua script for an atomic, ownership-checked removal, so that an instance ending its own copy of a + /// session cannot erase the entry of the instance still holding the connection. + /// + private const string ReleaseScript = @" +local current = redis.call('GET', KEYS[1]) +if not current then return 0 end +local separator = string.find(current, '|', 1, true) +if string.sub(current, 1, separator - 1) ~= ARGV[1] then return 0 end +redis.call('DEL', KEYS[1], KEYS[2]) +return 1"; private static readonly JsonSerializerOptions _jsonOptions = JsonDefaults.Options; @@ -47,35 +83,52 @@ public sealed class RedisSessionDirectory : ISessionDirectory _logger = logger; } - private TimeSpan EntryTtl => TimeSpan.FromSeconds(Math.Max(1, _options.EntryTtlSeconds)); + private long EntryTtlMs => Math.Max(1, _options.EntryTtlSeconds) * 1000L; + + private TimeSpan OperationTimeout => TimeSpan.FromSeconds(Math.Max(1, _options.OperationTimeoutSeconds)); /// - public async Task PublishAsync(SessionDirectoryEntry entry, CancellationToken cancellationToken = default) + public async Task PublishAsync(SessionDirectoryEntry entry, long connectedUtcTicks, CancellationToken cancellationToken = default) { ArgumentNullException.ThrowIfNull(entry); var sessionId = entry.Session?.Id; if (string.IsNullOrEmpty(sessionId)) { - return; + return false; } try { - await _db.StringSetAsync(KeyPrefix + sessionId, JsonSerializer.Serialize(entry, _jsonOptions), EntryTtl).ConfigureAwait(false); + var claimed = (long?)await _db.ScriptEvaluateAsync( + ClaimScript, + keys: new RedisKey[] { OwnerKeyPrefix + sessionId, KeyPrefix + sessionId }, + values: new RedisValue[] + { + entry.OwnerPod, + connectedUtcTicks.ToString(CultureInfo.InvariantCulture), + JsonSerializer.Serialize(entry, _jsonOptions), + EntryTtlMs + }).WaitAsync(OperationTimeout, cancellationToken).ConfigureAwait(false); + + return claimed == 1; } catch (Exception ex) { _logger.LogWarning(ex, "Failed to publish session {SessionId}; it stays invisible to the other instances.", sessionId); + return false; } } /// - public async Task RemoveAsync(string sessionId, CancellationToken cancellationToken = default) + public async Task RemoveAsync(string sessionId, string ownerPod, CancellationToken cancellationToken = default) { try { - await _db.KeyDeleteAsync(KeyPrefix + sessionId).ConfigureAwait(false); + await _db.ScriptEvaluateAsync( + ReleaseScript, + keys: new RedisKey[] { OwnerKeyPrefix + sessionId, KeyPrefix + sessionId }, + values: new RedisValue[] { ownerPod }).WaitAsync(OperationTimeout, cancellationToken).ConfigureAwait(false); } catch (Exception ex) { @@ -83,6 +136,22 @@ public sealed class RedisSessionDirectory : ISessionDirectory } } + /// + public async Task GetAsync(string sessionId, CancellationToken cancellationToken = default) + { + try + { + var raw = await _db.StringGetAsync(KeyPrefix + sessionId).WaitAsync(OperationTimeout, cancellationToken).ConfigureAwait(false); + + return raw.HasValue ? Deserialize(raw) : null; + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Failed to read session {SessionId} from the directory.", sessionId); + return null; + } + } + /// public async Task> GetAllAsync(CancellationToken cancellationToken = default) { @@ -103,7 +172,8 @@ public sealed class RedisSessionDirectory : ISessionDirectory keys.Add(key); } - var values = await Task.WhenAll(keys.Select(key => _db.StringGetAsync(key))).ConfigureAwait(false); + var values = await Task.WhenAll(keys.Select(key => _db.StringGetAsync(key))) + .WaitAsync(OperationTimeout, cancellationToken).ConfigureAwait(false); foreach (var raw in values) { @@ -112,17 +182,10 @@ public sealed class RedisSessionDirectory : ISessionDirectory continue; } - try + var entry = Deserialize(raw); + if (entry?.Session is not null) { - var entry = JsonSerializer.Deserialize(raw.ToString(), _jsonOptions); - if (entry?.Session is not null) - { - entries.Add(entry); - } - } - catch (JsonException ex) - { - _logger.LogWarning(ex, "Failed to deserialize a session directory entry."); + entries.Add(entry); } } } @@ -136,4 +199,17 @@ public sealed class RedisSessionDirectory : ISessionDirectory return entries; } + + private SessionDirectoryEntry? Deserialize(RedisValue raw) + { + try + { + return JsonSerializer.Deserialize(raw.ToString(), _jsonOptions); + } + catch (JsonException ex) + { + _logger.LogWarning(ex, "Failed to deserialize a session directory entry."); + return null; + } + } } diff --git a/Emby.Server.Implementations/Session/RemoteSessionController.cs b/Emby.Server.Implementations/Session/RemoteSessionController.cs index 51cd17a91a..e1c3cb005d 100644 --- a/Emby.Server.Implementations/Session/RemoteSessionController.cs +++ b/Emby.Server.Implementations/Session/RemoteSessionController.cs @@ -1,10 +1,13 @@ using System; +using System.Globalization; using System.Text.Json; using System.Threading; using System.Threading.Tasks; using Jellyfin.Extensions.Json; +using MediaBrowser.Common.Extensions; using MediaBrowser.Controller.Session; using MediaBrowser.Model.Session; +using Microsoft.Extensions.Logging; namespace Emby.Server.Implementations.Session; @@ -15,6 +18,7 @@ namespace Emby.Server.Implementations.Session; public sealed class RemoteSessionController : ISessionController { private readonly IPodMessageBus _bus; + private readonly ILogger _logger; private readonly string _ownerPod; private readonly string _sessionId; @@ -22,27 +26,27 @@ public sealed class RemoteSessionController : ISessionController /// Initializes a new instance of the class. /// /// The cross-instance bus. + /// The logger. /// The instance holding the connection. /// The session identifier. - /// Whether the owner reported the session as active. /// Whether the owner reported the session as controllable. - public RemoteSessionController(IPodMessageBus bus, string ownerPod, string sessionId, bool isSessionActive, bool supportsMediaControl) + public RemoteSessionController(IPodMessageBus bus, ILogger logger, string ownerPod, string sessionId, bool supportsMediaControl) { _bus = bus; + _logger = logger; _ownerPod = ownerPod; _sessionId = sessionId; - IsSessionActive = isSessionActive; SupportsMediaControl = supportsMediaControl; } /// - public bool IsSessionActive { get; } + public bool IsSessionActive => true; /// public bool SupportsMediaControl { get; } /// - public Task SendMessage(SessionMessageType name, Guid messageId, T data, CancellationToken cancellationToken) + public async Task SendMessage(SessionMessageType name, Guid messageId, T data, CancellationToken cancellationToken) { var routed = new RoutedSessionMessage { @@ -52,14 +56,21 @@ public sealed class RemoteSessionController : ISessionController Data = JsonSerializer.Serialize(data, JsonDefaults.Options) }; - _bus.Publish( + var delivered = await _bus.PublishAsync( _ownerPod, new PodMessage { Kind = RoutedSessionMessage.Kind, Payload = JsonSerializer.Serialize(routed, JsonDefaults.Options) - }); + }, + cancellationToken).ConfigureAwait(false); - return Task.CompletedTask; + if (delivered == 0) + { + _logger.LogWarning("Instance {OwnerPod} holds session {SessionId} but is not listening; the {MessageType} message was not delivered.", _ownerPod, _sessionId, name); + + throw new ResourceNotFoundException( + string.Format(CultureInfo.InvariantCulture, "The instance holding session {0} is unreachable.", _sessionId)); + } } } diff --git a/Emby.Server.Implementations/Session/SessionManager.cs b/Emby.Server.Implementations/Session/SessionManager.cs index c0adde9a5f..5a62ecca77 100644 --- a/Emby.Server.Implementations/Session/SessionManager.cs +++ b/Emby.Server.Implementations/Session/SessionManager.cs @@ -71,6 +71,9 @@ namespace Emby.Server.Implementations.Session private readonly ConcurrentDictionary _activeConnections = new(StringComparer.OrdinalIgnoreCase); + private readonly ConcurrentDictionary _connectionEpochs = new(StringComparer.Ordinal); + private readonly ConcurrentDictionary _lastDirectoryPublish = new(StringComparer.Ordinal); + private readonly ConcurrentDictionary> _activeLiveStreamSessions = new(StringComparer.OrdinalIgnoreCase); @@ -316,13 +319,13 @@ namespace Emby.Server.Implementations.Session }); } - await PublishToDirectoryAsync(session).ConfigureAwait(false); + QueueDirectoryPublish(session); return session; } /// - 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 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(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(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); + } + } + /// 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 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(string sessionId, GroupUpdate 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 /// The controlling session identifier. /// The session identifier. /// The user identifier. + /// A task representing the operation. /// The controlling user is not allowed to attach the user to the session. /// The requested user is already the primary user of the session. - 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); } /// @@ -1804,17 +1915,18 @@ namespace Emby.Server.Implementations.Session /// The controlling session identifier. /// The session identifier. /// The user identifier. + /// A task representing the operation. /// The controlling user is not allowed to control the session. /// The requested user is already the primary user of the session. - 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); + } + } + /// /// Authenticates the new session. /// @@ -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 result = Sessions + .Where(i => !ownedElsewhere.Contains(i.Id)) .Select(ToSessionInfoDto) .Concat(remote.Select(entry => entry.Session)) .OrderByDescending(i => i.LastActivityDate); diff --git a/Emby.Server.Implementations/Session/SessionWebSocketListener.cs b/Emby.Server.Implementations/Session/SessionWebSocketListener.cs index e81edc82c6..563a5abef0 100644 --- a/Emby.Server.Implementations/Session/SessionWebSocketListener.cs +++ b/Emby.Server.Implementations/Session/SessionWebSocketListener.cs @@ -114,11 +114,11 @@ namespace Emby.Server.Implementations.Session public async Task ProcessWebSocketConnectedAsync(IWebSocketConnection connection, HttpContext httpContext) { var session = await RequestHelpers.GetSession(_sessionManager, _userManager, httpContext).ConfigureAwait(false); - EnsureController(session, connection); + await EnsureController(session, connection).ConfigureAwait(false); await KeepAliveWebSocket(connection).ConfigureAwait(false); } - private void EnsureController(SessionInfo session, IWebSocketConnection connection) + private async Task EnsureController(SessionInfo session, IWebSocketConnection connection) { var controllerInfo = session.EnsureController( s => new WebSocketController(_loggerFactory.CreateLogger(), s, _sessionManager)); @@ -126,7 +126,7 @@ namespace Emby.Server.Implementations.Session var controller = (WebSocketController)controllerInfo.Item1; controller.AddWebSocket(connection); - _sessionManager.OnSessionControllerConnected(session); + await _sessionManager.OnSessionControllerConnected(session).ConfigureAwait(false); } /// diff --git a/Jellyfin.Api/Controllers/SessionController.cs b/Jellyfin.Api/Controllers/SessionController.cs index 7d6a162d8f..05506d7b0d 100644 --- a/Jellyfin.Api/Controllers/SessionController.cs +++ b/Jellyfin.Api/Controllers/SessionController.cs @@ -311,10 +311,10 @@ public class SessionController : BaseJellyfinApiController [FromRoute, Required] string sessionId, [FromRoute, Required] Guid userId) { - _sessionManager.AddAdditionalUser( + await _sessionManager.AddAdditionalUser( await RequestHelpers.GetSessionId(_sessionManager, _userManager, HttpContext).ConfigureAwait(false), sessionId, - userId); + userId).ConfigureAwait(false); return NoContent(); } @@ -332,10 +332,10 @@ public class SessionController : BaseJellyfinApiController [FromRoute, Required] string sessionId, [FromRoute, Required] Guid userId) { - _sessionManager.RemoveAdditionalUser( + await _sessionManager.RemoveAdditionalUser( await RequestHelpers.GetSessionId(_sessionManager, _userManager, HttpContext).ConfigureAwait(false), sessionId, - userId); + userId).ConfigureAwait(false); return NoContent(); } diff --git a/MediaBrowser.Controller/Session/IPodMessageBus.cs b/MediaBrowser.Controller/Session/IPodMessageBus.cs index 053299e6c2..4caa6c90c7 100644 --- a/MediaBrowser.Controller/Session/IPodMessageBus.cs +++ b/MediaBrowser.Controller/Session/IPodMessageBus.cs @@ -1,4 +1,5 @@ using System; +using System.Threading; using System.Threading.Tasks; namespace MediaBrowser.Controller.Session; @@ -15,11 +16,14 @@ public interface IPodMessageBus string PodId { get; } /// - /// Sends a message to one instance. Delivery is best effort and never throws. + /// Sends a message to one instance and reports how many listeners took it, so that a message + /// addressed to an instance that is no longer there is not mistaken for a delivered one. /// /// The instance to deliver to. /// The message. - void Publish(string targetPod, PodMessage message); + /// The cancellation token. + /// The number of instances the message reached. + Task PublishAsync(string targetPod, PodMessage message, CancellationToken cancellationToken = default); /// /// Registers a handler for the messages addressed to this instance. diff --git a/MediaBrowser.Controller/Session/ISessionDirectory.cs b/MediaBrowser.Controller/Session/ISessionDirectory.cs index f1752f9f63..5a35782571 100644 --- a/MediaBrowser.Controller/Session/ISessionDirectory.cs +++ b/MediaBrowser.Controller/Session/ISessionDirectory.cs @@ -11,20 +11,32 @@ namespace MediaBrowser.Controller.Session; public interface ISessionDirectory { /// - /// Publishes an entry and restarts its expiry. + /// Claims a session for the publishing instance and restarts its expiry. The claim is refused when + /// another instance holds the connection, so an instance that merely served a request for the session + /// cannot take ownership of it. /// /// The entry. + /// When the publishing instance's connection to the session was established, or zero when it holds none. /// The cancellation token. - /// A task representing the operation. - Task PublishAsync(SessionDirectoryEntry entry, CancellationToken cancellationToken = default); + /// true if the entry was written. + Task PublishAsync(SessionDirectoryEntry entry, long connectedUtcTicks, CancellationToken cancellationToken = default); /// - /// Removes an entry. + /// Removes an entry, but only while the calling instance still owns it. + /// + /// The session identifier. + /// The instance requesting the removal. + /// The cancellation token. + /// A task representing the operation. + Task RemoveAsync(string sessionId, string ownerPod, CancellationToken cancellationToken = default); + + /// + /// Gets one entry by session identifier. /// /// The session identifier. /// The cancellation token. - /// A task representing the operation. - Task RemoveAsync(string sessionId, CancellationToken cancellationToken = default); + /// The entry, or null when the session is in no instance's directory. + Task GetAsync(string sessionId, CancellationToken cancellationToken = default); /// /// Gets every entry that has not expired. diff --git a/MediaBrowser.Controller/Session/ISessionManager.cs b/MediaBrowser.Controller/Session/ISessionManager.cs index 58bbd72876..3dbd716bfd 100644 --- a/MediaBrowser.Controller/Session/ISessionManager.cs +++ b/MediaBrowser.Controller/Session/ISessionManager.cs @@ -80,7 +80,8 @@ namespace MediaBrowser.Controller.Session /// Used to report that a session controller has connected. /// /// The session. - void OnSessionControllerConnected(SessionInfo session); + /// A task representing the operation. + Task OnSessionControllerConnected(SessionInfo session); void UpdateDeviceName(string sessionId, string reportedDeviceName); @@ -241,7 +242,8 @@ namespace MediaBrowser.Controller.Session /// The controlling session identifier. /// The session identifier. /// The user identifier. - void AddAdditionalUser(string controllingSessionId, string sessionId, Guid userId); + /// A task representing the operation. + Task AddAdditionalUser(string controllingSessionId, string sessionId, Guid userId); /// /// Removes the additional user. @@ -249,7 +251,8 @@ namespace MediaBrowser.Controller.Session /// The controlling session identifier. /// The session identifier. /// The user identifier. - void RemoveAdditionalUser(string controllingSessionId, string sessionId, Guid userId); + /// A task representing the operation. + Task RemoveAdditionalUser(string controllingSessionId, string sessionId, Guid userId); /// /// Reports the now viewing item. diff --git a/MediaBrowser.Controller/Session/NullPodMessageBus.cs b/MediaBrowser.Controller/Session/NullPodMessageBus.cs index e34d6d9233..a0b175a53e 100644 --- a/MediaBrowser.Controller/Session/NullPodMessageBus.cs +++ b/MediaBrowser.Controller/Session/NullPodMessageBus.cs @@ -1,4 +1,5 @@ using System; +using System.Threading; using System.Threading.Tasks; namespace MediaBrowser.Controller.Session; @@ -17,9 +18,8 @@ public sealed class NullPodMessageBus : IPodMessageBus public string PodId => PodIdentity.Current; /// - public void Publish(string targetPod, PodMessage message) - { - } + public Task PublishAsync(string targetPod, PodMessage message, CancellationToken cancellationToken = default) + => Task.FromResult(0L); /// public void Subscribe(Func handler) diff --git a/MediaBrowser.Controller/Session/NullSessionDirectory.cs b/MediaBrowser.Controller/Session/NullSessionDirectory.cs index 3fcd3bed36..63ef843f9f 100644 --- a/MediaBrowser.Controller/Session/NullSessionDirectory.cs +++ b/MediaBrowser.Controller/Session/NullSessionDirectory.cs @@ -17,12 +17,16 @@ public sealed class NullSessionDirectory : ISessionDirectory public static NullSessionDirectory Instance { get; } = new NullSessionDirectory(); /// - public Task PublishAsync(SessionDirectoryEntry entry, CancellationToken cancellationToken = default) + public Task PublishAsync(SessionDirectoryEntry entry, long connectedUtcTicks, CancellationToken cancellationToken = default) + => Task.FromResult(false); + + /// + public Task RemoveAsync(string sessionId, string ownerPod, CancellationToken cancellationToken = default) => Task.CompletedTask; /// - public Task RemoveAsync(string sessionId, CancellationToken cancellationToken = default) - => Task.CompletedTask; + public Task GetAsync(string sessionId, CancellationToken cancellationToken = default) + => Task.FromResult(null); /// public Task> GetAllAsync(CancellationToken cancellationToken = default) diff --git a/MediaBrowser.Controller/Session/RoutedAdditionalUserChange.cs b/MediaBrowser.Controller/Session/RoutedAdditionalUserChange.cs new file mode 100644 index 0000000000..d494f52d05 --- /dev/null +++ b/MediaBrowser.Controller/Session/RoutedAdditionalUserChange.cs @@ -0,0 +1,30 @@ +using System; + +namespace MediaBrowser.Controller.Session; + +/// +/// An additional-user change for a session held by another instance, carried as a +/// . The calling instance has already authorized it. +/// +public sealed class RoutedAdditionalUserChange +{ + /// + /// The this payload travels under. + /// + public const string Kind = "AdditionalUserChange"; + + /// + /// Gets or sets the session the change applies to. + /// + public string SessionId { get; set; } = string.Empty; + + /// + /// Gets or sets the user to attach or detach. + /// + public Guid UserId { get; set; } + + /// + /// Gets or sets a value indicating whether the user is being attached rather than detached. + /// + public bool Add { get; set; } +} diff --git a/MediaBrowser.Controller/Session/SessionDirectoryEntry.cs b/MediaBrowser.Controller/Session/SessionDirectoryEntry.cs index 3f8aaa12bb..069325c8ca 100644 --- a/MediaBrowser.Controller/Session/SessionDirectoryEntry.cs +++ b/MediaBrowser.Controller/Session/SessionDirectoryEntry.cs @@ -12,6 +12,12 @@ public sealed class SessionDirectoryEntry /// public string OwnerPod { get; set; } = string.Empty; + /// + /// Gets or sets a value indicating whether the owner holds a live connection to the session. Only an + /// owner that does can be routed a remote-control message. + /// + public bool HoldsConnection { get; set; } + /// /// Gets or sets the session as its owner last rendered it. /// diff --git a/MediaBrowser.Controller/Session/SessionDirectoryOptions.cs b/MediaBrowser.Controller/Session/SessionDirectoryOptions.cs index 17ff304c2a..9baf25369c 100644 --- a/MediaBrowser.Controller/Session/SessionDirectoryOptions.cs +++ b/MediaBrowser.Controller/Session/SessionDirectoryOptions.cs @@ -20,4 +20,9 @@ public sealed class SessionDirectoryOptions /// Gets or sets how often in seconds an instance republishes the sessions it holds. /// public int RefreshIntervalSeconds { get; set; } = 20; + + /// + /// Gets or sets how long in seconds a single directory operation may take before it is abandoned. + /// + public int OperationTimeoutSeconds { get; set; } = 5; } diff --git a/README.md b/README.md index daffb95682..c82dba9330 100644 --- a/README.md +++ b/README.md @@ -118,6 +118,7 @@ Without a connection string the line reads `Transcode session store: NullTransco | `Jellyfin:TranscodeStore:SessionRetentionSeconds` | `300` | How long an unrenewed session record is kept so another pod can still take it over. | | `Jellyfin:SessionDirectory:EntryTtlSeconds` | `60` | How long a published session stays visible to the other pods without being refreshed. | | `Jellyfin:SessionDirectory:RefreshIntervalSeconds` | `20` | How often a pod republishes the sessions it holds. | +| `Jellyfin:SessionDirectory:OperationTimeoutSeconds` | `5` | How long a single session directory read or write may take before it is abandoned. | ### Redis connection string examples diff --git a/tests/Jellyfin.Server.Implementations.Tests/SessionManager/SessionManagerTests.cs b/tests/Jellyfin.Server.Implementations.Tests/SessionManager/SessionManagerTests.cs index d032539390..5edbfde690 100644 --- a/tests/Jellyfin.Server.Implementations.Tests/SessionManager/SessionManagerTests.cs +++ b/tests/Jellyfin.Server.Implementations.Tests/SessionManager/SessionManagerTests.cs @@ -180,7 +180,7 @@ public class SessionManagerTests var attackerSession = await LogSessionActivity(sessionManager, attacker); - Assert.Throws(() => sessionManager.AddAdditionalUser(attackerSession.Id, attackerSession.Id, victim.Id)); + await Assert.ThrowsAsync(() => sessionManager.AddAdditionalUser(attackerSession.Id, attackerSession.Id, victim.Id)); } [Fact] @@ -193,7 +193,7 @@ public class SessionManagerTests var adminSession = await LogSessionActivity(sessionManager, admin); - sessionManager.AddAdditionalUser(adminSession.Id, adminSession.Id, guest.Id); + await sessionManager.AddAdditionalUser(adminSession.Id, adminSession.Id, guest.Id); Assert.Contains(adminSession.AdditionalUsers, i => i.UserId.Equals(guest.Id)); } @@ -208,7 +208,7 @@ public class SessionManagerTests var victimSession = await LogSessionActivity(sessionManager, victim); var attackerSession = await LogSessionActivity(sessionManager, attacker); - Assert.Throws(() => sessionManager.RemoveAdditionalUser(attackerSession.Id, victimSession.Id, attacker.Id)); + await Assert.ThrowsAsync(() => sessionManager.RemoveAdditionalUser(attackerSession.Id, victimSession.Id, attacker.Id)); } [Fact] diff --git a/tests/Jellyfin.Server.Tests/HighAvailability/SessionDirectoryReplicaTests.cs b/tests/Jellyfin.Server.Tests/HighAvailability/SessionDirectoryReplicaTests.cs index 4d5f3f8464..05a67fa06a 100644 --- a/tests/Jellyfin.Server.Tests/HighAvailability/SessionDirectoryReplicaTests.cs +++ b/tests/Jellyfin.Server.Tests/HighAvailability/SessionDirectoryReplicaTests.cs @@ -1,4 +1,5 @@ using System; +using System.Linq; using System.Text.Json; using System.Threading; using System.Threading.Tasks; @@ -9,6 +10,7 @@ using Jellyfin.Database.Implementations.Locking; using Jellyfin.Database.Providers.PostgreSQL; using Jellyfin.Server.Implementations.Devices; using Jellyfin.Server.Tests.Migrations; +using MediaBrowser.Common.Extensions; using MediaBrowser.Controller; using MediaBrowser.Controller.Configuration; using MediaBrowser.Controller.Drawing; @@ -39,11 +41,18 @@ namespace Jellyfin.Server.Tests.HighAvailability; [Trait("Category", "RequiresDocker")] public sealed class SessionDirectoryReplicaTests : IAsyncLifetime { + private const string AppName = "Jellyfin Web"; + private const string AppVersion = "1.0.0"; + private const string DeviceName = "Living Room TV"; + private const string RemoteEndPoint = "127.0.0.1"; + private PostgreSqlTestServer _postgres = null!; private RedisTestServer _redis = null!; private NpgsqlDataSource _dataSource = null!; private IConnectionMultiplexer _connection = null!; + private ISessionDirectory _directory = null!; private User _user = null!; + private User _guest = null!; /// public async ValueTask InitializeAsync() @@ -51,6 +60,10 @@ public sealed class SessionDirectoryReplicaTests : IAsyncLifetime _postgres = await PostgreSqlTestServer.StartAsync(); _redis = await RedisTestServer.StartAsync(); _connection = await _redis.ConnectAsync(); + _directory = new RedisSessionDirectory( + _connection, + Options.Create(new SessionDirectoryOptions()), + NullLogger.Instance); var connectionString = await _postgres.CreateDatabaseAsync("session_directory", CancellationToken.None); _dataSource = new NpgsqlDataSourceBuilder(connectionString).Build(); @@ -61,7 +74,9 @@ public sealed class SessionDirectoryReplicaTests : IAsyncLifetime await context.Database.EnsureCreatedAsync(CancellationToken.None); _user = new User("replica-user", "provider", "provider"); + _guest = new User("replica-guest", "provider", "provider"); context.Users.Add(_user); + context.Users.Add(_guest); await context.SaveChangesAsync(CancellationToken.None); } } @@ -87,7 +102,7 @@ public sealed class SessionDirectoryReplicaTests : IAsyncLifetime await using var replicaA = CreateReplica("pod-a"); await using var replicaB = CreateReplica("pod-b"); - var session = await replicaA.LogSessionActivity("Jellyfin Web", "1.0.0", "device-listed", "Living Room TV", "127.0.0.1", _user); + var session = await Request(replicaA, "device-listed"); var listedByB = await replicaB.GetSessions(_user.Id, null, null, null, false, cancellationToken); var listedByA = await replicaA.GetSessions(_user.Id, null, null, null, false, cancellationToken); @@ -100,23 +115,45 @@ public sealed class SessionDirectoryReplicaTests : IAsyncLifetime } /// - /// Remote control and "send message to session" currently succeed and do nothing when the device is - /// connected to another replica; the message has to reach the connection wherever it is held. + /// The deployment has no sticky sessions, so one device's requests land on either replica while its + /// websocket stays on one of them. Ownership has to follow the connection rather than the last + /// request served, or the directory names the wrong replica, the session list doubles up and remote + /// control is delivered to a replica with nothing to deliver it to. /// /// A representing the asynchronous operation. [Fact] - public async Task MessageSentOnOneReplica_ReachesTheConnectionHeldByAnother() + public async Task RequestsAlternatingBetweenReplicas_KeepOwnershipWithTheConnection() { var cancellationToken = TestContext.Current.CancellationToken; - await using var replicaA = CreateReplica("pod-a"); - await using var replicaB = CreateReplica("pod-b"); + var options = new SessionDirectoryOptions { EntryTtlSeconds = 60, RefreshIntervalSeconds = 2 }; + await using var replicaA = CreateReplica("pod-a", options); + await using var replicaB = CreateReplica("pod-b", options); + // The device is first seen by the replica that will not hold its websocket. + await Request(replicaB, "device-roaming"); + + var session = await Request(replicaA, "device-roaming"); var controller = new RecordingSessionController(); - var session = await replicaA.LogSessionActivity("Jellyfin Web", "1.0.0", "device-controlled", "Living Room TV", "127.0.0.1", _user); session.AddController(controller); + await replicaA.OnSessionControllerConnected(session); - // Republish now that the session has a connection, the way the websocket handshake does. - await replicaA.LogSessionActivity("Jellyfin Web", "1.0.0", "device-controlled", "Living Room TV", "127.0.0.1", _user); + for (var i = 0; i < 6; i++) + { + await Request(replicaB, "device-roaming"); + await Request(replicaA, "device-roaming"); + await Task.Delay(300, cancellationToken); + + var entry = await _directory.GetAsync(session.Id, cancellationToken); + Assert.NotNull(entry); + Assert.Equal("pod-a", entry.OwnerPod); + Assert.True(entry.HoldsConnection); + } + + var listedByA = await replicaA.GetSessions(_user.Id, null, null, null, false, cancellationToken); + var listedByB = await replicaB.GetSessions(_user.Id, null, null, null, false, cancellationToken); + + Assert.Single(listedByA, i => string.Equals(i.Id, session.Id, StringComparison.Ordinal)); + Assert.Single(listedByB, i => string.Equals(i.Id, session.Id, StringComparison.Ordinal)); await replicaB.SendMessageCommand( string.Empty, @@ -129,6 +166,118 @@ public sealed class SessionDirectoryReplicaTests : IAsyncLifetime Assert.Contains("Dinner is ready", data, StringComparison.Ordinal); } + /// + /// Remote control and "send message to session" used to succeed and do nothing when the device is + /// connected to another replica; the message has to reach the connection wherever it is held. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task MessageSentOnOneReplica_ReachesTheConnectionHeldByAnother() + { + var cancellationToken = TestContext.Current.CancellationToken; + await using var replicaA = CreateReplica("pod-a"); + await using var replicaB = CreateReplica("pod-b"); + + var session = await Request(replicaA, "device-controlled"); + var controller = new RecordingSessionController(); + session.AddController(controller); + await replicaA.OnSessionControllerConnected(session); + + await replicaB.SendMessageCommand( + string.Empty, + session.Id, + new MessageCommand { Header = "Header", Text = "Dinner is ready" }, + cancellationToken); + + var (messageType, data) = await controller.WaitForMessageAsync(cancellationToken); + Assert.Equal(SessionMessageType.GeneralCommand, messageType); + Assert.Contains("Dinner is ready", data, StringComparison.Ordinal); + } + + /// + /// An entry outlives the replica that wrote it by up to its expiry, and a command routed into that + /// gap reaches nobody. Reporting it as delivered is the failure this directory exists to remove. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task MessageRoutedToADeadOwner_IsReportedAsUndelivered() + { + var cancellationToken = TestContext.Current.CancellationToken; + await using var replicaA = CreateReplica("pod-a"); + await using var replicaB = CreateReplica("pod-b"); + + var session = await Request(replicaA, "device-dead-owner"); + session.AddController(new RecordingSessionController()); + await replicaA.OnSessionControllerConnected(session); + + var entry = await _directory.GetAsync(session.Id, cancellationToken); + Assert.NotNull(entry); + + // A replica that is no longer listening, holding the entry until it expires. + entry.OwnerPod = "pod-gone"; + Assert.True(await _directory.PublishAsync(entry, DateTime.UtcNow.Ticks, cancellationToken)); + + await Assert.ThrowsAsync( + () => replicaB.SendMessageCommand( + string.Empty, + session.Id, + new MessageCommand { Header = "Header", Text = "Dinner is ready" }, + cancellationToken)); + } + + /// + /// Both replicas keep a copy of a session whose requests they have served, so the replica ending its + /// own copy must not erase the entry of the one still holding the connection. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task ReplicaEndingItsOwnCopy_LeavesTheOwnersEntryAlone() + { + var cancellationToken = TestContext.Current.CancellationToken; + await using var replicaA = CreateReplica("pod-a"); + await using var replicaB = CreateReplica("pod-b"); + + var session = await Request(replicaA, "device-shared-end"); + session.AddController(new RecordingSessionController()); + await replicaA.OnSessionControllerConnected(session); + + await Request(replicaB, "device-shared-end"); + await replicaB.ReportSessionEnded(session.Id); + + var entry = await _directory.GetAsync(session.Id, cancellationToken); + Assert.NotNull(entry); + Assert.Equal("pod-a", entry.OwnerPod); + + await replicaA.ReportSessionEnded(session.Id); + + Assert.Null(await _directory.GetAsync(session.Id, cancellationToken)); + } + + /// + /// The session list now shows sessions from every replica, so an action offered against one of them + /// has to reach it rather than fail as missing on the replica serving the request. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task AdditionalUserAddedOnOneReplica_ReachesTheOwner() + { + var cancellationToken = TestContext.Current.CancellationToken; + await using var replicaA = CreateReplica("pod-a"); + await using var replicaB = CreateReplica("pod-b"); + + var session = await Request(replicaA, "device-additional-user"); + session.AddController(new RecordingSessionController()); + await replicaA.OnSessionControllerConnected(session); + + await replicaB.AddAdditionalUser(string.Empty, session.Id, _guest.Id); + + await WaitUntil(() => session.AdditionalUsers.Any(i => i.UserId.Equals(_guest.Id)), cancellationToken); + + await replicaB.RemoveAdditionalUser(string.Empty, session.Id, _guest.Id); + + await WaitUntil(() => !session.AdditionalUsers.Any(i => i.UserId.Equals(_guest.Id)), cancellationToken); + } + /// /// A replica that dies stops refreshing its entries, and the sessions it held have to leave the /// directory rather than linger in every other replica's session list forever. @@ -143,7 +292,7 @@ public sealed class SessionDirectoryReplicaTests : IAsyncLifetime await using var replicaA = CreateReplica("pod-a", new SessionDirectoryOptions { EntryTtlSeconds = 1, RefreshIntervalSeconds = 3600 }); await using var replicaB = CreateReplica("pod-b"); - var session = await replicaA.LogSessionActivity("Jellyfin Web", "1.0.0", "device-expiring", "Living Room TV", "127.0.0.1", _user); + var session = await Request(replicaA, "device-expiring"); var listedWhileAlive = await replicaB.GetSessions(_user.Id, null, null, null, false, cancellationToken); Assert.Contains(listedWhileAlive, i => string.Equals(i.Id, session.Id, StringComparison.Ordinal)); @@ -166,12 +315,37 @@ public sealed class SessionDirectoryReplicaTests : IAsyncLifetime await using var replicaA = CreateReplica("pod-a", directory: NullSessionDirectory.Instance, bus: NullPodMessageBus.Instance); await using var replicaB = CreateReplica("pod-b", directory: NullSessionDirectory.Instance, bus: NullPodMessageBus.Instance); - var session = await replicaA.LogSessionActivity("Jellyfin Web", "1.0.0", "device-local", "Living Room TV", "127.0.0.1", _user); + var session = await Request(replicaA, "device-local"); var listedByB = await replicaB.GetSessions(_user.Id, null, null, null, false, cancellationToken); Assert.DoesNotContain(listedByB, i => string.Equals(i.Id, session.Id, StringComparison.Ordinal)); } + private static async Task WaitUntil(Func condition, CancellationToken cancellationToken) + { + var deadline = DateTime.UtcNow.AddSeconds(10); + while (!condition()) + { + Assert.True(DateTime.UtcNow < deadline, "The expected change never arrived."); + await Task.Delay(50, cancellationToken); + } + } + + private static JellyfinDbContext CreateContext(NpgsqlDataSource dataSource) + { + var optionsBuilder = new DbContextOptionsBuilder(); + var provider = new PostgreSqlDatabaseProvider(dataSource); + provider.Initialise(optionsBuilder, new DatabaseConfigurationOptions { DatabaseType = "PostgreSQL" }); + return new JellyfinDbContext( + optionsBuilder.Options, + NullLogger.Instance, + provider, + new NoLockBehavior(NullLogger.Instance)); + } + + private Task Request(SessionManager replica, string deviceId) + => replica.LogSessionActivity(AppName, AppVersion, deviceId, DeviceName, RemoteEndPoint, _user); + private SessionManager CreateReplica(string podId, SessionDirectoryOptions? options = null) { options ??= new SessionDirectoryOptions { EntryTtlSeconds = 60, RefreshIntervalSeconds = 3600 }; @@ -191,6 +365,7 @@ public sealed class SessionDirectoryReplicaTests : IAsyncLifetime { var userManager = new Mock(); userManager.Setup(i => i.GetUserById(_user.Id)).Returns(_user); + userManager.Setup(i => i.GetUserById(_guest.Id)).Returns(_guest); var appHost = new Mock(); appHost.SetupGet(i => i.SystemId).Returns("server-" + podId); @@ -231,18 +406,6 @@ public sealed class SessionDirectoryReplicaTests : IAsyncLifetime } } - private static JellyfinDbContext CreateContext(NpgsqlDataSource dataSource) - { - var optionsBuilder = new DbContextOptionsBuilder(); - var provider = new PostgreSqlDatabaseProvider(dataSource); - provider.Initialise(optionsBuilder, new DatabaseConfigurationOptions { DatabaseType = "PostgreSQL" }); - return new JellyfinDbContext( - optionsBuilder.Options, - NullLogger.Instance, - provider, - new NoLockBehavior(NullLogger.Instance)); - } - /// /// Hands every replica its own context over the one shared database, the way the pooled factory does /// in the server.