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.