diff --git a/.woodpecker/ci.yaml b/.woodpecker/ci.yaml
index 668be16db5..7e021cf725 100644
--- a/.woodpecker/ci.yaml
+++ b/.woodpecker/ci.yaml
@@ -65,7 +65,7 @@ steps:
- su postgres -c "$PGBIN/initdb -D /tmp/pgdata -A trust -U postgres"
- su postgres -c "$PGBIN/pg_ctl -D /tmp/pgdata -o \"-c listen_addresses=127.0.0.1 -k /tmp/pgrun\" -l /tmp/pg.log -w start"
- valkey-server --daemonize yes --bind 127.0.0.1 --port 6379 --save ''
- - valkey-cli ping
+ - for i in $(seq 30); do valkey-cli -h 127.0.0.1 ping && break; sleep 1; done
- dotnet build tests/Jellyfin.Server.Tests/Jellyfin.Server.Tests.csproj -c Release
- dotnet build tests/Jellyfin.Database.Tests.PostgreSQL/Jellyfin.Database.Tests.PostgreSQL.csproj -c Release
- dotnet test tests/Jellyfin.Server.Tests/Jellyfin.Server.Tests.csproj -c Release --no-build --verbosity minimal --filter "Category=RequiresDocker"
diff --git a/Emby.Server.Implementations/Session/RedisPodMessageBus.cs b/Emby.Server.Implementations/Session/RedisPodMessageBus.cs
new file mode 100644
index 0000000000..5b6a48063c
--- /dev/null
+++ b/Emby.Server.Implementations/Session/RedisPodMessageBus.cs
@@ -0,0 +1,181 @@
+using System;
+using System.Collections.Concurrent;
+using System.Text.Json;
+using System.Threading;
+using System.Threading.Tasks;
+using Jellyfin.Extensions.Json;
+using MediaBrowser.Controller.Session;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using StackExchange.Redis;
+
+namespace Emby.Server.Implementations.Session;
+
+///
+/// A Redis pub/sub . Every instance subscribes to a channel named after
+/// itself, which keeps addressed delivery working without the instances being routable to each other.
+/// A request is answered on the sender's own channel, so the sender learns what the receiver did with
+/// it rather than only that something was subscribed.
+///
+public sealed class RedisPodMessageBus : IPodMessageBus
+{
+ private const string ChannelPrefix = "jellyfin:pod:";
+
+ private static readonly JsonSerializerOptions _jsonOptions = JsonDefaults.Options;
+
+ private readonly ConcurrentDictionary> _pending = new(StringComparer.Ordinal);
+ private readonly ISubscriber _subscriber;
+ private readonly ILogger _logger;
+ private readonly TimeSpan _timeout;
+
+ private Func>? _handler;
+
+ ///
+ /// Initializes a new instance of the class.
+ ///
+ /// The Redis connection multiplexer.
+ /// The session directory configuration options.
+ /// The identity of this instance.
+ /// The logger.
+ public RedisPodMessageBus(
+ IConnectionMultiplexer redis,
+ IOptions options,
+ string podId,
+ ILogger logger)
+ {
+ ArgumentNullException.ThrowIfNull(redis);
+ ArgumentNullException.ThrowIfNull(options);
+ ArgumentException.ThrowIfNullOrEmpty(podId);
+
+ _subscriber = redis.GetSubscriber();
+ _logger = logger;
+ _timeout = TimeSpan.FromSeconds(Math.Max(1, options.Value.OperationTimeoutSeconds));
+ PodId = podId;
+
+ // A bus that cannot subscribe can only send, so every request it makes waits out the timeout and
+ // nothing routed here is ever answered. The caller degrades the pair to single-instance instead.
+ _subscriber.Subscribe(RedisChannel.Literal(ChannelPrefix + PodId), (_, value) => Dispatch(value));
+ }
+
+ ///
+ public string PodId { get; }
+
+ ///
+ public async Task RequestAsync(string targetPod, PodMessage message, CancellationToken cancellationToken = default)
+ {
+ ArgumentException.ThrowIfNullOrEmpty(targetPod);
+ ArgumentNullException.ThrowIfNull(message);
+
+ message.OriginPod = PodId;
+ message.CorrelationId = Guid.NewGuid().ToString("N");
+
+ var acknowledged = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ _pending[message.CorrelationId] = acknowledged;
+
+ // Publish and acknowledgement share one deadline, so a request is bounded by the timeout rather
+ // than by twice it.
+ using var deadline = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+ deadline.CancelAfter(_timeout);
+
+ try
+ {
+ var subscribers = await _subscriber.PublishAsync(
+ RedisChannel.Literal(ChannelPrefix + targetPod),
+ JsonSerializer.Serialize(message, _jsonOptions)).WaitAsync(deadline.Token).ConfigureAwait(false);
+
+ if (subscribers == 0)
+ {
+ return false;
+ }
+
+ return await acknowledged.Task.WaitAsync(deadline.Token).ConfigureAwait(false);
+ }
+ catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
+ {
+ _logger.LogWarning("Instance {TargetPod} did not acknowledge a {Kind} message within {Timeout}.", targetPod, message.Kind, _timeout);
+ return false;
+ }
+ catch (Exception ex) when (ex is not OperationCanceledException)
+ {
+ _logger.LogWarning(ex, "Failed to send a {Kind} message to {TargetPod}.", message.Kind, targetPod);
+ return false;
+ }
+ finally
+ {
+ _pending.TryRemove(message.CorrelationId, out _);
+ }
+ }
+
+ ///
+ public void Subscribe(Func> handler)
+ {
+ ArgumentNullException.ThrowIfNull(handler);
+
+ _handler = handler;
+ }
+
+ private async void Dispatch(RedisValue value)
+ {
+ PodMessage? message = null;
+
+ try
+ {
+ message = JsonSerializer.Deserialize(value.ToString(), _jsonOptions);
+ if (message is null)
+ {
+ return;
+ }
+
+ if (string.Equals(message.Kind, PodMessage.AckKind, StringComparison.Ordinal))
+ {
+ if (_pending.TryRemove(message.CorrelationId, out var acknowledged))
+ {
+ acknowledged.TrySetResult(message.Handled);
+ }
+
+ return;
+ }
+
+ var handler = _handler;
+ var handled = handler is not null && await handler(message).ConfigureAwait(false);
+
+ await AcknowledgeAsync(message, handled).ConfigureAwait(false);
+ }
+ catch (Exception ex)
+ {
+ _logger.LogWarning(ex, "Failed to handle a message routed to this instance.");
+
+ if (message is not null && !string.Equals(message.Kind, PodMessage.AckKind, StringComparison.Ordinal))
+ {
+ await AcknowledgeAsync(message, false).ConfigureAwait(false);
+ }
+ }
+ }
+
+ private async Task AcknowledgeAsync(PodMessage message, bool handled)
+ {
+ if (string.IsNullOrEmpty(message.CorrelationId) || string.IsNullOrEmpty(message.OriginPod))
+ {
+ return;
+ }
+
+ var ack = new PodMessage
+ {
+ Kind = PodMessage.AckKind,
+ OriginPod = PodId,
+ CorrelationId = message.CorrelationId,
+ Handled = handled
+ };
+
+ try
+ {
+ await _subscriber.PublishAsync(
+ RedisChannel.Literal(ChannelPrefix + message.OriginPod),
+ JsonSerializer.Serialize(ack, _jsonOptions)).WaitAsync(_timeout).ConfigureAwait(false);
+ }
+ catch (Exception ex)
+ {
+ _logger.LogWarning(ex, "Failed to acknowledge a {Kind} message to {OriginPod}.", message.Kind, message.OriginPod);
+ }
+ }
+}
diff --git a/Emby.Server.Implementations/Session/RedisSessionDirectory.cs b/Emby.Server.Implementations/Session/RedisSessionDirectory.cs
new file mode 100644
index 0000000000..b6953d1a49
--- /dev/null
+++ b/Emby.Server.Implementations/Session/RedisSessionDirectory.cs
@@ -0,0 +1,251 @@
+using System;
+using System.Collections.Generic;
+using System.Globalization;
+using System.Linq;
+using System.Text.Json;
+using System.Threading;
+using System.Threading.Tasks;
+using Jellyfin.Extensions.Json;
+using MediaBrowser.Controller.Session;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using StackExchange.Redis;
+
+namespace Emby.Server.Implementations.Session;
+
+///
+/// A Redis-backed . A session is owned by the instance holding its
+/// connection: ownership is claimed through a check-and-set against a connection epoch handed out by
+/// Redis, so an instance that only served a request for the session cannot take it from the instance the
+/// device is actually connected to, and no instance's clock is compared against another's. 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:";
+ private const string EpochKeyPrefix = "jellyfin:sessionepoch:";
+
+ ///
+ /// Lua script for an atomic ownership claim. The owner key holds epoch|pod, where the epoch is
+ /// zero for an instance that holds no connection. Another instance takes ownership only by presenting
+ /// a connection epoch newer than the recorded one, so neither the instances serving the session's
+ /// requests nor two instances without a connection can take it from the one that has it.
+ ///
+ private const string ClaimScript = @"
+redis.call('PEXPIRE', KEYS[3], ARGV[5])
+local current = redis.call('GET', KEYS[1])
+if current then
+ local separator = string.find(current, '|', 1, true)
+ local connected = tonumber(string.sub(current, 1, separator - 1))
+ local owner = string.sub(current, separator + 1)
+ local claiming = tonumber(ARGV[2])
+ if owner ~= ARGV[1] and (claiming == 0 or claiming <= connected) then
+ return 0
+ end
+end
+redis.call('SET', KEYS[1], ARGV[2] .. '|' .. ARGV[1], '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, separator + 1) ~= ARGV[1] then return 0 end
+redis.call('DEL', KEYS[1], KEYS[2])
+return 1";
+
+ ///
+ /// Lua script allocating the next connection epoch. The counter outlives the entries that reference
+ /// it, so it never restarts underneath a recorded epoch.
+ ///
+ private const string EpochScript = @"
+local epoch = redis.call('INCR', KEYS[1])
+redis.call('PEXPIRE', KEYS[1], ARGV[1])
+return epoch";
+
+ private static readonly JsonSerializerOptions _jsonOptions = JsonDefaults.Options;
+
+ private readonly IConnectionMultiplexer _redis;
+ private readonly IDatabase _db;
+ private readonly SessionDirectoryOptions _options;
+ private readonly ILogger _logger;
+
+ ///
+ /// Initializes a new instance of the class.
+ ///
+ /// The Redis connection multiplexer.
+ /// The session directory configuration options.
+ /// The logger.
+ public RedisSessionDirectory(
+ IConnectionMultiplexer redis,
+ IOptions options,
+ ILogger logger)
+ {
+ ArgumentNullException.ThrowIfNull(redis);
+ ArgumentNullException.ThrowIfNull(options);
+
+ _redis = redis;
+ _db = redis.GetDatabase();
+ _options = options.Value;
+ _logger = logger;
+ }
+
+ private long EntryTtlMs => Math.Max(1, _options.EntryTtlSeconds) * 1000L;
+
+ // Outlives the entries that name an epoch, so a live entry never outlives the counter it came from.
+ private long EpochTtlMs => EntryTtlMs * 4;
+
+ private TimeSpan OperationTimeout => TimeSpan.FromSeconds(Math.Max(1, _options.OperationTimeoutSeconds));
+
+ ///
+ public async Task AllocateConnectionEpochAsync(string sessionId, CancellationToken cancellationToken = default)
+ {
+ ArgumentException.ThrowIfNullOrEmpty(sessionId);
+
+ var epoch = (long?)await _db.ScriptEvaluateAsync(
+ EpochScript,
+ keys: new RedisKey[] { EpochKeyPrefix + sessionId },
+ values: new RedisValue[] { EpochTtlMs }).WaitAsync(OperationTimeout, cancellationToken).ConfigureAwait(false);
+
+ return epoch ?? 0;
+ }
+
+ ///
+ public async Task PublishAsync(SessionDirectoryEntry entry, long connectionEpoch, CancellationToken cancellationToken = default)
+ {
+ ArgumentNullException.ThrowIfNull(entry);
+
+ var sessionId = entry.Session?.Id;
+ if (string.IsNullOrEmpty(sessionId))
+ {
+ return false;
+ }
+
+ try
+ {
+ var claimed = (long?)await _db.ScriptEvaluateAsync(
+ ClaimScript,
+ keys: new RedisKey[] { OwnerKeyPrefix + sessionId, KeyPrefix + sessionId, EpochKeyPrefix + sessionId },
+ values: new RedisValue[]
+ {
+ entry.OwnerPod,
+ connectionEpoch.ToString(CultureInfo.InvariantCulture),
+ JsonSerializer.Serialize(entry, _jsonOptions),
+ EntryTtlMs,
+ EpochTtlMs
+ }).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, string ownerPod, CancellationToken cancellationToken = default)
+ {
+ try
+ {
+ await _db.ScriptEvaluateAsync(
+ ReleaseScript,
+ keys: new RedisKey[] { OwnerKeyPrefix + sessionId, KeyPrefix + sessionId },
+ values: new RedisValue[] { ownerPod }).WaitAsync(OperationTimeout, cancellationToken).ConfigureAwait(false);
+ }
+ catch (Exception ex)
+ {
+ _logger.LogWarning(ex, "Failed to remove session {SessionId}; it expires on its own.", sessionId);
+ }
+ }
+
+ ///
+ public async Task GetAsync(string sessionId, CancellationToken cancellationToken = default)
+ {
+ // A store that cannot be read says nothing about where the session is, so the failure is raised
+ // rather than reported as "no such session", which would be acted on as a local-only session.
+ var raw = await _db.StringGetAsync(KeyPrefix + sessionId).WaitAsync(OperationTimeout, cancellationToken).ConfigureAwait(false);
+
+ return raw.HasValue ? Deserialize(raw) : null;
+ }
+
+ ///
+ public async Task> GetAllAsync(CancellationToken cancellationToken = default)
+ {
+ var entries = new List();
+
+ foreach (var server in _redis.GetServers())
+ {
+ if (!server.IsConnected)
+ {
+ continue;
+ }
+
+ var keys = new List();
+
+ try
+ {
+ await foreach (var key in server.KeysAsync(database: _db.Database, pattern: KeyPrefix + "*", pageSize: 1000).WithCancellation(cancellationToken).ConfigureAwait(false))
+ {
+ keys.Add(key);
+ }
+ }
+ catch (Exception ex) when (ex is not OperationCanceledException)
+ {
+ // Degrade to the sessions that could be read rather than failing the request outright.
+ _logger.LogWarning(ex, "Failed to list the session directory on {Server}; its sessions are not reported.", server.EndPoint);
+ continue;
+ }
+
+ foreach (var raw in await Task.WhenAll(keys.Select(key => ReadAsync(key, cancellationToken))).ConfigureAwait(false))
+ {
+ if (!raw.HasValue)
+ {
+ continue;
+ }
+
+ var entry = Deserialize(raw);
+ if (entry?.Session is not null)
+ {
+ entries.Add(entry);
+ }
+ }
+ }
+
+ return entries;
+ }
+
+ // One unreadable key must not discard the entries that did load.
+ private async Task ReadAsync(RedisKey key, CancellationToken cancellationToken)
+ {
+ try
+ {
+ return await _db.StringGetAsync(key).WaitAsync(OperationTimeout, cancellationToken).ConfigureAwait(false);
+ }
+ catch (Exception ex) when (ex is not OperationCanceledException)
+ {
+ _logger.LogWarning(ex, "Failed to read a session directory entry.");
+ return RedisValue.Null;
+ }
+ }
+
+ 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
new file mode 100644
index 0000000000..278de9c08d
--- /dev/null
+++ b/Emby.Server.Implementations/Session/RemoteSessionController.cs
@@ -0,0 +1,78 @@
+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;
+
+///
+/// Stands in for the websocket of a session another instance holds: messages are forwarded to that
+/// instance, which writes them to the connection it owns and reports back whether it did.
+///
+public sealed class RemoteSessionController : ISessionController
+{
+ private readonly IPodMessageBus _bus;
+ private readonly ILogger _logger;
+ private readonly string _ownerPod;
+ private readonly string _sessionId;
+
+ ///
+ /// 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 controllable.
+ /// Whether the owner reported that it holds the session's connection.
+ public RemoteSessionController(IPodMessageBus bus, ILogger logger, string ownerPod, string sessionId, bool supportsMediaControl, bool holdsConnection)
+ {
+ _bus = bus;
+ _logger = logger;
+ _ownerPod = ownerPod;
+ _sessionId = sessionId;
+ SupportsMediaControl = supportsMediaControl;
+ IsSessionActive = holdsConnection;
+ }
+
+ ///
+ public bool IsSessionActive { get; }
+
+ ///
+ public bool SupportsMediaControl { get; }
+
+ ///
+ public async Task SendMessage(SessionMessageType name, Guid messageId, T data, CancellationToken cancellationToken)
+ {
+ var routed = new RoutedSessionMessage
+ {
+ SessionId = _sessionId,
+ MessageType = name,
+ MessageId = messageId,
+ Data = JsonSerializer.Serialize(data, JsonDefaults.Options)
+ };
+
+ var delivered = await _bus.RequestAsync(
+ _ownerPod,
+ new PodMessage
+ {
+ Kind = RoutedSessionMessage.Kind,
+ Payload = JsonSerializer.Serialize(routed, JsonDefaults.Options)
+ },
+ cancellationToken).ConfigureAwait(false);
+
+ if (!delivered)
+ {
+ _logger.LogWarning("Instance {OwnerPod} did not write the {MessageType} message for session {SessionId} to a connection.", _ownerPod, name, _sessionId);
+
+ throw new ResourceNotFoundException(
+ string.Format(CultureInfo.InvariantCulture, "No instance holds a connection to session {0}.", _sessionId));
+ }
+ }
+}
diff --git a/Emby.Server.Implementations/Session/SessionManager.cs b/Emby.Server.Implementations/Session/SessionManager.cs
index 13bf42f437..0b5ac43d22 100644
--- a/Emby.Server.Implementations/Session/SessionManager.cs
+++ b/Emby.Server.Implementations/Session/SessionManager.cs
@@ -5,6 +5,7 @@ using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Globalization;
using System.Linq;
+using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using Jellyfin.Data;
@@ -15,6 +16,7 @@ using Jellyfin.Database.Implementations.Entities;
using Jellyfin.Database.Implementations.Entities.Security;
using Jellyfin.Database.Implementations.Enums;
using Jellyfin.Extensions;
+using Jellyfin.Extensions.Json;
using MediaBrowser.Common.Events;
using MediaBrowser.Common.Extensions;
using MediaBrowser.Controller;
@@ -39,6 +41,7 @@ using MediaBrowser.Model.SyncPlay;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
using Episode = MediaBrowser.Controller.Entities.TV.Episode;
namespace Emby.Server.Implementations.Session
@@ -60,15 +63,25 @@ namespace Emby.Server.Implementations.Session
private readonly IMediaSourceManager _mediaSourceManager;
private readonly IServerApplicationHost _appHost;
private readonly IDeviceManager _deviceManager;
+ private readonly ISessionDirectory _sessionDirectory;
+ private readonly IPodMessageBus _podMessageBus;
+ private readonly SessionDirectoryOptions _sessionDirectoryOptions;
+ private readonly bool _directoryEnabled;
private readonly CancellationTokenRegistration _shutdownCallback;
private readonly ConcurrentDictionary _activeConnections
= new(StringComparer.OrdinalIgnoreCase);
+ private readonly ConcurrentDictionary _connectionEpochs = new(StringComparer.Ordinal);
+ private readonly ConcurrentDictionary _lastDirectoryPublish = new(StringComparer.Ordinal);
+ private readonly ConcurrentDictionary _directoryOwned = new(StringComparer.Ordinal);
+
private readonly ConcurrentDictionary> _activeLiveStreamSessions
= new(StringComparer.OrdinalIgnoreCase);
private Timer _idleTimer;
private Timer _inactiveTimer;
+ private Timer _directoryTimer;
+ private int _refreshingDirectory;
private DtoOptions _itemInfoDtoOptions;
private bool _disposed;
@@ -89,6 +102,9 @@ namespace Emby.Server.Implementations.Session
/// Instance of interface.
/// Instance of interface.
/// Instance of interface.
+ /// Instance of interface.
+ /// Instance of interface.
+ /// The session directory options.
public SessionManager(
ILogger logger,
IEventManager eventManager,
@@ -102,7 +118,10 @@ namespace Emby.Server.Implementations.Session
IServerApplicationHost appHost,
IDeviceManager deviceManager,
IMediaSourceManager mediaSourceManager,
- IHostApplicationLifetime hostApplicationLifetime)
+ IHostApplicationLifetime hostApplicationLifetime,
+ ISessionDirectory sessionDirectory,
+ IPodMessageBus podMessageBus,
+ IOptions sessionDirectoryOptions)
{
_logger = logger;
_eventManager = eventManager;
@@ -116,9 +135,21 @@ namespace Emby.Server.Implementations.Session
_appHost = appHost;
_deviceManager = deviceManager;
_mediaSourceManager = mediaSourceManager;
+ _sessionDirectory = sessionDirectory;
+ _podMessageBus = podMessageBus;
+ _sessionDirectoryOptions = sessionDirectoryOptions.Value;
_shutdownCallback = hostApplicationLifetime.ApplicationStopping.Register(OnApplicationStopping);
_deviceManager.DeviceOptionsUpdated += OnDeviceManagerDeviceOptionsUpdated;
+
+ _directoryEnabled = _sessionDirectory is not NullSessionDirectory;
+
+ if (_directoryEnabled)
+ {
+ _podMessageBus.Subscribe(OnPodMessage);
+ var interval = TimeSpan.FromSeconds(Math.Max(1, _sessionDirectoryOptions.RefreshIntervalSeconds));
+ _directoryTimer = new Timer(RefreshSessionDirectory, null, interval, interval);
+ }
}
///
@@ -218,6 +249,8 @@ namespace Emby.Server.Implementations.Session
_eventManager.Publish(new SessionEndedEventArgs(info));
+ await RemoveFromDirectoryAsync(info).ConfigureAwait(false);
+
await info.DisposeAsync().ConfigureAwait(false);
}
@@ -288,11 +321,13 @@ namespace Emby.Server.Implementations.Session
});
}
+ QueueDirectoryPublish(session);
+
return session;
}
///
- public void OnSessionControllerConnected(SessionInfo session)
+ public async Task OnSessionControllerConnected(SessionInfo session)
{
EventHelper.QueueEventIfNotNull(
SessionControllerConnected,
@@ -302,6 +337,382 @@ namespace Emby.Server.Implementations.Session
SessionInfo = session
},
_logger);
+
+ // 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);
+ }
+ }
+
+ // A playback transition changes what every instance's session list shows, so it is not left to
+ // the throttle.
+ private Task PublishToDirectoryNowAsync(SessionInfo session)
+ {
+ if (!_directoryEnabled || string.IsNullOrEmpty(session.Id))
+ {
+ return Task.CompletedTask;
+ }
+
+ _lastDirectoryPublish[session.Id] = Environment.TickCount64;
+
+ return PublishToDirectoryAsync(session);
+ }
+
+ private async Task PublishToDirectoryAsync(SessionInfo session)
+ {
+ if (!_directoryEnabled || string.IsNullOrEmpty(session.Id))
+ {
+ return;
+ }
+
+ try
+ {
+ var connectionEpoch = await GetConnectionEpochAsync(session).ConfigureAwait(false);
+
+ _directoryOwned[session.Id] = await _sessionDirectory.PublishAsync(
+ new SessionDirectoryEntry
+ {
+ OwnerPod = _podMessageBus.PodId,
+ HoldsConnection = connectionEpoch > 0,
+ Session = ToSessionInfoDto(session)
+ },
+ connectionEpoch).ConfigureAwait(false);
+ }
+ catch (Exception ex)
+ {
+ _logger.LogDebug(ex, "Error publishing session {Session} to the directory.", session.Id);
+ }
+ }
+
+ // Ownership follows the connection, not the last request served: an instance without a live
+ // controller claims with epoch zero, which never displaces an entry another instance holds. The
+ // epoch is handed out by the shared store, so the epochs of two instances are ordered by one
+ // clock rather than by whichever machine's wall clock wrote them.
+ private async Task GetConnectionEpochAsync(SessionInfo session)
+ {
+ if (!session.SessionControllers.Any(i => i.IsSessionActive))
+ {
+ _connectionEpochs.TryRemove(session.Id, out _);
+ return 0;
+ }
+
+ if (_connectionEpochs.TryGetValue(session.Id, out var epoch))
+ {
+ return epoch;
+ }
+
+ var allocated = await _sessionDirectory.AllocateConnectionEpochAsync(session.Id).ConfigureAwait(false);
+
+ return _connectionEpochs.GetOrAdd(session.Id, allocated);
+ }
+
+ private async ValueTask RemoveFromDirectoryAsync(SessionInfo session)
+ {
+ _connectionEpochs.TryRemove(session.Id, out _);
+ _lastDirectoryPublish.TryRemove(session.Id, out _);
+ _directoryOwned.TryRemove(session.Id, out _);
+
+ if (!_directoryEnabled || string.IsNullOrEmpty(session.Id))
+ {
+ return;
+ }
+
+ await _sessionDirectory.RemoveAsync(session.Id, _podMessageBus.PodId).ConfigureAwait(false);
+ }
+
+ private void RefreshSessionDirectory(object state)
+ {
+ if (Interlocked.CompareExchange(ref _refreshingDirectory, 1, 0) == 0)
+ {
+ _ = RefreshSessionDirectoryAsync();
+ }
+ }
+
+ // Only the entries this instance can hold are refreshed: republishing a copy of a session another
+ // instance owns just has the claim refused.
+ private async Task RefreshSessionDirectoryAsync()
+ {
+ try
+ {
+ foreach (var session in _activeConnections.Values)
+ {
+ if (session.SessionControllers.Any(i => i.IsSessionActive)
+ || (_directoryOwned.TryGetValue(session.Id, out var owned) && owned))
+ {
+ await PublishToDirectoryAsync(session).ConfigureAwait(false);
+ }
+ }
+ }
+ catch (Exception ex)
+ {
+ _logger.LogDebug(ex, "Error refreshing the session directory.");
+ }
+ finally
+ {
+ Interlocked.Exchange(ref _refreshingDirectory, 0);
+ }
+ }
+
+ private async Task> GetRemoteEntriesAsync(CancellationToken cancellationToken)
+ {
+ if (!_directoryEnabled)
+ {
+ return Array.Empty();
+ }
+
+ var entries = await _sessionDirectory.GetAllAsync(cancellationToken).ConfigureAwait(false);
+
+ return entries
+ .Where(entry => entry.Session is not null
+ && !string.Equals(entry.OwnerPod, _podMessageBus.PodId, StringComparison.Ordinal))
+ .ToList();
+ }
+
+ private async Task GetRemoteSession(string sessionId)
+ {
+ if (!_directoryEnabled)
+ {
+ return null;
+ }
+
+ var entry = await _sessionDirectory.GetAsync(sessionId).ConfigureAwait(false);
+
+ if (entry?.Session is null
+ || string.Equals(entry.OwnerPod, _podMessageBus.PodId, StringComparison.Ordinal))
+ {
+ return null;
+ }
+
+ var dto = entry.Session;
+ var session = new SessionInfo(this, _logger)
+ {
+ Id = dto.Id,
+ UserId = dto.UserId,
+ UserName = dto.UserName,
+ Client = dto.Client,
+ DeviceId = dto.DeviceId,
+ DeviceName = dto.DeviceName,
+ DeviceType = dto.DeviceType,
+ ApplicationVersion = dto.ApplicationVersion,
+ RemoteEndPoint = dto.RemoteEndPoint,
+ LastActivityDate = dto.LastActivityDate,
+ ServerId = dto.ServerId,
+ AdditionalUsers = dto.AdditionalUsers ?? [],
+ Capabilities = dto.Capabilities?.ToClientCapabilities()
+ };
+
+ session.AddController(new RemoteSessionController(_podMessageBus, _logger, entry.OwnerPod, dto.Id, dto.SupportsMediaControl, entry.HoldsConnection));
+
+ return session;
+ }
+
+ // What is returned here becomes the sender's answer, so a message this instance could not carry
+ // out is reported as undelivered instead of being counted by the sender as having arrived.
+ private Task OnPodMessage(PodMessage message)
+ {
+ switch (message.Kind)
+ {
+ case RoutedSessionMessage.Kind:
+ return OnRoutedSessionMessage(message);
+ case RoutedAdditionalUserChange.Kind:
+ return Task.FromResult(OnRoutedAdditionalUserChange(message));
+ case RoutedNowViewingItem.Kind:
+ return Task.FromResult(OnRoutedNowViewingItem(message));
+ case RoutedCapabilities.Kind:
+ return Task.FromResult(OnRoutedCapabilities(message));
+ case RoutedPlaybackReport.StartKind:
+ case RoutedPlaybackReport.ProgressKind:
+ case RoutedPlaybackReport.StoppedKind:
+ return OnRoutedPlaybackReport(message);
+ default:
+ return Task.FromResult(false);
+ }
+ }
+
+ private async Task OnRoutedSessionMessage(PodMessage message)
+ {
+ var routed = JsonSerializer.Deserialize(message.Payload, JsonDefaults.Options);
+ if (routed is null)
+ {
+ return false;
+ }
+
+ var session = Sessions.FirstOrDefault(i => string.Equals(i.Id, routed.SessionId, StringComparison.Ordinal));
+ var controllers = session?.SessionControllers.Where(i => i.IsSessionActive).ToList();
+
+ if (controllers is null || controllers.Count == 0)
+ {
+ _logger.LogWarning(
+ "A {MessageType} message for session {Session} was routed to this instance, which no longer holds its connection.",
+ routed.MessageType,
+ routed.SessionId);
+ return false;
+ }
+
+ using var data = JsonDocument.Parse(routed.Data);
+ foreach (var controller in controllers)
+ {
+ await controller.SendMessage(routed.MessageType, routed.MessageId, data.RootElement, CancellationToken.None).ConfigureAwait(false);
+ }
+
+ return true;
+ }
+
+ private bool 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 false;
+ }
+
+ if (routed.Add)
+ {
+ AttachAdditionalUser(session, routed.UserId, _userManager.GetUserById(routed.UserId)?.Username);
+ }
+ else
+ {
+ DetachAdditionalUser(session, routed.UserId);
+ }
+
+ QueueDirectoryPublish(session);
+
+ return true;
+ }
+
+ private bool OnRoutedNowViewingItem(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 false;
+ }
+
+ SetNowViewingItem(session, routed.ItemId);
+
+ return true;
+ }
+
+ private bool OnRoutedCapabilities(PodMessage message)
+ {
+ var routed = JsonSerializer.Deserialize(message.Payload, JsonDefaults.Options);
+ var session = routed?.Capabilities is null
+ ? null
+ : Sessions.FirstOrDefault(i => string.Equals(i.Id, routed.SessionId, StringComparison.Ordinal));
+
+ if (session is null)
+ {
+ return false;
+ }
+
+ ReportCapabilities(session, routed.Capabilities, true);
+
+ return true;
+ }
+
+ private async Task OnRoutedPlaybackReport(PodMessage message)
+ {
+ try
+ {
+ switch (message.Kind)
+ {
+ case RoutedPlaybackReport.StartKind:
+ await OnPlaybackStartCore(Deserialize(message)).ConfigureAwait(false);
+ return true;
+ case RoutedPlaybackReport.ProgressKind:
+ return await OnPlaybackProgressCore(Deserialize(message), false).ConfigureAwait(false);
+ default:
+ await OnPlaybackStoppedCore(Deserialize(message)).ConfigureAwait(false);
+ return true;
+ }
+ }
+ catch (Exception ex)
+ {
+ _logger.LogWarning(ex, "A {Kind} report routed to this instance could not be applied.", message.Kind);
+ return false;
+ }
+ }
+
+ private static T Deserialize(PodMessage message)
+ => JsonSerializer.Deserialize(message.Payload, JsonDefaults.Options);
+
+ // A sessionId-addressed mutation belongs to the instance whose copy of the session everyone else
+ // is shown. An undeliverable route falls back to handling it here, which is what a deployment
+ // without a directory does anyway.
+ private async Task TryRouteToOwnerAsync(string sessionId, string kind, object payload, CancellationToken cancellationToken)
+ {
+ if (!_directoryEnabled || string.IsNullOrEmpty(sessionId))
+ {
+ return false;
+ }
+
+ try
+ {
+ var entry = await _sessionDirectory.GetAsync(sessionId, cancellationToken).ConfigureAwait(false);
+
+ if (entry is null || string.Equals(entry.OwnerPod, _podMessageBus.PodId, StringComparison.Ordinal))
+ {
+ return false;
+ }
+
+ var routed = await _podMessageBus.RequestAsync(
+ entry.OwnerPod,
+ new PodMessage
+ {
+ Kind = kind,
+ Payload = JsonSerializer.Serialize(payload, JsonDefaults.Options)
+ },
+ cancellationToken).ConfigureAwait(false);
+
+ if (!routed)
+ {
+ // Delivery is at-least-once: an owner that applied the report and then failed to
+ // acknowledge it has it applied here as well, so a stop can be applied twice. A
+ // duplicate re-saves the same position; dropping the report loses it outright.
+ _logger.LogWarning(
+ "Instance {OwnerPod} did not acknowledge the {Kind} report for session {Session}; it is applied here as well, which may repeat one the owner already applied.",
+ entry.OwnerPod,
+ kind,
+ sessionId);
+ }
+
+ return routed;
+ }
+ catch (Exception ex) when (ex is not OperationCanceledException)
+ {
+ _logger.LogWarning(ex, "Could not reach the owner of session {Session}; the {Kind} report is applied here instead.", sessionId, kind);
+ return false;
+ }
}
///
@@ -604,6 +1015,29 @@ namespace Emby.Server.Implementations.Session
return users;
}
+ // The sweeps stop playback and save user data, so they act only on sessions this instance owns.
+ // A copy left on a non-owner stops checking in as soon as reports route away, and stopping it would
+ // end the owner's live playback and save the stale position the copy last saw.
+ private async Task OwnsSessionAsync(SessionInfo session)
+ {
+ if (!_directoryEnabled)
+ {
+ return true;
+ }
+
+ try
+ {
+ var entry = await _sessionDirectory.GetAsync(session.Id).ConfigureAwait(false);
+
+ return entry is null || string.Equals(entry.OwnerPod, _podMessageBus.PodId, StringComparison.Ordinal);
+ }
+ catch (Exception ex)
+ {
+ _logger.LogDebug(ex, "Could not read the owner of session {Session}; it is left alone.", session.Id);
+ return false;
+ }
+ }
+
private void StartCheckTimers()
{
_idleTimer ??= new Timer(CheckForIdlePlayback, null, TimeSpan.FromMinutes(5), TimeSpan.FromMinutes(5));
@@ -648,6 +1082,11 @@ namespace Emby.Server.Implementations.Session
foreach (var session in idle)
{
+ if (!await OwnsSessionAsync(session).ConfigureAwait(false))
+ {
+ continue;
+ }
+
_logger.LogDebug("Session {0} has gone idle while playing", session.Id);
try
@@ -682,6 +1121,11 @@ namespace Emby.Server.Implementations.Session
foreach (var session in inactiveSessions)
{
+ if (!await OwnsSessionAsync(session).ConfigureAwait(false))
+ {
+ continue;
+ }
+
_logger.LogDebug("Session {Session} has been inactive for {InactiveTime} minutes. Stopping it.", session.Id, _config.Configuration.InactiveSessionThreshold);
try
@@ -768,6 +1212,16 @@ namespace Emby.Server.Implementations.Session
ArgumentNullException.ThrowIfNull(info);
+ if (await TryRouteToOwnerAsync(info.SessionId, RoutedPlaybackReport.StartKind, info, CancellationToken.None).ConfigureAwait(false))
+ {
+ return;
+ }
+
+ await OnPlaybackStartCore(info).ConfigureAwait(false);
+ }
+
+ private async Task OnPlaybackStartCore(PlaybackStartInfo info)
+ {
var session = GetSession(info.SessionId);
var libraryItem = info.ItemId.IsEmpty()
@@ -834,6 +1288,8 @@ namespace Emby.Server.Implementations.Session
_logger);
StartCheckTimers();
+
+ await PublishToDirectoryNowAsync(session).ConfigureAwait(false);
}
///
@@ -900,10 +1356,23 @@ namespace Emby.Server.Implementations.Session
ArgumentNullException.ThrowIfNull(info);
+ // An automated report is generated from the copy of the session this instance already holds,
+ // so it is never the one that belongs somewhere else.
+ if (!isAutomated
+ && await TryRouteToOwnerAsync(info.SessionId, RoutedPlaybackReport.ProgressKind, info, CancellationToken.None).ConfigureAwait(false))
+ {
+ return;
+ }
+
+ await OnPlaybackProgressCore(info, isAutomated).ConfigureAwait(false);
+ }
+
+ private async Task OnPlaybackProgressCore(PlaybackProgressInfo info, bool isAutomated)
+ {
var session = GetSession(info.SessionId, false);
if (session is null)
{
- return;
+ return false;
}
var libraryItem = info.ItemId.IsEmpty()
@@ -960,6 +1429,10 @@ namespace Emby.Server.Implementations.Session
}
StartCheckTimers();
+
+ QueueDirectoryPublish(session);
+
+ return true;
}
private void OnPlaybackProgress(User user, BaseItem item, PlaybackProgressInfo info)
@@ -1056,6 +1529,16 @@ namespace Emby.Server.Implementations.Session
ArgumentNullException.ThrowIfNull(info);
+ if (await TryRouteToOwnerAsync(info.SessionId, RoutedPlaybackReport.StoppedKind, info, CancellationToken.None).ConfigureAwait(false))
+ {
+ return;
+ }
+
+ await OnPlaybackStoppedCore(info).ConfigureAwait(false);
+ }
+
+ private async Task OnPlaybackStoppedCore(PlaybackStopInfo info)
+ {
var session = GetSession(info.SessionId);
session.StopAutomaticProgress();
@@ -1160,6 +1643,8 @@ namespace Emby.Server.Implementations.Session
await _eventManager.PublishAsync(eventArgs).ConfigureAwait(false);
EventHelper.QueueEventIfNotNull(PlaybackStopped, this, eventArgs, _logger);
+
+ await PublishToDirectoryNowAsync(session).ConfigureAwait(false);
}
private bool OnPlaybackStopped(User user, BaseItem item, long? positionTicks, bool playbackFailed)
@@ -1219,18 +1704,35 @@ namespace Emby.Server.Implementations.Session
return session;
}
- private SessionInfo GetSessionToRemoteControl(string sessionId)
+ // Resolves the session a message has to be written to. A local SessionInfo without a live
+ // controller is a copy left behind by a request this instance happened to serve, not the
+ // connection, so it is never the answer: either the owner named by the directory can take the
+ // message or nothing can, and the caller is told so rather than handed a copy that silently
+ // swallows it. A directory that cannot be read raises rather than reading as "no such session".
+ private async Task GetSessionToRemoteControl(string sessionId)
{
- // Accept either device id or session id
- var session = Sessions.FirstOrDefault(i => string.Equals(i.Id, sessionId, StringComparison.Ordinal));
+ var local = Sessions.FirstOrDefault(i => string.Equals(i.Id, sessionId, StringComparison.Ordinal));
- if (session is null)
+ if (local is not null && local.SessionControllers.Any(i => i.IsSessionActive))
{
- throw new ResourceNotFoundException(
- string.Format(CultureInfo.InvariantCulture, "Session {0} not found.", sessionId));
+ return local;
}
- return session;
+ return await GetRemoteSession(sessionId).ConfigureAwait(false)
+ ?? throw new ResourceNotFoundException(
+ string.Format(CultureInfo.InvariantCulture, "No instance holds a connection to session {0}.", sessionId));
+ }
+
+ // Resolves a session to authorize against or to change server-side state on. Nothing is written to
+ // a connection here, so a copy without one still answers the question.
+ private async Task GetSessionForControl(string sessionId)
+ {
+ var local = Sessions.FirstOrDefault(i => string.Equals(i.Id, sessionId, StringComparison.Ordinal));
+
+ return local
+ ?? await GetRemoteSession(sessionId).ConfigureAwait(false)
+ ?? throw new ResourceNotFoundException(
+ string.Format(CultureInfo.InvariantCulture, "Session {0} not found.", sessionId));
}
///
@@ -1291,24 +1793,31 @@ namespace Emby.Server.Implementations.Session
}
///
- public Task SendGeneralCommand(string controllingSessionId, string sessionId, GeneralCommand command, CancellationToken cancellationToken)
+ public async Task SendGeneralCommand(string controllingSessionId, string sessionId, GeneralCommand command, CancellationToken cancellationToken)
{
CheckDisposed();
- var session = GetSessionToRemoteControl(sessionId);
+ var session = await GetSessionToRemoteControl(sessionId).ConfigureAwait(false);
if (!string.IsNullOrEmpty(controllingSessionId))
{
- var controllingSession = GetSession(controllingSessionId);
+ var controllingSession = await GetSessionForControl(controllingSessionId).ConfigureAwait(false);
AssertCanControl(session, controllingSession);
}
- return SendMessageToSession(session, SessionMessageType.GeneralCommand, command, cancellationToken);
+ await SendMessageToSession(session, SessionMessageType.GeneralCommand, command, cancellationToken).ConfigureAwait(false);
}
private static async Task SendMessageToSession(SessionInfo session, SessionMessageType name, T data, CancellationToken cancellationToken)
{
- var controllers = session.SessionControllers;
+ var controllers = session.SessionControllers.Where(i => i.IsSessionActive).ToList();
+
+ if (controllers.Count == 0)
+ {
+ throw new ResourceNotFoundException(
+ string.Format(CultureInfo.InvariantCulture, "No instance holds a connection to session {0}.", session.Id));
+ }
+
var messageId = Guid.NewGuid();
foreach (var controller in controllers)
@@ -1340,7 +1849,7 @@ namespace Emby.Server.Implementations.Session
{
CheckDisposed();
- var session = GetSessionToRemoteControl(sessionId);
+ var session = await GetSessionToRemoteControl(sessionId).ConfigureAwait(false);
var user = session.UserId.IsEmpty() ? null : _userManager.GetUserById(session.UserId);
@@ -1410,7 +1919,7 @@ namespace Emby.Server.Implementations.Session
if (!string.IsNullOrEmpty(controllingSessionId))
{
- var controllingSession = GetSession(controllingSessionId);
+ var controllingSession = await GetSessionForControl(controllingSessionId).ConfigureAwait(false);
AssertCanControl(session, controllingSession);
if (!controllingSession.UserId.IsEmpty())
{
@@ -1425,7 +1934,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 = GetConnectedSession(sessionId);
+ if (session is null)
+ {
+ _logger.LogDebug("SyncPlay command for session {Session} dropped; this instance does not hold its connection.", sessionId);
+ return;
+ }
+
await SendMessageToSession(session, SessionMessageType.SyncPlayCommand, command, cancellationToken).ConfigureAwait(false);
}
@@ -1433,10 +1951,26 @@ namespace Emby.Server.Implementations.Session
public async Task SendSyncPlayGroupUpdate(string sessionId, GroupUpdate command, CancellationToken cancellationToken)
{
CheckDisposed();
- var session = GetSession(sessionId);
+
+ var session = GetConnectedSession(sessionId);
+ if (session is null)
+ {
+ _logger.LogDebug("SyncPlay group update for session {Session} dropped; this instance does not hold its connection.", sessionId);
+ return;
+ }
+
await SendMessageToSession(session, SessionMessageType.SyncPlayGroupUpdate, command, cancellationToken).ConfigureAwait(false);
}
+ // Both instances keep a copy of a session whose requests they have served, so holding a copy is
+ // not holding the connection.
+ private SessionInfo GetConnectedSession(string sessionId)
+ {
+ var session = Sessions.FirstOrDefault(i => string.Equals(i.Id, sessionId, StringComparison.Ordinal));
+
+ return session?.SessionControllers.Any(i => i.IsSessionActive) == true ? session : null;
+ }
+
private IEnumerable TranslateItemForPlayback(Guid id, User user)
{
var item = _libraryManager.GetItemById(id);
@@ -1521,15 +2055,15 @@ namespace Emby.Server.Implementations.Session
}
///
- public Task SendPlaystateCommand(string controllingSessionId, string sessionId, PlaystateRequest command, CancellationToken cancellationToken)
+ public async Task SendPlaystateCommand(string controllingSessionId, string sessionId, PlaystateRequest command, CancellationToken cancellationToken)
{
CheckDisposed();
- var session = GetSessionToRemoteControl(sessionId);
+ var session = await GetSessionToRemoteControl(sessionId).ConfigureAwait(false);
if (!string.IsNullOrEmpty(controllingSessionId))
{
- var controllingSession = GetSession(controllingSessionId);
+ var controllingSession = await GetSessionForControl(controllingSessionId).ConfigureAwait(false);
AssertCanControl(session, controllingSession);
if (!controllingSession.UserId.IsEmpty())
{
@@ -1537,7 +2071,7 @@ namespace Emby.Server.Implementations.Session
}
}
- return SendMessageToSession(session, SessionMessageType.Playstate, command, cancellationToken);
+ await SendMessageToSession(session, SessionMessageType.Playstate, command, cancellationToken).ConfigureAwait(false);
}
private void AssertCanControl(SessionInfo session, SessionInfo controllingSession)
@@ -1606,17 +2140,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 GetSessionForControl(sessionId).ConfigureAwait(false);
if (!string.IsNullOrEmpty(controllingSessionId))
{
- var controllingSession = GetSession(controllingSessionId);
+ var controllingSession = await GetSessionForControl(controllingSessionId).ConfigureAwait(false);
AssertCanControl(session, controllingSession);
AssertCanAttachUser(controllingSession, userId);
}
@@ -1626,17 +2161,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];
+ await RouteAdditionalUserChange(sessionId, userId, true).ConfigureAwait(false);
+
+ var local = GetSession(sessionId, false);
+ if (local is not null)
+ {
+ AttachAdditionalUser(local, userId, user.Username);
+ QueueDirectoryPublish(local);
}
}
@@ -1646,17 +2180,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 GetSessionForControl(sessionId).ConfigureAwait(false);
if (!string.IsNullOrEmpty(controllingSessionId))
{
- AssertCanControl(session, GetSession(controllingSessionId));
+ AssertCanControl(session, await GetSessionForControl(controllingSessionId).ConfigureAwait(false));
}
if (session.UserId.Equals(userId))
@@ -1664,17 +2199,75 @@ 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));
+ await RouteAdditionalUserChange(sessionId, userId, false).ConfigureAwait(false);
- if (user is not null)
+ var local = GetSession(sessionId, false);
+ if (local is not null)
+ {
+ DetachAdditionalUser(local, userId);
+ QueueDirectoryPublish(local);
+ }
+ }
+
+ 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.RequestAsync(
+ entry.OwnerPod,
+ new PodMessage
+ {
+ Kind = RoutedAdditionalUserChange.Kind,
+ Payload = JsonSerializer.Serialize(payload, JsonDefaults.Options)
+ }).ConfigureAwait(false);
+
+ if (!delivered)
+ {
+ throw new ResourceNotFoundException(
+ string.Format(CultureInfo.InvariantCulture, "The instance holding session {0} did not apply the change.", sessionId));
+ }
+ }
+
///
/// Authenticates the new session.
///
@@ -1865,19 +2458,34 @@ namespace Emby.Server.Implementations.Session
/// The controlling session identifier.
/// The session identifier.
/// The capabilities.
+ /// A task representing the operation.
/// The controlling user is not allowed to control the session.
- public void ReportCapabilities(string controllingSessionId, string sessionId, ClientCapabilities capabilities)
+ public async Task ReportCapabilities(string controllingSessionId, string sessionId, ClientCapabilities capabilities)
{
CheckDisposed();
- var session = GetSession(sessionId);
+ var session = await GetSessionForControl(sessionId).ConfigureAwait(false);
if (!string.IsNullOrEmpty(controllingSessionId))
{
- AssertCanControl(session, GetSession(controllingSessionId));
+ AssertCanControl(session, await GetSessionForControl(controllingSessionId).ConfigureAwait(false));
}
- ReportCapabilities(session, capabilities, true);
+ // Capabilities decide whether the session is offered for remote control, and they are held in
+ // the owner's process, so the report has to reach the instance everyone else is shown.
+ var payload = new RoutedCapabilities { SessionId = sessionId, Capabilities = capabilities };
+ var routed = await TryRouteToOwnerAsync(sessionId, RoutedCapabilities.Kind, payload, CancellationToken.None).ConfigureAwait(false);
+
+ var local = GetSession(sessionId, false);
+ if (local is not null)
+ {
+ ReportCapabilities(local, capabilities, true);
+ }
+ else if (!routed)
+ {
+ throw new ResourceNotFoundException(
+ string.Format(CultureInfo.InvariantCulture, "Session {0} not found.", sessionId));
+ }
}
private void ReportCapabilities(
@@ -1897,6 +2505,10 @@ namespace Emby.Server.Implementations.Session
});
_deviceManager.SaveCapabilities(session.DeviceId, capabilities);
+
+ // Capabilities decide whether the session is listed as controllable, so the change is not
+ // left to the throttle.
+ _ = PublishToDirectoryNowAsync(session);
}
}
@@ -1971,19 +2583,36 @@ namespace Emby.Server.Implementations.Session
}
///
- public void ReportNowViewingItem(string controllingSessionId, string sessionId, string itemId)
+ public async Task ReportNowViewingItem(string controllingSessionId, string sessionId, string itemId)
{
ArgumentException.ThrowIfNullOrEmpty(itemId);
- var item = _libraryManager.GetItemById(new Guid(itemId));
- var session = GetSession(sessionId);
+ var session = await GetSessionForControl(sessionId).ConfigureAwait(false);
if (!string.IsNullOrEmpty(controllingSessionId))
{
- AssertCanControl(session, GetSession(controllingSessionId));
+ AssertCanControl(session, await GetSessionForControl(controllingSessionId).ConfigureAwait(false));
}
- session.NowViewingItem = GetItemInfo(item, null);
+ var payload = new RoutedNowViewingItem { SessionId = sessionId, ItemId = itemId };
+
+ if (await TryRouteToOwnerAsync(sessionId, RoutedNowViewingItem.Kind, payload, CancellationToken.None).ConfigureAwait(false))
+ {
+ return;
+ }
+
+ var local = GetSession(sessionId, false)
+ ?? throw new ResourceNotFoundException(
+ string.Format(CultureInfo.InvariantCulture, "Session {0} not found.", sessionId));
+
+ SetNowViewingItem(local, itemId);
+ }
+
+ private void SetNowViewingItem(SessionInfo session, string itemId)
+ {
+ session.NowViewingItem = GetItemInfo(_libraryManager.GetItemById(new Guid(itemId)), null);
+
+ QueueDirectoryPublish(session);
}
///
@@ -2060,14 +2689,25 @@ namespace Emby.Server.Implementations.Session
}
///
- public IReadOnlyList GetSessions(
+ public async Task> GetSessions(
Guid userId,
string deviceId,
int? activeWithinSeconds,
Guid? controllableUserToCheck,
- bool isApiKey)
+ bool isApiKey,
+ CancellationToken cancellationToken)
{
- var result = Sessions;
+ var remote = await GetRemoteEntriesAsync(cancellationToken).ConfigureAwait(false);
+ 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);
+
if (!string.IsNullOrEmpty(deviceId))
{
result = result.Where(i => string.Equals(i.DeviceId, deviceId, StringComparison.OrdinalIgnoreCase));
@@ -2115,7 +2755,7 @@ namespace Emby.Server.Implementations.Session
if (!userCanControlOthers)
{
// User cannot control other user's sessions, validate user id.
- result = result.Where(i => i.UserId.IsEmpty() || i.ContainsUser(userId));
+ result = result.Where(i => i.UserId.IsEmpty() || ContainsUser(i, userId));
}
result = result.Where(i =>
@@ -2136,7 +2776,7 @@ namespace Emby.Server.Implementations.Session
else if (!userIsAdmin)
{
// Request isn't from administrator, limit to "own" sessions.
- result = result.Where(i => i.UserId.IsEmpty() || i.ContainsUser(userId));
+ result = result.Where(i => i.UserId.IsEmpty() || ContainsUser(i, userId));
}
if (!userIsAdmin)
@@ -2159,7 +2799,18 @@ namespace Emby.Server.Implementations.Session
result = result.Where(i => i.LastActivityDate >= minActiveDate);
}
- return result.Select(ToSessionInfoDto).ToList();
+ return result.ToList();
+ }
+
+ private static bool ContainsUser(SessionInfoDto session, Guid userId)
+ {
+ if (session.UserId.Equals(userId))
+ {
+ return true;
+ }
+
+ return session.AdditionalUsers is not null
+ && session.AdditionalUsers.Any(i => i.UserId.Equals(userId));
}
///
@@ -2234,6 +2885,12 @@ namespace Emby.Server.Implementations.Session
_inactiveTimer = null;
}
+ if (_directoryTimer is not null)
+ {
+ await _directoryTimer.DisposeAsync().ConfigureAwait(false);
+ _directoryTimer = null;
+ }
+
await _shutdownCallback.DisposeAsync().ConfigureAwait(false);
_deviceManager.DeviceOptionsUpdated -= OnDeviceManagerDeviceOptionsUpdated;
@@ -2257,6 +2914,7 @@ namespace Emby.Server.Implementations.Session
// Close open websockets to allow Kestrel to shut down cleanly
foreach (var session in _activeConnections.Values)
{
+ await RemoveFromDirectoryAsync(session).ConfigureAwait(false);
await session.DisposeAsync().ConfigureAwait(false);
}
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 84c2d90fb1..48045e6220 100644
--- a/Jellyfin.Api/Controllers/SessionController.cs
+++ b/Jellyfin.Api/Controllers/SessionController.cs
@@ -52,18 +52,19 @@ public class SessionController : BaseJellyfinApiController
[HttpGet("Sessions")]
[Authorize]
[ProducesResponseType(StatusCodes.Status200OK)]
- public ActionResult> GetSessions(
+ public async Task>> GetSessions(
[FromQuery] Guid? controllableByUserId,
[FromQuery] string? deviceId,
[FromQuery] int? activeWithinSeconds)
{
Guid? controllableUserToCheck = controllableByUserId is null ? null : RequestHelpers.GetUserId(User, controllableByUserId);
- var result = _sessionManager.GetSessions(
+ var result = await _sessionManager.GetSessions(
User.GetUserId(),
deviceId,
activeWithinSeconds,
controllableUserToCheck,
- User.GetIsApiKey());
+ User.GetIsApiKey(),
+ HttpContext.RequestAborted).ConfigureAwait(false);
return Ok(result);
}
@@ -310,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();
}
@@ -331,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();
}
@@ -364,13 +365,13 @@ public class SessionController : BaseJellyfinApiController
id = currentSessionId;
}
- _sessionManager.ReportCapabilities(currentSessionId, id, new ClientCapabilities
+ await _sessionManager.ReportCapabilities(currentSessionId, id, new ClientCapabilities
{
PlayableMediaTypes = playableMediaTypes,
SupportedCommands = supportedCommands,
SupportsMediaControl = supportsMediaControl,
SupportsPersistentIdentifier = supportsPersistentIdentifier
- });
+ }).ConfigureAwait(false);
return NoContent();
}
@@ -394,7 +395,7 @@ public class SessionController : BaseJellyfinApiController
id = currentSessionId;
}
- _sessionManager.ReportCapabilities(currentSessionId, id, capabilities.ToClientCapabilities());
+ await _sessionManager.ReportCapabilities(currentSessionId, id, capabilities.ToClientCapabilities()).ConfigureAwait(false);
return NoContent();
}
@@ -415,7 +416,7 @@ public class SessionController : BaseJellyfinApiController
{
var currentSessionId = await RequestHelpers.GetSessionId(_sessionManager, _userManager, HttpContext).ConfigureAwait(false);
- _sessionManager.ReportNowViewingItem(currentSessionId, sessionId ?? currentSessionId, itemId);
+ await _sessionManager.ReportNowViewingItem(currentSessionId, sessionId ?? currentSessionId, itemId).ConfigureAwait(false);
return NoContent();
}
diff --git a/Jellyfin.Server/CoreAppHost.cs b/Jellyfin.Server/CoreAppHost.cs
index 4e30328870..783f8a9dd1 100644
--- a/Jellyfin.Server/CoreAppHost.cs
+++ b/Jellyfin.Server/CoreAppHost.cs
@@ -116,6 +116,10 @@ namespace Jellyfin.Server
// to the other instances. Redis-backed when configured, no-op otherwise.
serviceCollection.AddConfigurationInvalidationBus(_startupConfig, Logger);
+ // Session directory: publishes which instance holds which session and routes remote-control
+ // messages to it. Redis-backed when configured, no-op otherwise.
+ serviceCollection.AddSessionDirectory(_startupConfig, Logger);
+
// Quick connect store: shares in-flight quick connect requests so the initiate, authorize and
// exchange legs can land on different instances. Redis-backed when configured, local otherwise.
serviceCollection.AddQuickConnectStore(_startupConfig, Logger);
diff --git a/Jellyfin.Server/Extensions/SessionDirectoryServiceCollectionExtensions.cs b/Jellyfin.Server/Extensions/SessionDirectoryServiceCollectionExtensions.cs
new file mode 100644
index 0000000000..fc00725a83
--- /dev/null
+++ b/Jellyfin.Server/Extensions/SessionDirectoryServiceCollectionExtensions.cs
@@ -0,0 +1,94 @@
+using System;
+using Emby.Server.Implementations.Session;
+using MediaBrowser.Controller.MediaEncoding;
+using MediaBrowser.Controller.Session;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using StackExchange.Redis;
+
+namespace Jellyfin.Server.Extensions;
+
+///
+/// Extensions for registering the session directory and the instance-addressed message bus.
+///
+public static class SessionDirectoryServiceCollectionExtensions
+{
+ ///
+ /// Registers the session directory and message bus, Redis-backed when a connection string is
+ /// configured and no-op otherwise, and reports the selection at .
+ ///
+ /// The service collection.
+ /// The configuration to read Jellyfin:SessionDirectory from.
+ /// The logger to report the selection on.
+ /// The updated service collection.
+ public static IServiceCollection AddSessionDirectory(
+ this IServiceCollection serviceCollection,
+ IConfiguration configuration,
+ ILogger logger)
+ {
+ ArgumentNullException.ThrowIfNull(configuration);
+ ArgumentNullException.ThrowIfNull(logger);
+
+ serviceCollection.Configure(configuration.GetSection(SessionDirectoryOptions.ConfigurationSection));
+
+ if (string.IsNullOrEmpty(configuration[TranscodeStoreOptions.RedisConnectionStringKey]))
+ {
+ logger.LogInformation(
+ "Session directory: {Directory}. The session list and remote control only reach the sessions this instance holds; set {Key} to share them.",
+ nameof(NullSessionDirectory),
+ TranscodeStoreOptions.RedisConnectionStringKey);
+ }
+ else
+ {
+ logger.LogInformation(
+ "Session directory: {Directory}. Sessions are visible to, and controllable from, every instance.",
+ nameof(RedisSessionDirectory));
+
+ serviceCollection.AddSingleton(SharedSessionServices.Create);
+ }
+
+ serviceCollection.AddSingleton(sp => sp.GetService()?.Bus ?? NullPodMessageBus.Instance);
+
+ return serviceCollection.AddSingleton(sp => sp.GetService()?.Directory ?? NullSessionDirectory.Instance);
+ }
+
+ // A Redis directory paired with a no-op bus would leave every routed command permanently
+ // undeliverable, so the pair is built together and falls back together.
+ private sealed class SharedSessionServices
+ {
+ private SharedSessionServices(ISessionDirectory directory, IPodMessageBus bus)
+ {
+ Directory = directory;
+ Bus = bus;
+ }
+
+ public ISessionDirectory Directory { get; }
+
+ public IPodMessageBus Bus { get; }
+
+ // Fail open: an unreachable Redis degrades to the single-instance behaviour rather than aborting startup.
+ public static SharedSessionServices Create(IServiceProvider serviceProvider)
+ {
+ try
+ {
+ var redis = serviceProvider.GetRequiredService();
+ var options = serviceProvider.GetRequiredService>();
+
+ return new SharedSessionServices(
+ new RedisSessionDirectory(redis, options, serviceProvider.GetRequiredService>()),
+ new RedisPodMessageBus(redis, options, PodIdentity.Current, serviceProvider.GetRequiredService>()));
+ }
+ catch (Exception ex)
+ {
+ serviceProvider.GetRequiredService>().LogError(
+ ex,
+ "Redis is configured but unavailable, so sessions will not be shared between instances. Check {Key}.",
+ TranscodeStoreOptions.RedisConnectionStringKey);
+
+ return new SharedSessionServices(NullSessionDirectory.Instance, NullPodMessageBus.Instance);
+ }
+ }
+ }
+}
diff --git a/MediaBrowser.Controller/Session/IPodMessageBus.cs b/MediaBrowser.Controller/Session/IPodMessageBus.cs
new file mode 100644
index 0000000000..3aca9e7181
--- /dev/null
+++ b/MediaBrowser.Controller/Session/IPodMessageBus.cs
@@ -0,0 +1,35 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MediaBrowser.Controller.Session;
+
+///
+/// Point-to-point delivery between instances: every instance listens on a channel of its own, so a
+/// message can be addressed to the one instance holding a given connection.
+///
+public interface IPodMessageBus
+{
+ ///
+ /// Gets the identity of this instance.
+ ///
+ string PodId { get; }
+
+ ///
+ /// Sends a message to one instance and waits for that instance to report what it did with it. The
+ /// number of subscribers only proves the target's connection to the broker is up, so delivery is
+ /// taken from the acknowledgement of the instance that has to act on the message.
+ ///
+ /// The instance to deliver to.
+ /// The message.
+ /// The cancellation token.
+ /// true if the target acknowledged having carried the message out.
+ Task RequestAsync(string targetPod, PodMessage message, CancellationToken cancellationToken = default);
+
+ ///
+ /// Registers a handler for the messages addressed to this instance. Whatever the handler returns is
+ /// sent back to the origin as the acknowledgement.
+ ///
+ /// The handler.
+ void Subscribe(Func> handler);
+}
diff --git a/MediaBrowser.Controller/Session/ISessionDirectory.cs b/MediaBrowser.Controller/Session/ISessionDirectory.cs
new file mode 100644
index 0000000000..fe5562566a
--- /dev/null
+++ b/MediaBrowser.Controller/Session/ISessionDirectory.cs
@@ -0,0 +1,57 @@
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MediaBrowser.Controller.Session;
+
+///
+/// The shared record of which instance holds which session. Entries expire, so an instance that stops
+/// refreshing them drops out of every other instance's view instead of lingering.
+///
+public interface ISessionDirectory
+{
+ ///
+ /// Allocates the next connection epoch for a session. The counter lives in the shared store, so the
+ /// epochs of every instance are ordered by one clock instead of being compared across machines.
+ ///
+ /// The session identifier.
+ /// The cancellation token.
+ /// The allocated epoch, which is greater than every epoch allocated for the session before it.
+ Task AllocateConnectionEpochAsync(string sessionId, CancellationToken cancellationToken = default);
+
+ ///
+ /// 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.
+ /// The epoch of the publishing instance's connection to the session, or zero when it holds none.
+ /// The cancellation token.
+ /// true if the entry was written.
+ Task PublishAsync(SessionDirectoryEntry entry, long connectionEpoch, CancellationToken cancellationToken = default);
+
+ ///
+ /// 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.
+ /// The entry, or null when the session is in no instance's directory.
+ /// The store could not be read. An unreadable store is not an absent session.
+ Task GetAsync(string sessionId, CancellationToken cancellationToken = default);
+
+ ///
+ /// Gets every entry that has not expired.
+ ///
+ /// The cancellation token.
+ /// The entries.
+ Task> GetAllAsync(CancellationToken cancellationToken = default);
+}
diff --git a/MediaBrowser.Controller/Session/ISessionManager.cs b/MediaBrowser.Controller/Session/ISessionManager.cs
index 9acff745b9..fce9f7853e 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.
@@ -257,7 +260,8 @@ namespace MediaBrowser.Controller.Session
/// The controlling session identifier.
/// The session identifier.
/// The item identifier.
- void ReportNowViewingItem(string controllingSessionId, string sessionId, string itemId);
+ /// A task representing the operation.
+ Task ReportNowViewingItem(string controllingSessionId, string sessionId, string itemId);
///
/// Authenticates the new session.
@@ -274,7 +278,8 @@ namespace MediaBrowser.Controller.Session
/// The controlling session identifier.
/// The session identifier.
/// The capabilities.
- void ReportCapabilities(string controllingSessionId, string sessionId, ClientCapabilities capabilities);
+ /// A task representing the operation.
+ Task ReportCapabilities(string controllingSessionId, string sessionId, ClientCapabilities capabilities);
///
/// Reports the transcoding information.
@@ -306,8 +311,9 @@ namespace MediaBrowser.Controller.Session
/// Active within session limit.
/// Filter for sessions remote controllable for this user.
/// Is the request authenticated with API key.
- /// IReadOnlyList{SessionInfoDto}.
- IReadOnlyList GetSessions(Guid userId, string deviceId, int? activeWithinSeconds, Guid? controllableUserToCheck, bool isApiKey);
+ /// The cancellation token.
+ /// IReadOnlyList{SessionInfoDto}, including the sessions held by the other instances.
+ Task> GetSessions(Guid userId, string deviceId, int? activeWithinSeconds, Guid? controllableUserToCheck, bool isApiKey, CancellationToken cancellationToken);
///
/// Gets the session by authentication token.
diff --git a/MediaBrowser.Controller/Session/NullPodMessageBus.cs b/MediaBrowser.Controller/Session/NullPodMessageBus.cs
new file mode 100644
index 0000000000..1e95d6d907
--- /dev/null
+++ b/MediaBrowser.Controller/Session/NullPodMessageBus.cs
@@ -0,0 +1,28 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MediaBrowser.Controller.Session;
+
+///
+/// The single-instance : there is no other instance to reach.
+///
+public sealed class NullPodMessageBus : IPodMessageBus
+{
+ ///
+ /// Gets the shared instance.
+ ///
+ public static NullPodMessageBus Instance { get; } = new NullPodMessageBus();
+
+ ///
+ public string PodId => PodIdentity.Current;
+
+ ///
+ public Task RequestAsync(string targetPod, PodMessage message, CancellationToken cancellationToken = default)
+ => Task.FromResult(false);
+
+ ///
+ public void Subscribe(Func> handler)
+ {
+ }
+}
diff --git a/MediaBrowser.Controller/Session/NullSessionDirectory.cs b/MediaBrowser.Controller/Session/NullSessionDirectory.cs
new file mode 100644
index 0000000000..fa4ca717c6
--- /dev/null
+++ b/MediaBrowser.Controller/Session/NullSessionDirectory.cs
@@ -0,0 +1,38 @@
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MediaBrowser.Controller.Session;
+
+///
+/// The single-instance : nothing is published and no session is held
+/// anywhere but here, which is exactly the behaviour of a deployment without a shared store.
+///
+public sealed class NullSessionDirectory : ISessionDirectory
+{
+ ///
+ /// Gets the shared instance.
+ ///
+ public static NullSessionDirectory Instance { get; } = new NullSessionDirectory();
+
+ ///
+ public Task AllocateConnectionEpochAsync(string sessionId, CancellationToken cancellationToken = default)
+ => Task.FromResult(0L);
+
+ ///
+ public Task PublishAsync(SessionDirectoryEntry entry, long connectionEpoch, CancellationToken cancellationToken = default)
+ => Task.FromResult(false);
+
+ ///
+ public Task RemoveAsync(string sessionId, string ownerPod, CancellationToken cancellationToken = default)
+ => Task.CompletedTask;
+
+ ///
+ public Task GetAsync(string sessionId, CancellationToken cancellationToken = default)
+ => Task.FromResult(null);
+
+ ///
+ public Task> GetAllAsync(CancellationToken cancellationToken = default)
+ => Task.FromResult>(Array.Empty());
+}
diff --git a/MediaBrowser.Controller/Session/PodIdentity.cs b/MediaBrowser.Controller/Session/PodIdentity.cs
new file mode 100644
index 0000000000..b042897709
--- /dev/null
+++ b/MediaBrowser.Controller/Session/PodIdentity.cs
@@ -0,0 +1,14 @@
+using System;
+
+namespace MediaBrowser.Controller.Session;
+
+///
+/// The identity of this instance among the replicas sharing a deployment.
+///
+public static class PodIdentity
+{
+ ///
+ /// Gets the identity of this instance.
+ ///
+ public static string Current => Environment.GetEnvironmentVariable("JELLYFIN_INSTANCE_ID") ?? Environment.MachineName;
+}
diff --git a/MediaBrowser.Controller/Session/PodMessage.cs b/MediaBrowser.Controller/Session/PodMessage.cs
new file mode 100644
index 0000000000..3a0801521f
--- /dev/null
+++ b/MediaBrowser.Controller/Session/PodMessage.cs
@@ -0,0 +1,39 @@
+namespace MediaBrowser.Controller.Session;
+
+///
+/// An envelope addressed to one instance. names the payload so that features other
+/// than session routing can share the same channel.
+///
+public sealed class PodMessage
+{
+ ///
+ /// The of the acknowledgement the receiving instance sends back.
+ ///
+ public const string AckKind = "Ack";
+
+ ///
+ /// Gets or sets the payload discriminator.
+ ///
+ public string Kind { get; set; } = string.Empty;
+
+ ///
+ /// Gets or sets the identity of the sending instance.
+ ///
+ public string OriginPod { get; set; } = string.Empty;
+
+ ///
+ /// Gets or sets the identifier tying an acknowledgement to the message it answers.
+ ///
+ public string CorrelationId { get; set; } = string.Empty;
+
+ ///
+ /// Gets or sets a value indicating whether the receiving instance carried the message out. Only
+ /// meaningful on an message.
+ ///
+ public bool Handled { get; set; }
+
+ ///
+ /// Gets or sets the serialized payload.
+ ///
+ public string Payload { get; set; } = string.Empty;
+}
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/RoutedCapabilities.cs b/MediaBrowser.Controller/Session/RoutedCapabilities.cs
new file mode 100644
index 0000000000..4e2ffd406f
--- /dev/null
+++ b/MediaBrowser.Controller/Session/RoutedCapabilities.cs
@@ -0,0 +1,25 @@
+using MediaBrowser.Model.Session;
+
+namespace MediaBrowser.Controller.Session;
+
+///
+/// A capabilities report for a session held by another instance, carried as a .
+/// The calling instance has already authorized it.
+///
+public sealed class RoutedCapabilities
+{
+ ///
+ /// The this payload travels under.
+ ///
+ public const string Kind = "Capabilities";
+
+ ///
+ /// Gets or sets the session the report applies to.
+ ///
+ public string SessionId { get; set; } = string.Empty;
+
+ ///
+ /// Gets or sets the reported capabilities.
+ ///
+ public ClientCapabilities? Capabilities { get; set; }
+}
diff --git a/MediaBrowser.Controller/Session/RoutedNowViewingItem.cs b/MediaBrowser.Controller/Session/RoutedNowViewingItem.cs
new file mode 100644
index 0000000000..71eb0ce55f
--- /dev/null
+++ b/MediaBrowser.Controller/Session/RoutedNowViewingItem.cs
@@ -0,0 +1,23 @@
+namespace MediaBrowser.Controller.Session;
+
+///
+/// A now-viewing report for a session held by another instance, carried as a .
+/// The calling instance has already authorized it.
+///
+public sealed class RoutedNowViewingItem
+{
+ ///
+ /// The this payload travels under.
+ ///
+ public const string Kind = "NowViewingItem";
+
+ ///
+ /// Gets or sets the session the report applies to.
+ ///
+ public string SessionId { get; set; } = string.Empty;
+
+ ///
+ /// Gets or sets the item being viewed.
+ ///
+ public string ItemId { get; set; } = string.Empty;
+}
diff --git a/MediaBrowser.Controller/Session/RoutedPlaybackReport.cs b/MediaBrowser.Controller/Session/RoutedPlaybackReport.cs
new file mode 100644
index 0000000000..70397216f7
--- /dev/null
+++ b/MediaBrowser.Controller/Session/RoutedPlaybackReport.cs
@@ -0,0 +1,23 @@
+namespace MediaBrowser.Controller.Session;
+
+///
+/// The values a playback report travels under when the session it is
+/// addressed to is held by another instance. The payload is the report itself.
+///
+public static class RoutedPlaybackReport
+{
+ ///
+ /// A PlaybackStartInfo.
+ ///
+ public const string StartKind = "PlaybackStart";
+
+ ///
+ /// A PlaybackProgressInfo.
+ ///
+ public const string ProgressKind = "PlaybackProgress";
+
+ ///
+ /// A PlaybackStopInfo.
+ ///
+ public const string StoppedKind = "PlaybackStopped";
+}
diff --git a/MediaBrowser.Controller/Session/RoutedSessionMessage.cs b/MediaBrowser.Controller/Session/RoutedSessionMessage.cs
new file mode 100644
index 0000000000..a8ae8bb620
--- /dev/null
+++ b/MediaBrowser.Controller/Session/RoutedSessionMessage.cs
@@ -0,0 +1,35 @@
+using System;
+using MediaBrowser.Model.Session;
+
+namespace MediaBrowser.Controller.Session;
+
+///
+/// A websocket message for a session held by another instance, carried as a .
+///
+public sealed class RoutedSessionMessage
+{
+ ///
+ /// The this payload travels under.
+ ///
+ public const string Kind = "SessionMessage";
+
+ ///
+ /// Gets or sets the session the message is addressed to.
+ ///
+ public string SessionId { get; set; } = string.Empty;
+
+ ///
+ /// Gets or sets the message type.
+ ///
+ public SessionMessageType MessageType { get; set; }
+
+ ///
+ /// Gets or sets the message identifier.
+ ///
+ public Guid MessageId { get; set; }
+
+ ///
+ /// Gets or sets the message data, serialized as JSON.
+ ///
+ public string Data { get; set; } = "null";
+}
diff --git a/MediaBrowser.Controller/Session/SessionDirectoryEntry.cs b/MediaBrowser.Controller/Session/SessionDirectoryEntry.cs
new file mode 100644
index 0000000000..069325c8ca
--- /dev/null
+++ b/MediaBrowser.Controller/Session/SessionDirectoryEntry.cs
@@ -0,0 +1,25 @@
+using MediaBrowser.Model.Dto;
+
+namespace MediaBrowser.Controller.Session;
+
+///
+/// A session held by one instance, as the other instances see it.
+///
+public sealed class SessionDirectoryEntry
+{
+ ///
+ /// Gets or sets the identity of the instance holding the connection.
+ ///
+ 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.
+ ///
+ public SessionInfoDto? Session { get; set; }
+}
diff --git a/MediaBrowser.Controller/Session/SessionDirectoryOptions.cs b/MediaBrowser.Controller/Session/SessionDirectoryOptions.cs
new file mode 100644
index 0000000000..9baf25369c
--- /dev/null
+++ b/MediaBrowser.Controller/Session/SessionDirectoryOptions.cs
@@ -0,0 +1,28 @@
+namespace MediaBrowser.Controller.Session;
+
+///
+/// Configuration options for the session directory and the cross-instance bus that goes with it.
+///
+public sealed class SessionDirectoryOptions
+{
+ ///
+ /// The configuration section these options bind from.
+ ///
+ public const string ConfigurationSection = "Jellyfin:SessionDirectory";
+
+ ///
+ /// Gets or sets how long in seconds a published entry survives without being refreshed. An instance
+ /// that dies stops refreshing, so its sessions leave the directory after this long.
+ ///
+ public int EntryTtlSeconds { get; set; } = 60;
+
+ ///
+ /// 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 7ac2845e7d..c82dba9330 100644
--- a/README.md
+++ b/README.md
@@ -116,6 +116,9 @@ Without a connection string the line reads `Transcode session store: NullTransco
| `Jellyfin:TranscodeStore:RedisConnectionString` | _(empty)_ | StackExchange.Redis connection string. Empty = single-instance mode. |
| `Jellyfin:TranscodeStore:LeaseDurationSeconds` | `30` | How long a pod's transcode lease is valid before another pod may take over. |
| `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/IdlePlaybackTests.cs b/tests/Jellyfin.Server.Implementations.Tests/SessionManager/IdlePlaybackTests.cs
index 7722707cbe..b6655ec995 100644
--- a/tests/Jellyfin.Server.Implementations.Tests/SessionManager/IdlePlaybackTests.cs
+++ b/tests/Jellyfin.Server.Implementations.Tests/SessionManager/IdlePlaybackTests.cs
@@ -13,6 +13,7 @@ using MediaBrowser.Model.Dto;
using MediaBrowser.Model.Session;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
using Moq;
using Xunit;
@@ -44,7 +45,10 @@ public class IdlePlaybackTests
Mock.Of(),
Mock.Of(),
Mock.Of(),
- Mock.Of());
+ Mock.Of(),
+ NullSessionDirectory.Instance,
+ NullPodMessageBus.Instance,
+ Options.Create(new SessionDirectoryOptions()));
var session = await sessionManager.LogSessionActivity(
"Test Client",
"1.0.0",
diff --git a/tests/Jellyfin.Server.Implementations.Tests/SessionManager/SessionManagerTests.cs b/tests/Jellyfin.Server.Implementations.Tests/SessionManager/SessionManagerTests.cs
index f803c69af2..8848dcc2ea 100644
--- a/tests/Jellyfin.Server.Implementations.Tests/SessionManager/SessionManagerTests.cs
+++ b/tests/Jellyfin.Server.Implementations.Tests/SessionManager/SessionManagerTests.cs
@@ -16,6 +16,7 @@ using MediaBrowser.Controller.Session;
using MediaBrowser.Model.Session;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
using Moq;
using Xunit;
@@ -41,7 +42,10 @@ public class SessionManagerTests
Mock.Of(),
Mock.Of(),
Mock.Of(),
- Mock.Of());
+ Mock.Of(),
+ NullSessionDirectory.Instance,
+ NullPodMessageBus.Instance,
+ Options.Create(new SessionDirectoryOptions()));
await Assert.ThrowsAsync(exceptionType, () => sessionManager.GetAuthorizationToken(
new User("test", "default", "default"),
@@ -68,7 +72,10 @@ public class SessionManagerTests
Mock.Of(),
Mock.Of(),
Mock.Of(),
- Mock.Of());
+ Mock.Of(),
+ NullSessionDirectory.Instance,
+ NullPodMessageBus.Instance,
+ Options.Create(new SessionDirectoryOptions()));
await Assert.ThrowsAsync(exceptionType, () => sessionManager.AuthenticateNewSessionInternal(authenticationRequest, false));
}
@@ -122,6 +129,7 @@ public class SessionManagerTests
await using var sessionManager = CreateSessionManager(victim, attacker);
var victimSession = await LogSessionActivity(sessionManager, victim);
+ victimSession.AddController(new StubSessionController());
var attackerSession = await LogSessionActivity(sessionManager, attacker);
await Assert.ThrowsAsync(() => sessionManager.SendMessageCommand(
@@ -140,6 +148,7 @@ public class SessionManagerTests
await using var sessionManager = CreateSessionManager(victim, attacker);
var victimSession = await LogSessionActivity(sessionManager, victim);
+ victimSession.AddController(new StubSessionController());
var controllingSession = await LogSessionActivity(sessionManager, attacker);
await sessionManager.SendMessageCommand(
@@ -173,7 +182,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]
@@ -186,7 +195,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));
}
@@ -201,7 +210,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]
@@ -214,7 +223,7 @@ public class SessionManagerTests
var victimSession = await LogSessionActivity(sessionManager, victim);
var attackerSession = await LogSessionActivity(sessionManager, attacker);
- Assert.Throws(() => sessionManager.ReportCapabilities(attackerSession.Id, victimSession.Id, new ClientCapabilities()));
+ await Assert.ThrowsAsync(() => sessionManager.ReportCapabilities(attackerSession.Id, victimSession.Id, new ClientCapabilities()));
}
private static Emby.Server.Implementations.Session.SessionManager CreateSessionManager(params User[] users)
@@ -238,11 +247,25 @@ public class SessionManagerTests
Mock.Of(),
Mock.Of(),
Mock.Of(),
- Mock.Of());
+ Mock.Of(),
+ NullSessionDirectory.Instance,
+ NullPodMessageBus.Instance,
+ Options.Create(new SessionDirectoryOptions()));
}
// All sessions are logged with the same client and device id on purpose, those values are taken
// from the request headers and are not bound to the access token of the calling user.
private static Task LogSessionActivity(ISessionManager sessionManager, User user)
=> sessionManager.LogSessionActivity("Jellyfin Web", "1.0.0", "victim-tv-01", "device_name", "127.0.0.1", user);
+
+ // A session a command can be delivered to is one with a live connection.
+ private sealed class StubSessionController : ISessionController
+ {
+ public bool IsSessionActive => true;
+
+ public bool SupportsMediaControl => true;
+
+ public Task SendMessage(SessionMessageType name, Guid messageId, T data, CancellationToken cancellationToken)
+ => Task.CompletedTask;
+ }
}
diff --git a/tests/Jellyfin.Server.Tests/HighAvailability/RedisTestServer.cs b/tests/Jellyfin.Server.Tests/HighAvailability/RedisTestServer.cs
index a17850522c..3e62b71b16 100644
--- a/tests/Jellyfin.Server.Tests/HighAvailability/RedisTestServer.cs
+++ b/tests/Jellyfin.Server.Tests/HighAvailability/RedisTestServer.cs
@@ -6,9 +6,9 @@ using Testcontainers.Redis;
namespace Jellyfin.Server.Tests.HighAvailability;
///
-/// Hands out a Redis server for the tests that need one. A server named by JELLYFIN_TEST_REDIS is
-/// used as is, so CI can run one beside the step instead of a docker daemon of its own; without it a
-/// container is started through testcontainers.
+/// Hands out a valkey/Redis server for the tests that need one. A server named by
+/// JELLYFIN_TEST_REDIS is used as is, so CI can run one in the step instead of a docker daemon
+/// of its own; without it a container is started through testcontainers.
///
public sealed class RedisTestServer : IAsyncDisposable
{
@@ -31,7 +31,7 @@ public sealed class RedisTestServer : IAsyncDisposable
public string ConnectionString { get; }
///
- /// Starts or attaches to a Redis server and waits until it accepts connections.
+ /// Starts or attaches to a server and waits until it accepts connections.
///
/// The running server.
public static async Task StartAsync()
@@ -44,7 +44,7 @@ public sealed class RedisTestServer : IAsyncDisposable
return attached;
}
- var container = new RedisBuilder("redis:7-alpine").Build();
+ var container = new RedisBuilder("valkey/valkey:8-alpine").Build();
await container.StartAsync().ConfigureAwait(false);
var started = new RedisTestServer(container, container.GetConnectionString());
diff --git a/tests/Jellyfin.Server.Tests/HighAvailability/SessionDirectoryRegistrationTests.cs b/tests/Jellyfin.Server.Tests/HighAvailability/SessionDirectoryRegistrationTests.cs
new file mode 100644
index 0000000000..3214ae592a
--- /dev/null
+++ b/tests/Jellyfin.Server.Tests/HighAvailability/SessionDirectoryRegistrationTests.cs
@@ -0,0 +1,64 @@
+using System.Collections.Generic;
+using Jellyfin.Server.Extensions;
+using MediaBrowser.Controller.MediaEncoding;
+using MediaBrowser.Controller.Session;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging.Abstractions;
+using Moq;
+using StackExchange.Redis;
+using Xunit;
+
+namespace Jellyfin.Server.Tests.HighAvailability;
+
+///
+/// The directory and the bus are two halves of one mechanism: a shared directory paired with a bus that
+/// reaches nobody advertises sessions from every instance and then fails every command sent to one.
+///
+public static class SessionDirectoryRegistrationTests
+{
+ [Fact]
+ public static void AddSessionDirectory_WithoutARedisConnection_RegistersNeitherHalf()
+ {
+ var services = Build(configured: false);
+
+ Assert.IsType(services.GetRequiredService());
+ Assert.IsType(services.GetRequiredService());
+ }
+
+ [Fact]
+ public static void AddSessionDirectory_WhenOneHalfCannotBeBuilt_FallsBackOnBoth()
+ {
+ // A connection that can serve the directory but not the bus: taking the directory on its own
+ // would advertise every instance's sessions and then fail every command sent to one.
+ var redis = new Mock();
+ redis.Setup(i => i.GetDatabase(It.IsAny(), It.IsAny