test(session): cover the maintenance sweeps and capabilities routing

# Conflicts:
#	.woodpecker/ci.yaml
#	Jellyfin.Server/CoreAppHost.cs
#	tests/Jellyfin.Server.Tests/HighAvailability/RedisTestServer.cs

.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"

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;
+
+/// <summary>
+/// A Redis pub/sub <see cref="IPodMessageBus"/>. 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.
+/// </summary>
+public sealed class RedisPodMessageBus : IPodMessageBus
+{
+    private const string ChannelPrefix = "jellyfin:pod:";
+
+    private static readonly JsonSerializerOptions _jsonOptions = JsonDefaults.Options;
+
+    private readonly ConcurrentDictionary<string, TaskCompletionSource<bool>> _pending = new(StringComparer.Ordinal);
+    private readonly ISubscriber _subscriber;
+    private readonly ILogger<RedisPodMessageBus> _logger;
+    private readonly TimeSpan _timeout;
+
+    private Func<PodMessage, Task<bool>>? _handler;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="RedisPodMessageBus"/> class.
+    /// </summary>
+    /// <param name="redis">The Redis connection multiplexer.</param>
+    /// <param name="options">The session directory configuration options.</param>
+    /// <param name="podId">The identity of this instance.</param>
+    /// <param name="logger">The logger.</param>
+    public RedisPodMessageBus(
+        IConnectionMultiplexer redis,
+        IOptions<SessionDirectoryOptions> options,
+        string podId,
+        ILogger<RedisPodMessageBus> 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));
+    }
+
+    /// <inheritdoc />
+    public string PodId { get; }
+
+    /// <inheritdoc />
+    public async Task<bool> 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<bool>(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 _);
+        }
+    }
+
+    /// <inheritdoc />
+    public void Subscribe(Func<PodMessage, Task<bool>> handler)
+    {
+        ArgumentNullException.ThrowIfNull(handler);
+
+        _handler = handler;
+    }
+
+    private async void Dispatch(RedisValue value)
+    {
+        PodMessage? message = null;
+
+        try
+        {
+            message = JsonSerializer.Deserialize<PodMessage>(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);
+        }
+    }
+}

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;
+
+/// <summary>
+/// A Redis-backed <see cref="ISessionDirectory"/>. 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.
+/// </summary>
+public sealed class RedisSessionDirectory : ISessionDirectory
+{
+    private const string KeyPrefix = "jellyfin:session:";
+    private const string OwnerKeyPrefix = "jellyfin:sessionowner:";
+    private const string EpochKeyPrefix = "jellyfin:sessionepoch:";
+
+    /// <summary>
+    /// Lua script for an atomic ownership claim. The owner key holds <c>epoch|pod</c>, 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.
+    /// </summary>
+    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";
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    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";
+
+    /// <summary>
+    /// Lua script allocating the next connection epoch. The counter outlives the entries that reference
+    /// it, so it never restarts underneath a recorded epoch.
+    /// </summary>
+    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<RedisSessionDirectory> _logger;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="RedisSessionDirectory"/> class.
+    /// </summary>
+    /// <param name="redis">The Redis connection multiplexer.</param>
+    /// <param name="options">The session directory configuration options.</param>
+    /// <param name="logger">The logger.</param>
+    public RedisSessionDirectory(
+        IConnectionMultiplexer redis,
+        IOptions<SessionDirectoryOptions> options,
+        ILogger<RedisSessionDirectory> 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));
+
+    /// <inheritdoc />
+    public async Task<long> 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;
+    }
+
+    /// <inheritdoc />
+    public async Task<bool> 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;
+        }
+    }
+
+    /// <inheritdoc />
+    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);
+        }
+    }
+
+    /// <inheritdoc />
+    public async Task<SessionDirectoryEntry?> 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;
+    }
+
+    /// <inheritdoc />
+    public async Task<IReadOnlyList<SessionDirectoryEntry>> GetAllAsync(CancellationToken cancellationToken = default)
+    {
+        var entries = new List<SessionDirectoryEntry>();
+
+        foreach (var server in _redis.GetServers())
+        {
+            if (!server.IsConnected)
+            {
+                continue;
+            }
+
+            var keys = new List<RedisKey>();
+
+            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<RedisValue> 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<SessionDirectoryEntry>(raw.ToString(), _jsonOptions);
+        }
+        catch (JsonException ex)
+        {
+            _logger.LogWarning(ex, "Failed to deserialize a session directory entry.");
+            return null;
+        }
+    }
+}

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;
+
+/// <summary>
+/// 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.
+/// </summary>
+public sealed class RemoteSessionController : ISessionController
+{
+    private readonly IPodMessageBus _bus;
+    private readonly ILogger _logger;
+    private readonly string _ownerPod;
+    private readonly string _sessionId;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="RemoteSessionController"/> class.
+    /// </summary>
+    /// <param name="bus">The cross-instance bus.</param>
+    /// <param name="logger">The logger.</param>
+    /// <param name="ownerPod">The instance holding the connection.</param>
+    /// <param name="sessionId">The session identifier.</param>
+    /// <param name="supportsMediaControl">Whether the owner reported the session as controllable.</param>
+    /// <param name="holdsConnection">Whether the owner reported that it holds the session's connection.</param>
+    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;
+    }
+
+    /// <inheritdoc />
+    public bool IsSessionActive { get; }
+
+    /// <inheritdoc />
+    public bool SupportsMediaControl { get; }
+
+    /// <inheritdoc />
+    public async Task SendMessage<T>(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));
+        }
+    }
+}

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<WebSocketController>(
                 s => new WebSocketController(_loggerFactory.CreateLogger<WebSocketController>(), 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);
         }
 
         /// <summary>

Jellyfin.Api/Controllers/SessionController.cs

@@ -52,18 +52,19 @@ public class SessionController : BaseJellyfinApiController
     [HttpGet("Sessions")]
     [Authorize]
     [ProducesResponseType(StatusCodes.Status200OK)]
-    public ActionResult<IReadOnlyList<SessionInfoDto>> GetSessions(
+    public async Task<ActionResult<IReadOnlyList<SessionInfoDto>>> 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();
     }
 

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);

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;
+
+/// <summary>
+/// Extensions for registering the session directory and the instance-addressed message bus.
+/// </summary>
+public static class SessionDirectoryServiceCollectionExtensions
+{
+    /// <summary>
+    /// Registers the session directory and message bus, Redis-backed when a connection string is
+    /// configured and no-op otherwise, and reports the selection at <see cref="LogLevel.Information"/>.
+    /// </summary>
+    /// <param name="serviceCollection">The service collection.</param>
+    /// <param name="configuration">The configuration to read <c>Jellyfin:SessionDirectory</c> from.</param>
+    /// <param name="logger">The logger to report the selection on.</param>
+    /// <returns>The updated service collection.</returns>
+    public static IServiceCollection AddSessionDirectory(
+        this IServiceCollection serviceCollection,
+        IConfiguration configuration,
+        ILogger logger)
+    {
+        ArgumentNullException.ThrowIfNull(configuration);
+        ArgumentNullException.ThrowIfNull(logger);
+
+        serviceCollection.Configure<SessionDirectoryOptions>(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<IPodMessageBus>(sp => sp.GetService<SharedSessionServices>()?.Bus ?? NullPodMessageBus.Instance);
+
+        return serviceCollection.AddSingleton<ISessionDirectory>(sp => sp.GetService<SharedSessionServices>()?.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<IConnectionMultiplexer>();
+                var options = serviceProvider.GetRequiredService<IOptions<SessionDirectoryOptions>>();
+
+                return new SharedSessionServices(
+                    new RedisSessionDirectory(redis, options, serviceProvider.GetRequiredService<ILogger<RedisSessionDirectory>>()),
+                    new RedisPodMessageBus(redis, options, PodIdentity.Current, serviceProvider.GetRequiredService<ILogger<RedisPodMessageBus>>()));
+            }
+            catch (Exception ex)
+            {
+                serviceProvider.GetRequiredService<ILogger<CoreAppHost>>().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);
+            }
+        }
+    }
+}

MediaBrowser.Controller/Session/IPodMessageBus.cs

@@ -0,0 +1,35 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MediaBrowser.Controller.Session;
+
+/// <summary>
+/// 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.
+/// </summary>
+public interface IPodMessageBus
+{
+    /// <summary>
+    /// Gets the identity of this instance.
+    /// </summary>
+    string PodId { get; }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    /// <param name="targetPod">The instance to deliver to.</param>
+    /// <param name="message">The message.</param>
+    /// <param name="cancellationToken">The cancellation token.</param>
+    /// <returns><c>true</c> if the target acknowledged having carried the message out.</returns>
+    Task<bool> RequestAsync(string targetPod, PodMessage message, CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// Registers a handler for the messages addressed to this instance. Whatever the handler returns is
+    /// sent back to the origin as the acknowledgement.
+    /// </summary>
+    /// <param name="handler">The handler.</param>
+    void Subscribe(Func<PodMessage, Task<bool>> handler);
+}

MediaBrowser.Controller/Session/ISessionDirectory.cs

@@ -0,0 +1,57 @@
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MediaBrowser.Controller.Session;
+
+/// <summary>
+/// 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.
+/// </summary>
+public interface ISessionDirectory
+{
+    /// <summary>
+    /// 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.
+    /// </summary>
+    /// <param name="sessionId">The session identifier.</param>
+    /// <param name="cancellationToken">The cancellation token.</param>
+    /// <returns>The allocated epoch, which is greater than every epoch allocated for the session before it.</returns>
+    Task<long> AllocateConnectionEpochAsync(string sessionId, CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    /// <param name="entry">The entry.</param>
+    /// <param name="connectionEpoch">The epoch of the publishing instance's connection to the session, or zero when it holds none.</param>
+    /// <param name="cancellationToken">The cancellation token.</param>
+    /// <returns><c>true</c> if the entry was written.</returns>
+    Task<bool> PublishAsync(SessionDirectoryEntry entry, long connectionEpoch, CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// Removes an entry, but only while the calling instance still owns it.
+    /// </summary>
+    /// <param name="sessionId">The session identifier.</param>
+    /// <param name="ownerPod">The instance requesting the removal.</param>
+    /// <param name="cancellationToken">The cancellation token.</param>
+    /// <returns>A task representing the operation.</returns>
+    Task RemoveAsync(string sessionId, string ownerPod, CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// Gets one entry by session identifier.
+    /// </summary>
+    /// <param name="sessionId">The session identifier.</param>
+    /// <param name="cancellationToken">The cancellation token.</param>
+    /// <returns>The entry, or <c>null</c> when the session is in no instance's directory.</returns>
+    /// <exception cref="System.Exception">The store could not be read. An unreadable store is not an absent session.</exception>
+    Task<SessionDirectoryEntry?> GetAsync(string sessionId, CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// Gets every entry that has not expired.
+    /// </summary>
+    /// <param name="cancellationToken">The cancellation token.</param>
+    /// <returns>The entries.</returns>
+    Task<IReadOnlyList<SessionDirectoryEntry>> GetAllAsync(CancellationToken cancellationToken = default);
+}

MediaBrowser.Controller/Session/ISessionManager.cs

@@ -80,7 +80,8 @@ namespace MediaBrowser.Controller.Session
         /// Used to report that a session controller has connected.
         /// </summary>
         /// <param name="session">The session.</param>
-        void OnSessionControllerConnected(SessionInfo session);
+        /// <returns>A task representing the operation.</returns>
+        Task OnSessionControllerConnected(SessionInfo session);
 
         void UpdateDeviceName(string sessionId, string reportedDeviceName);
 
@@ -241,7 +242,8 @@ namespace MediaBrowser.Controller.Session
         /// <param name="controllingSessionId">The controlling session identifier.</param>
         /// <param name="sessionId">The session identifier.</param>
         /// <param name="userId">The user identifier.</param>
-        void AddAdditionalUser(string controllingSessionId, string sessionId, Guid userId);
+        /// <returns>A task representing the operation.</returns>
+        Task AddAdditionalUser(string controllingSessionId, string sessionId, Guid userId);
 
         /// <summary>
         /// Removes the additional user.
@@ -249,7 +251,8 @@ namespace MediaBrowser.Controller.Session
         /// <param name="controllingSessionId">The controlling session identifier.</param>
         /// <param name="sessionId">The session identifier.</param>
         /// <param name="userId">The user identifier.</param>
-        void RemoveAdditionalUser(string controllingSessionId, string sessionId, Guid userId);
+        /// <returns>A task representing the operation.</returns>
+        Task RemoveAdditionalUser(string controllingSessionId, string sessionId, Guid userId);
 
         /// <summary>
         /// Reports the now viewing item.
@@ -257,7 +260,8 @@ namespace MediaBrowser.Controller.Session
         /// <param name="controllingSessionId">The controlling session identifier.</param>
         /// <param name="sessionId">The session identifier.</param>
         /// <param name="itemId">The item identifier.</param>
-        void ReportNowViewingItem(string controllingSessionId, string sessionId, string itemId);
+        /// <returns>A task representing the operation.</returns>
+        Task ReportNowViewingItem(string controllingSessionId, string sessionId, string itemId);
 
         /// <summary>
         /// Authenticates the new session.
@@ -274,7 +278,8 @@ namespace MediaBrowser.Controller.Session
         /// <param name="controllingSessionId">The controlling session identifier.</param>
         /// <param name="sessionId">The session identifier.</param>
         /// <param name="capabilities">The capabilities.</param>
-        void ReportCapabilities(string controllingSessionId, string sessionId, ClientCapabilities capabilities);
+        /// <returns>A task representing the operation.</returns>
+        Task ReportCapabilities(string controllingSessionId, string sessionId, ClientCapabilities capabilities);
 
         /// <summary>
         /// Reports the transcoding information.
@@ -306,8 +311,9 @@ namespace MediaBrowser.Controller.Session
         /// <param name="activeWithinSeconds">Active within session limit.</param>
         /// <param name="controllableUserToCheck">Filter for sessions remote controllable for this user.</param>
         /// <param name="isApiKey">Is the request authenticated with API key.</param>
-        /// <returns>IReadOnlyList{SessionInfoDto}.</returns>
-        IReadOnlyList<SessionInfoDto> GetSessions(Guid userId, string deviceId, int? activeWithinSeconds, Guid? controllableUserToCheck, bool isApiKey);
+        /// <param name="cancellationToken">The cancellation token.</param>
+        /// <returns>IReadOnlyList{SessionInfoDto}, including the sessions held by the other instances.</returns>
+        Task<IReadOnlyList<SessionInfoDto>> GetSessions(Guid userId, string deviceId, int? activeWithinSeconds, Guid? controllableUserToCheck, bool isApiKey, CancellationToken cancellationToken);
 
         /// <summary>
         /// Gets the session by authentication token.

MediaBrowser.Controller/Session/NullPodMessageBus.cs

@@ -0,0 +1,28 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MediaBrowser.Controller.Session;
+
+/// <summary>
+/// The single-instance <see cref="IPodMessageBus"/>: there is no other instance to reach.
+/// </summary>
+public sealed class NullPodMessageBus : IPodMessageBus
+{
+    /// <summary>
+    /// Gets the shared instance.
+    /// </summary>
+    public static NullPodMessageBus Instance { get; } = new NullPodMessageBus();
+
+    /// <inheritdoc />
+    public string PodId => PodIdentity.Current;
+
+    /// <inheritdoc />
+    public Task<bool> RequestAsync(string targetPod, PodMessage message, CancellationToken cancellationToken = default)
+        => Task.FromResult(false);
+
+    /// <inheritdoc />
+    public void Subscribe(Func<PodMessage, Task<bool>> handler)
+    {
+    }
+}

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;
+
+/// <summary>
+/// The single-instance <see cref="ISessionDirectory"/>: nothing is published and no session is held
+/// anywhere but here, which is exactly the behaviour of a deployment without a shared store.
+/// </summary>
+public sealed class NullSessionDirectory : ISessionDirectory
+{
+    /// <summary>
+    /// Gets the shared instance.
+    /// </summary>
+    public static NullSessionDirectory Instance { get; } = new NullSessionDirectory();
+
+    /// <inheritdoc />
+    public Task<long> AllocateConnectionEpochAsync(string sessionId, CancellationToken cancellationToken = default)
+        => Task.FromResult(0L);
+
+    /// <inheritdoc />
+    public Task<bool> PublishAsync(SessionDirectoryEntry entry, long connectionEpoch, CancellationToken cancellationToken = default)
+        => Task.FromResult(false);
+
+    /// <inheritdoc />
+    public Task RemoveAsync(string sessionId, string ownerPod, CancellationToken cancellationToken = default)
+        => Task.CompletedTask;
+
+    /// <inheritdoc />
+    public Task<SessionDirectoryEntry?> GetAsync(string sessionId, CancellationToken cancellationToken = default)
+        => Task.FromResult<SessionDirectoryEntry?>(null);
+
+    /// <inheritdoc />
+    public Task<IReadOnlyList<SessionDirectoryEntry>> GetAllAsync(CancellationToken cancellationToken = default)
+        => Task.FromResult<IReadOnlyList<SessionDirectoryEntry>>(Array.Empty<SessionDirectoryEntry>());
+}

MediaBrowser.Controller/Session/PodIdentity.cs

@@ -0,0 +1,14 @@
+using System;
+
+namespace MediaBrowser.Controller.Session;
+
+/// <summary>
+/// The identity of this instance among the replicas sharing a deployment.
+/// </summary>
+public static class PodIdentity
+{
+    /// <summary>
+    /// Gets the identity of this instance.
+    /// </summary>
+    public static string Current => Environment.GetEnvironmentVariable("JELLYFIN_INSTANCE_ID") ?? Environment.MachineName;
+}

MediaBrowser.Controller/Session/PodMessage.cs

@@ -0,0 +1,39 @@
+namespace MediaBrowser.Controller.Session;
+
+/// <summary>
+/// An envelope addressed to one instance. <see cref="Kind"/> names the payload so that features other
+/// than session routing can share the same channel.
+/// </summary>
+public sealed class PodMessage
+{
+    /// <summary>
+    /// The <see cref="Kind"/> of the acknowledgement the receiving instance sends back.
+    /// </summary>
+    public const string AckKind = "Ack";
+
+    /// <summary>
+    /// Gets or sets the payload discriminator.
+    /// </summary>
+    public string Kind { get; set; } = string.Empty;
+
+    /// <summary>
+    /// Gets or sets the identity of the sending instance.
+    /// </summary>
+    public string OriginPod { get; set; } = string.Empty;
+
+    /// <summary>
+    /// Gets or sets the identifier tying an acknowledgement to the message it answers.
+    /// </summary>
+    public string CorrelationId { get; set; } = string.Empty;
+
+    /// <summary>
+    /// Gets or sets a value indicating whether the receiving instance carried the message out. Only
+    /// meaningful on an <see cref="AckKind"/> message.
+    /// </summary>
+    public bool Handled { get; set; }
+
+    /// <summary>
+    /// Gets or sets the serialized payload.
+    /// </summary>
+    public string Payload { get; set; } = string.Empty;
+}

MediaBrowser.Controller/Session/RoutedAdditionalUserChange.cs

@@ -0,0 +1,30 @@
+using System;
+
+namespace MediaBrowser.Controller.Session;
+
+/// <summary>
+/// An additional-user change for a session held by another instance, carried as a
+/// <see cref="PodMessage"/>. The calling instance has already authorized it.
+/// </summary>
+public sealed class RoutedAdditionalUserChange
+{
+    /// <summary>
+    /// The <see cref="PodMessage.Kind"/> this payload travels under.
+    /// </summary>
+    public const string Kind = "AdditionalUserChange";
+
+    /// <summary>
+    /// Gets or sets the session the change applies to.
+    /// </summary>
+    public string SessionId { get; set; } = string.Empty;
+
+    /// <summary>
+    /// Gets or sets the user to attach or detach.
+    /// </summary>
+    public Guid UserId { get; set; }
+
+    /// <summary>
+    /// Gets or sets a value indicating whether the user is being attached rather than detached.
+    /// </summary>
+    public bool Add { get; set; }
+}

MediaBrowser.Controller/Session/RoutedCapabilities.cs

@@ -0,0 +1,25 @@
+using MediaBrowser.Model.Session;
+
+namespace MediaBrowser.Controller.Session;
+
+/// <summary>
+/// A capabilities report for a session held by another instance, carried as a <see cref="PodMessage"/>.
+/// The calling instance has already authorized it.
+/// </summary>
+public sealed class RoutedCapabilities
+{
+    /// <summary>
+    /// The <see cref="PodMessage.Kind"/> this payload travels under.
+    /// </summary>
+    public const string Kind = "Capabilities";
+
+    /// <summary>
+    /// Gets or sets the session the report applies to.
+    /// </summary>
+    public string SessionId { get; set; } = string.Empty;
+
+    /// <summary>
+    /// Gets or sets the reported capabilities.
+    /// </summary>
+    public ClientCapabilities? Capabilities { get; set; }
+}

MediaBrowser.Controller/Session/RoutedNowViewingItem.cs

@@ -0,0 +1,23 @@
+namespace MediaBrowser.Controller.Session;
+
+/// <summary>
+/// A now-viewing report for a session held by another instance, carried as a <see cref="PodMessage"/>.
+/// The calling instance has already authorized it.
+/// </summary>
+public sealed class RoutedNowViewingItem
+{
+    /// <summary>
+    /// The <see cref="PodMessage.Kind"/> this payload travels under.
+    /// </summary>
+    public const string Kind = "NowViewingItem";
+
+    /// <summary>
+    /// Gets or sets the session the report applies to.
+    /// </summary>
+    public string SessionId { get; set; } = string.Empty;
+
+    /// <summary>
+    /// Gets or sets the item being viewed.
+    /// </summary>
+    public string ItemId { get; set; } = string.Empty;
+}

MediaBrowser.Controller/Session/RoutedPlaybackReport.cs

@@ -0,0 +1,23 @@
+namespace MediaBrowser.Controller.Session;
+
+/// <summary>
+/// The <see cref="PodMessage.Kind"/> values a playback report travels under when the session it is
+/// addressed to is held by another instance. The payload is the report itself.
+/// </summary>
+public static class RoutedPlaybackReport
+{
+    /// <summary>
+    /// A <c>PlaybackStartInfo</c>.
+    /// </summary>
+    public const string StartKind = "PlaybackStart";
+
+    /// <summary>
+    /// A <c>PlaybackProgressInfo</c>.
+    /// </summary>
+    public const string ProgressKind = "PlaybackProgress";
+
+    /// <summary>
+    /// A <c>PlaybackStopInfo</c>.
+    /// </summary>
+    public const string StoppedKind = "PlaybackStopped";
+}

MediaBrowser.Controller/Session/RoutedSessionMessage.cs

@@ -0,0 +1,35 @@
+using System;
+using MediaBrowser.Model.Session;
+
+namespace MediaBrowser.Controller.Session;
+
+/// <summary>
+/// A websocket message for a session held by another instance, carried as a <see cref="PodMessage"/>.
+/// </summary>
+public sealed class RoutedSessionMessage
+{
+    /// <summary>
+    /// The <see cref="PodMessage.Kind"/> this payload travels under.
+    /// </summary>
+    public const string Kind = "SessionMessage";
+
+    /// <summary>
+    /// Gets or sets the session the message is addressed to.
+    /// </summary>
+    public string SessionId { get; set; } = string.Empty;
+
+    /// <summary>
+    /// Gets or sets the message type.
+    /// </summary>
+    public SessionMessageType MessageType { get; set; }
+
+    /// <summary>
+    /// Gets or sets the message identifier.
+    /// </summary>
+    public Guid MessageId { get; set; }
+
+    /// <summary>
+    /// Gets or sets the message data, serialized as JSON.
+    /// </summary>
+    public string Data { get; set; } = "null";
+}

MediaBrowser.Controller/Session/SessionDirectoryEntry.cs

@@ -0,0 +1,25 @@
+using MediaBrowser.Model.Dto;
+
+namespace MediaBrowser.Controller.Session;
+
+/// <summary>
+/// A session held by one instance, as the other instances see it.
+/// </summary>
+public sealed class SessionDirectoryEntry
+{
+    /// <summary>
+    /// Gets or sets the identity of the instance holding the connection.
+    /// </summary>
+    public string OwnerPod { get; set; } = string.Empty;
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    public bool HoldsConnection { get; set; }
+
+    /// <summary>
+    /// Gets or sets the session as its owner last rendered it.
+    /// </summary>
+    public SessionInfoDto? Session { get; set; }
+}

MediaBrowser.Controller/Session/SessionDirectoryOptions.cs

@@ -0,0 +1,28 @@
+namespace MediaBrowser.Controller.Session;
+
+/// <summary>
+/// Configuration options for the session directory and the cross-instance bus that goes with it.
+/// </summary>
+public sealed class SessionDirectoryOptions
+{
+    /// <summary>
+    /// The configuration section these options bind from.
+    /// </summary>
+    public const string ConfigurationSection = "Jellyfin:SessionDirectory";
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    public int EntryTtlSeconds { get; set; } = 60;
+
+    /// <summary>
+    /// Gets or sets how often in seconds an instance republishes the sessions it holds.
+    /// </summary>
+    public int RefreshIntervalSeconds { get; set; } = 20;
+
+    /// <summary>
+    /// Gets or sets how long in seconds a single directory operation may take before it is abandoned.
+    /// </summary>
+    public int OperationTimeoutSeconds { get; set; } = 5;
+}

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
 

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<IServerApplicationHost>(),
             Mock.Of<IDeviceManager>(),
             Mock.Of<IMediaSourceManager>(),
-            Mock.Of<IHostApplicationLifetime>());
+            Mock.Of<IHostApplicationLifetime>(),
+            NullSessionDirectory.Instance,
+            NullPodMessageBus.Instance,
+            Options.Create(new SessionDirectoryOptions()));
         var session = await sessionManager.LogSessionActivity(
             "Test Client",
             "1.0.0",

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<IServerApplicationHost>(),
             Mock.Of<IDeviceManager>(),
             Mock.Of<IMediaSourceManager>(),
-            Mock.Of<IHostApplicationLifetime>());
+            Mock.Of<IHostApplicationLifetime>(),
+            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<IServerApplicationHost>(),
             Mock.Of<IDeviceManager>(),
             Mock.Of<IMediaSourceManager>(),
-            Mock.Of<IHostApplicationLifetime>());
+            Mock.Of<IHostApplicationLifetime>(),
+            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<SecurityException>(() => 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<SecurityException>(() => sessionManager.AddAdditionalUser(attackerSession.Id, attackerSession.Id, victim.Id));
+        await Assert.ThrowsAsync<SecurityException>(() => 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<SecurityException>(() => sessionManager.RemoveAdditionalUser(attackerSession.Id, victimSession.Id, attacker.Id));
+        await Assert.ThrowsAsync<SecurityException>(() => 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<SecurityException>(() => sessionManager.ReportCapabilities(attackerSession.Id, victimSession.Id, new ClientCapabilities()));
+        await Assert.ThrowsAsync<SecurityException>(() => 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<IServerApplicationHost>(),
             Mock.Of<IDeviceManager>(),
             Mock.Of<IMediaSourceManager>(),
-            Mock.Of<IHostApplicationLifetime>());
+            Mock.Of<IHostApplicationLifetime>(),
+            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<SessionInfo> 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<T>(SessionMessageType name, Guid messageId, T data, CancellationToken cancellationToken)
+            => Task.CompletedTask;
+    }
 }

tests/Jellyfin.Server.Tests/HighAvailability/RedisTestServer.cs

@@ -6,9 +6,9 @@ using Testcontainers.Redis;
 namespace Jellyfin.Server.Tests.HighAvailability;
 
 /// <summary>
-/// Hands out a Redis server for the tests that need one. A server named by <c>JELLYFIN_TEST_REDIS</c> 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
+/// <c>JELLYFIN_TEST_REDIS</c> 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.
 /// </summary>
 public sealed class RedisTestServer : IAsyncDisposable
 {
@@ -31,7 +31,7 @@ public sealed class RedisTestServer : IAsyncDisposable
     public string ConnectionString { get; }
 
     /// <summary>
-    /// 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.
     /// </summary>
     /// <returns>The running server.</returns>
     public static async Task<RedisTestServer> 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());

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;
+
+/// <summary>
+/// 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.
+/// </summary>
+public static class SessionDirectoryRegistrationTests
+{
+    [Fact]
+    public static void AddSessionDirectory_WithoutARedisConnection_RegistersNeitherHalf()
+    {
+        var services = Build(configured: false);
+
+        Assert.IsType<NullSessionDirectory>(services.GetRequiredService<ISessionDirectory>());
+        Assert.IsType<NullPodMessageBus>(services.GetRequiredService<IPodMessageBus>());
+    }
+
+    [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<IConnectionMultiplexer>();
+        redis.Setup(i => i.GetDatabase(It.IsAny<int>(), It.IsAny<object>())).Returns(Mock.Of<IDatabase>());
+        redis.Setup(i => i.GetSubscriber(It.IsAny<object>())).Throws(new RedisConnectionException(ConnectionFailureType.UnableToConnect, "unreachable"));
+
+        var services = Build(configured: true, redis.Object);
+
+        Assert.IsType<NullSessionDirectory>(services.GetRequiredService<ISessionDirectory>());
+        Assert.IsType<NullPodMessageBus>(services.GetRequiredService<IPodMessageBus>());
+    }
+
+    private static ServiceProvider Build(bool configured, IConnectionMultiplexer? redis = null)
+    {
+        var settings = new Dictionary<string, string?>();
+        if (configured)
+        {
+            settings[TranscodeStoreOptions.RedisConnectionStringKey] = "127.0.0.1:6379";
+        }
+
+        var configuration = new ConfigurationBuilder().AddInMemoryCollection(settings).Build();
+        var serviceCollection = new ServiceCollection().AddLogging();
+
+        if (redis is not null)
+        {
+            serviceCollection.AddSingleton(redis);
+        }
+
+        return serviceCollection
+            .AddSessionDirectory(configuration, NullLogger.Instance)
+            .BuildServiceProvider();
+    }
+}

tests/Jellyfin.Server.Tests/HighAvailability/SessionDirectoryReplicaTests.cs

@@ -0,0 +1,997 @@
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Globalization;
+using System.Linq;
+using System.Reflection;
+using System.Text.Json;
+using System.Threading;
+using System.Threading.Tasks;
+using Jellyfin.Data;
+using Jellyfin.Data.Enums;
+using Jellyfin.Database.Implementations;
+using Jellyfin.Database.Implementations.DbConfiguration;
+using Jellyfin.Database.Implementations.Entities;
+using Jellyfin.Database.Implementations.Enums;
+using Jellyfin.Database.Implementations.Locking;
+using Jellyfin.Database.Providers.PostgreSQL;
+using Jellyfin.Server.Implementations.Devices;
+using Jellyfin.Server.Tests.Migrations;
+using MediaBrowser.Common.Extensions;
+using MediaBrowser.Controller;
+using MediaBrowser.Controller.Configuration;
+using MediaBrowser.Controller.Drawing;
+using MediaBrowser.Controller.Dto;
+using MediaBrowser.Controller.Entities;
+using MediaBrowser.Controller.Entities.Movies;
+using MediaBrowser.Controller.Events;
+using MediaBrowser.Controller.Library;
+using MediaBrowser.Controller.Session;
+using MediaBrowser.Model.Configuration;
+using MediaBrowser.Model.Dto;
+using MediaBrowser.Model.Entities;
+using MediaBrowser.Model.Session;
+using MediaBrowser.Model.SyncPlay;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+using Moq;
+using Npgsql;
+using StackExchange.Redis;
+using Xunit;
+using RedisPodMessageBus = Emby.Server.Implementations.Session.RedisPodMessageBus;
+using RedisSessionDirectory = Emby.Server.Implementations.Session.RedisSessionDirectory;
+using SessionManager = Emby.Server.Implementations.Session.SessionManager;
+
+namespace Jellyfin.Server.Tests.HighAvailability;
+
+/// <summary>
+/// Two independently constructed <see cref="SessionManager"/> instances over one PostgreSQL database and
+/// one valkey are the in-process stand-in for two replicas without sticky sessions: a session either of
+/// them holds has to be visible to, and controllable from, the other.
+/// </summary>
+[Trait("Category", "RequiresDocker")]
+public sealed class SessionDirectoryReplicaTests : IAsyncLifetime
+{
+    private const string AppName = "Jellyfin Web";
+    private const string AppVersion = "1.0.0";
+    private const string DeviceName = "Living Room TV";
+    private const string RemoteEndPoint = "127.0.0.1";
+
+    private PostgreSqlTestServer _postgres = null!;
+    private RedisTestServer _redis = null!;
+    private NpgsqlDataSource _dataSource = null!;
+    private IConnectionMultiplexer _connection = null!;
+    private ISessionDirectory _directory = null!;
+    private User _user = null!;
+    private User _guest = null!;
+
+    /// <inheritdoc/>
+    public async ValueTask InitializeAsync()
+    {
+        _postgres = await PostgreSqlTestServer.StartAsync();
+        _redis = await RedisTestServer.StartAsync();
+        _connection = await _redis.ConnectAsync();
+        _directory = new RedisSessionDirectory(
+            _connection,
+            Options.Create(new SessionDirectoryOptions()),
+            NullLogger<RedisSessionDirectory>.Instance);
+
+        var connectionString = await _postgres.CreateDatabaseAsync("session_directory", CancellationToken.None);
+        _dataSource = new NpgsqlDataSourceBuilder(connectionString).Build();
+
+        var context = CreateContext(_dataSource);
+        await using (context.ConfigureAwait(false))
+        {
+            await context.Database.EnsureCreatedAsync(CancellationToken.None);
+
+            _user = new User("replica-user", "provider", "provider");
+            _guest = new User("replica-guest", "provider", "provider");
+            context.Users.Add(_user);
+            context.Users.Add(_guest);
+            await context.SaveChangesAsync(CancellationToken.None);
+        }
+    }
+
+    /// <inheritdoc/>
+    public async ValueTask DisposeAsync()
+    {
+        await _connection.DisposeAsync();
+        await _dataSource.DisposeAsync();
+        await _redis.DisposeAsync();
+        await _postgres.DisposeAsync();
+    }
+
+    /// <summary>
+    /// Half the active playback is invisible when the session list only reports what the replica serving
+    /// the request happens to hold, so a session registered on one replica has to appear on the other.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    [Fact]
+    public async Task SessionRegisteredOnOneReplica_IsListedByAnother()
+    {
+        var cancellationToken = TestContext.Current.CancellationToken;
+        await using var replicaA = CreateReplica("pod-a");
+        await using var replicaB = CreateReplica("pod-b");
+
+        var session = await Request(replicaA, "device-listed");
+
+        var listedByB = await replicaB.GetSessions(_user.Id, null, null, null, false, cancellationToken);
+        var listedByA = await replicaA.GetSessions(_user.Id, null, null, null, false, cancellationToken);
+
+        Assert.Contains(listedByB, i => string.Equals(i.Id, session.Id, StringComparison.Ordinal));
+        Assert.Contains(listedByA, i => string.Equals(i.Id, session.Id, StringComparison.Ordinal));
+
+        // The session is reported once, not once per replica that can see it.
+        Assert.Single(listedByB, i => string.Equals(i.Id, session.Id, StringComparison.Ordinal));
+    }
+
+    /// <summary>
+    /// The deployment has no sticky sessions, so one device's requests land on either replica while its
+    /// websocket stays on one of them. Ownership has to follow the connection rather than the last
+    /// request served, or the directory names the wrong replica, the session list doubles up and remote
+    /// control is delivered to a replica with nothing to deliver it to.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    [Fact]
+    public async Task RequestsAlternatingBetweenReplicas_KeepOwnershipWithTheConnection()
+    {
+        var cancellationToken = TestContext.Current.CancellationToken;
+        var options = new SessionDirectoryOptions { EntryTtlSeconds = 60, RefreshIntervalSeconds = 2 };
+        await using var replicaA = CreateReplica("pod-a", options);
+        await using var replicaB = CreateReplica("pod-b", options);
+
+        // The device is first seen by the replica that will not hold its websocket.
+        await Request(replicaB, "device-roaming");
+
+        var session = await Request(replicaA, "device-roaming");
+        var controller = new RecordingSessionController();
+        session.AddController(controller);
+        await replicaA.OnSessionControllerConnected(session);
+
+        // The load balancer keeps handing the device's requests to whichever replica it likes, and the
+        // replica without the websocket must never take the session from the one that has it.
+        for (var i = 0; i < 8; i++)
+        {
+            await Request(replicaB, "device-roaming");
+            await Task.Delay(250, cancellationToken);
+
+            var entry = await _directory.GetAsync(session.Id, cancellationToken);
+            Assert.NotNull(entry);
+            Assert.Equal("pod-a", entry.OwnerPod);
+            Assert.True(entry.HoldsConnection);
+
+            await Request(replicaA, "device-roaming");
+            await Task.Delay(50, cancellationToken);
+        }
+
+        var listedByA = await replicaA.GetSessions(_user.Id, null, null, null, false, cancellationToken);
+        var listedByB = await replicaB.GetSessions(_user.Id, null, null, null, false, cancellationToken);
+
+        Assert.Single(listedByA, i => string.Equals(i.Id, session.Id, StringComparison.Ordinal));
+        Assert.Single(listedByB, i => string.Equals(i.Id, session.Id, StringComparison.Ordinal));
+
+        await replicaB.SendMessageCommand(
+            string.Empty,
+            session.Id,
+            new MessageCommand { Header = "Header", Text = "Dinner is ready" },
+            cancellationToken);
+
+        var (messageType, data) = await controller.WaitForMessageAsync(cancellationToken);
+        Assert.Equal(SessionMessageType.GeneralCommand, messageType);
+        Assert.Contains("Dinner is ready", data, StringComparison.Ordinal);
+    }
+
+    /// <summary>
+    /// Remote control and "send message to session" used to succeed and do nothing when the device is
+    /// connected to another replica; the message has to reach the connection wherever it is held.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    [Fact]
+    public async Task MessageSentOnOneReplica_ReachesTheConnectionHeldByAnother()
+    {
+        var cancellationToken = TestContext.Current.CancellationToken;
+        await using var replicaA = CreateReplica("pod-a");
+        await using var replicaB = CreateReplica("pod-b");
+
+        var session = await Request(replicaA, "device-controlled");
+        var controller = new RecordingSessionController();
+        session.AddController(controller);
+        await replicaA.OnSessionControllerConnected(session);
+
+        await replicaB.SendMessageCommand(
+            string.Empty,
+            session.Id,
+            new MessageCommand { Header = "Header", Text = "Dinner is ready" },
+            cancellationToken);
+
+        var (messageType, data) = await controller.WaitForMessageAsync(cancellationToken);
+        Assert.Equal(SessionMessageType.GeneralCommand, messageType);
+        Assert.Contains("Dinner is ready", data, StringComparison.Ordinal);
+    }
+
+    /// <summary>
+    /// An entry outlives the replica that wrote it by up to its expiry, and a command routed into that
+    /// gap reaches nobody. Reporting it as delivered is the failure this directory exists to remove.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    [Fact]
+    public async Task MessageRoutedToADeadOwner_IsReportedAsUndelivered()
+    {
+        var cancellationToken = TestContext.Current.CancellationToken;
+        await using var replicaA = CreateReplica("pod-a");
+        await using var replicaB = CreateReplica("pod-b");
+
+        var session = await Request(replicaA, "device-dead-owner");
+        session.AddController(new RecordingSessionController());
+        await replicaA.OnSessionControllerConnected(session);
+
+        var entry = await _directory.GetAsync(session.Id, cancellationToken);
+        Assert.NotNull(entry);
+
+        // A replica that is no longer listening, holding the entry until it expires.
+        entry.OwnerPod = "pod-gone";
+        Assert.True(await _directory.PublishAsync(entry, DateTime.UtcNow.Ticks, cancellationToken));
+
+        await Assert.ThrowsAsync<ResourceNotFoundException>(
+            () => replicaB.SendMessageCommand(
+                string.Empty,
+                session.Id,
+                new MessageCommand { Header = "Header", Text = "Dinner is ready" },
+                cancellationToken));
+    }
+
+    /// <summary>
+    /// The owner's entry is only as fresh as its last refresh, so a websocket that closes in between
+    /// leaves an entry claiming a connection that is gone. The command has to be reported undelivered,
+    /// which only the replica that would have written it to the socket can say.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    [Fact]
+    public async Task MessageRoutedToAnOwnerWhoseSocketDied_IsReportedAsUndelivered()
+    {
+        var cancellationToken = TestContext.Current.CancellationToken;
+        await using var replicaA = CreateReplica("pod-a");
+        await using var replicaB = CreateReplica("pod-b");
+
+        var session = await Request(replicaA, "device-dead-socket");
+        var controller = new RecordingSessionController();
+        session.AddController(controller);
+        await replicaA.OnSessionControllerConnected(session);
+
+        // The socket closes. Nothing rewrites the entry: it still names pod-a and still says the
+        // connection is held, exactly as it does for the rest of the refresh interval.
+        controller.IsSessionActive = false;
+
+        var entry = await _directory.GetAsync(session.Id, cancellationToken);
+        Assert.NotNull(entry);
+        Assert.Equal("pod-a", entry.OwnerPod);
+        Assert.True(entry.HoldsConnection);
+
+        await Assert.ThrowsAsync<ResourceNotFoundException>(
+            () => replicaB.SendMessageCommand(
+                string.Empty,
+                session.Id,
+                new MessageCommand { Header = "Header", Text = "Dinner is ready" },
+                cancellationToken));
+    }
+
+    /// <summary>
+    /// A device reconnecting lands on either replica, so both can hold a live connection for the same
+    /// deterministic session id at once. Exactly one of them owns the entry, and it stays that one.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    [Fact]
+    public async Task BothReplicasHoldingAConnection_AgreeOnOneOwner()
+    {
+        var cancellationToken = TestContext.Current.CancellationToken;
+        var options = new SessionDirectoryOptions { EntryTtlSeconds = 60, RefreshIntervalSeconds = 1 };
+        await using var replicaA = CreateReplica("pod-a", options);
+        await using var replicaB = CreateReplica("pod-b", options);
+
+        var sessionA = await Request(replicaA, "device-two-sockets");
+        sessionA.AddController(new RecordingSessionController());
+        await replicaA.OnSessionControllerConnected(sessionA);
+
+        var sessionB = await Request(replicaB, "device-two-sockets");
+        sessionB.AddController(new RecordingSessionController());
+        await replicaB.OnSessionControllerConnected(sessionB);
+
+        Assert.Equal(sessionA.Id, sessionB.Id);
+
+        // The later connection owns the session; both replicas keep republishing theirs.
+        for (var i = 0; i < 8; i++)
+        {
+            await Task.Delay(500, cancellationToken);
+
+            var entry = await _directory.GetAsync(sessionA.Id, cancellationToken);
+            Assert.NotNull(entry);
+            Assert.Equal("pod-b", entry.OwnerPod);
+            Assert.True(entry.HoldsConnection);
+        }
+
+        var listedByA = await replicaA.GetSessions(_user.Id, null, null, null, false, cancellationToken);
+        Assert.Single(listedByA, i => string.Equals(i.Id, sessionA.Id, StringComparison.Ordinal));
+    }
+
+    /// <summary>
+    /// A session with no websocket is claimed with a zero epoch by every replica that serves a request
+    /// for it. The first claim has to stand, or the listed session flips between two partial copies.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    [Fact]
+    public async Task TwoReplicasWithoutAConnection_DoNotTakeTheSessionFromEachOther()
+    {
+        var cancellationToken = TestContext.Current.CancellationToken;
+        var options = new SessionDirectoryOptions { EntryTtlSeconds = 60, RefreshIntervalSeconds = 1 };
+        await using var replicaA = CreateReplica("pod-a", options);
+        await using var replicaB = CreateReplica("pod-b", options);
+
+        var session = await Request(replicaA, "device-no-socket");
+        await Request(replicaB, "device-no-socket");
+
+        var claimed = await _directory.GetAsync(session.Id, cancellationToken);
+        Assert.NotNull(claimed);
+        Assert.False(claimed.HoldsConnection);
+
+        var owner = claimed.OwnerPod;
+
+        Assert.False(await _directory.PublishAsync(
+            new SessionDirectoryEntry
+            {
+                OwnerPod = owner == "pod-a" ? "pod-b" : "pod-a",
+                HoldsConnection = false,
+                Session = claimed.Session
+            },
+            0,
+            cancellationToken));
+
+        for (var i = 0; i < 6; i++)
+        {
+            await Request(replicaB, "device-no-socket");
+            await Task.Delay(400, cancellationToken);
+            await Request(replicaA, "device-no-socket");
+
+            var entry = await _directory.GetAsync(session.Id, cancellationToken);
+            Assert.NotNull(entry);
+            Assert.Equal(owner, entry.OwnerPod);
+        }
+    }
+
+    /// <summary>
+    /// Without sticky sessions a playback report lands on either replica while the websocket stays on
+    /// one. The report belongs to the replica everyone else is shown, so it is applied there and the
+    /// session reads as playing from every replica rather than idle on all of them.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    [Fact]
+    public async Task PlaybackReportedToTheNonOwner_IsVisibleFromBothReplicas()
+    {
+        var cancellationToken = TestContext.Current.CancellationToken;
+        await using var replicaA = CreateReplica("pod-a");
+        await using var replicaB = CreateReplica("pod-b");
+
+        var session = await Request(replicaA, "device-playing");
+        session.AddController(new RecordingSessionController());
+        await replicaA.OnSessionControllerConnected(session);
+
+        // The load balancer hands the playback report to the replica without the websocket.
+        await Request(replicaB, "device-playing");
+        await replicaB.OnPlaybackStart(new PlaybackStartInfo
+        {
+            SessionId = session.Id,
+            Item = new BaseItemDto { Id = Guid.NewGuid(), Name = "Routed Movie" },
+            PositionTicks = 0
+        });
+
+        Assert.Equal("Routed Movie", session.NowPlayingItem?.Name);
+
+        var listedByA = await replicaA.GetSessions(_user.Id, null, null, null, false, cancellationToken);
+        var listedByB = await replicaB.GetSessions(_user.Id, null, null, null, false, cancellationToken);
+
+        Assert.Equal("Routed Movie", Single(listedByA, session.Id).NowPlayingItem?.Name);
+        Assert.Equal("Routed Movie", Single(listedByB, session.Id).NowPlayingItem?.Name);
+
+        await replicaB.OnPlaybackStopped(new PlaybackStopInfo { SessionId = session.Id, PositionTicks = 1 });
+
+        Assert.Null(session.NowPlayingItem);
+        Assert.Null(Single(await replicaB.GetSessions(_user.Id, null, null, null, false, cancellationToken), session.Id).NowPlayingItem);
+    }
+
+    /// <summary>
+    /// A report that cannot be handed to the owner is still applied here, so an unreachable owner is
+    /// never worse than the single-instance behaviour of keeping the state on the replica that served
+    /// the request.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    [Fact]
+    public async Task PlaybackReportedWithTheOwnerUnreachable_IsAppliedLocally()
+    {
+        var cancellationToken = TestContext.Current.CancellationToken;
+        await using var replicaA = CreateReplica("pod-a");
+        await using var replicaB = CreateReplica("pod-b");
+
+        var session = await Request(replicaA, "device-orphaned");
+        session.AddController(new RecordingSessionController());
+        await replicaA.OnSessionControllerConnected(session);
+
+        var local = await Request(replicaB, "device-orphaned");
+
+        var entry = await _directory.GetAsync(session.Id, cancellationToken);
+        Assert.NotNull(entry);
+        entry.OwnerPod = "pod-gone";
+        Assert.True(await _directory.PublishAsync(entry, long.MaxValue, cancellationToken));
+
+        await replicaB.OnPlaybackStart(new PlaybackStartInfo
+        {
+            SessionId = session.Id,
+            Item = new BaseItemDto { Id = Guid.NewGuid(), Name = "Orphaned Movie" },
+            PositionTicks = 0
+        });
+
+        Assert.Equal("Orphaned Movie", local.NowPlayingItem?.Name);
+    }
+
+    /// <summary>
+    /// A directory that cannot be read says nothing about where a session is. Treating the failure as
+    /// "no such entry" hands the command to a local copy with no connection, which reports success and
+    /// delivers nothing.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    [Fact]
+    public async Task DirectoryReadFailingDuringRemoteControl_DoesNotSilentlyDoNothing()
+    {
+        var cancellationToken = TestContext.Current.CancellationToken;
+        var failing = new FailableSessionDirectory(new RedisSessionDirectory(
+            _connection,
+            Options.Create(new SessionDirectoryOptions()),
+            NullLogger<RedisSessionDirectory>.Instance));
+
+        await using var replicaA = CreateReplica("pod-a");
+        await using var replicaB = CreateReplica("pod-b", failing);
+
+        var session = await Request(replicaA, "device-unreadable");
+        session.AddController(new RecordingSessionController());
+        await replicaA.OnSessionControllerConnected(session);
+
+        // The replica serving the request holds a copy of the session, and only a copy.
+        await Request(replicaB, "device-unreadable");
+
+        failing.FailReads = true;
+
+        await Assert.ThrowsAsync<RedisTimeoutException>(
+            () => replicaB.SendMessageCommand(
+                string.Empty,
+                session.Id,
+                new MessageCommand { Header = "Header", Text = "Dinner is ready" },
+                cancellationToken));
+    }
+
+    /// <summary>
+    /// Ownership is decided by comparing the two replicas' connection epochs, so the epochs cannot come
+    /// from the replicas' own clocks: a lagging clock would keep a genuinely newer connection from ever
+    /// taking the session. They are handed out per session by the shared store instead.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    [Fact]
+    public async Task ConnectionEpochs_AreHandedOutByTheStore()
+    {
+        var cancellationToken = TestContext.Current.CancellationToken;
+        await using var replicaA = CreateReplica("pod-a");
+
+        var session = await Request(replicaA, "device-epoch");
+        session.AddController(new RecordingSessionController());
+        await replicaA.OnSessionControllerConnected(session);
+
+        var recorded = await ReadOwnerEpoch(session.Id, cancellationToken);
+        var allocated = await _directory.AllocateConnectionEpochAsync(session.Id, cancellationToken);
+        var other = await _directory.AllocateConnectionEpochAsync(session.Id + "-other", cancellationToken);
+
+        // A per-session counter the store hands out in order, orders of magnitude below any tick count,
+        // so it cannot be a reading of a replica's clock.
+        Assert.True(recorded > 0);
+        Assert.True(allocated > recorded);
+        Assert.True(other > 0);
+        Assert.True(other < await _directory.AllocateConnectionEpochAsync(session.Id + "-other", cancellationToken));
+        Assert.True(allocated < TimeSpan.TicksPerSecond);
+    }
+
+    /// <summary>
+    /// SyncPlay groups are still instance-local, so the replica serving the request has to notice that
+    /// its copy of the session has no connection. Holding a copy is not holding the connection, and a
+    /// command handed to a copy would be dropped without a word.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    [Fact]
+    public async Task SyncPlayCommandOnTheReplicaWithoutTheConnection_IsSkippedAndLogged()
+    {
+        var cancellationToken = TestContext.Current.CancellationToken;
+        var logger = new CapturingLogger<SessionManager>();
+        await using var replicaA = CreateReplica("pod-a");
+        await using var replicaB = CreateReplica("pod-b", logger: logger);
+
+        var session = await Request(replicaA, "device-syncplay");
+        var controller = new RecordingSessionController();
+        session.AddController(controller);
+        await replicaA.OnSessionControllerConnected(session);
+
+        await Request(replicaB, "device-syncplay");
+
+        await replicaB.SendSyncPlayCommand(
+            session.Id,
+            new SendCommand(Guid.NewGuid(), Guid.NewGuid(), DateTime.UtcNow, SendCommandType.Pause, 0, DateTime.UtcNow),
+            cancellationToken);
+
+        Assert.Contains(logger.Messages, i => i.Contains("SyncPlay command", StringComparison.Ordinal) && i.Contains(session.Id, StringComparison.Ordinal));
+        Assert.False(controller.HasMessage);
+    }
+
+    /// <summary>
+    /// Both replicas keep a copy of a session whose requests they have served, so the replica ending its
+    /// own copy must not erase the entry of the one still holding the connection.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    [Fact]
+    public async Task ReplicaEndingItsOwnCopy_LeavesTheOwnersEntryAlone()
+    {
+        var cancellationToken = TestContext.Current.CancellationToken;
+        await using var replicaA = CreateReplica("pod-a");
+        await using var replicaB = CreateReplica("pod-b");
+
+        var session = await Request(replicaA, "device-shared-end");
+        session.AddController(new RecordingSessionController());
+        await replicaA.OnSessionControllerConnected(session);
+
+        await Request(replicaB, "device-shared-end");
+        await replicaB.ReportSessionEnded(session.Id);
+
+        var entry = await _directory.GetAsync(session.Id, cancellationToken);
+        Assert.NotNull(entry);
+        Assert.Equal("pod-a", entry.OwnerPod);
+
+        await replicaA.ReportSessionEnded(session.Id);
+
+        Assert.Null(await _directory.GetAsync(session.Id, cancellationToken));
+    }
+
+    /// <summary>
+    /// The session list now shows sessions from every replica, so an action offered against one of them
+    /// has to reach it rather than fail as missing on the replica serving the request.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    [Fact]
+    public async Task AdditionalUserAddedOnOneReplica_ReachesTheOwner()
+    {
+        var cancellationToken = TestContext.Current.CancellationToken;
+        await using var replicaA = CreateReplica("pod-a");
+        await using var replicaB = CreateReplica("pod-b");
+
+        var session = await Request(replicaA, "device-additional-user");
+        session.AddController(new RecordingSessionController());
+        await replicaA.OnSessionControllerConnected(session);
+
+        await replicaB.AddAdditionalUser(string.Empty, session.Id, _guest.Id);
+
+        await WaitUntil(() => session.AdditionalUsers.Any(i => i.UserId.Equals(_guest.Id)), cancellationToken);
+
+        await replicaB.RemoveAdditionalUser(string.Empty, session.Id, _guest.Id);
+
+        await WaitUntil(() => !session.AdditionalUsers.Any(i => i.UserId.Equals(_guest.Id)), cancellationToken);
+    }
+
+    /// <summary>
+    /// The idle sweep runs on every replica but stops playback and rewrites the resume position. A
+    /// replica holding a copy it does not own sees a check-in that froze when reports started routing
+    /// away, so an unguarded sweep would stop a film the other replica is still playing.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    [Fact]
+    public async Task IdleSweepOnANonOwner_LeavesTheOwnersPlaybackAlone()
+    {
+        var cancellationToken = TestContext.Current.CancellationToken;
+        var stopped = new TaskCompletionSource<PlaybackStopEventArgs>(TaskCreationOptions.RunContinuationsAsynchronously);
+        var events = new Mock<IEventManager>();
+        events.Setup(i => i.PublishAsync(It.IsAny<PlaybackStopEventArgs>()))
+            .Callback<PlaybackStopEventArgs>(args => stopped.TrySetResult(args))
+            .Returns(Task.CompletedTask);
+
+        var movie = new Movie { Id = Guid.NewGuid(), Name = "Live Movie" };
+        var libraryManager = new Mock<ILibraryManager>();
+        libraryManager.Setup(i => i.GetItemById(movie.Id)).Returns(movie);
+
+        var userDataManager = new Mock<IUserDataManager>();
+        userDataManager.Setup(i => i.GetUserData(It.IsAny<User>(), It.IsAny<BaseItem>())).Returns(new UserItemData { Key = "test" });
+
+        await using var replicaA = CreateReplica(
+            "pod-a",
+            userDataManager: userDataManager.Object,
+            libraryManager: libraryManager.Object,
+            eventManager: events.Object);
+        await using var replicaB = CreateReplica("pod-b");
+
+        var owned = await Request(replicaA, "device-idle-sweep");
+        owned.AddController(new RecordingSessionController());
+        await replicaA.OnSessionControllerConnected(owned);
+
+        await replicaA.OnPlaybackStart(new PlaybackStartInfo
+        {
+            SessionId = owned.Id,
+            ItemId = movie.Id,
+            MediaSourceId = movie.Id.ToString("N", CultureInfo.InvariantCulture),
+            Item = new BaseItemDto { Id = movie.Id, Name = movie.Name },
+            PositionTicks = 0
+        });
+
+        // The copy a non-owner is left with: it still remembers a now playing item, and its check-in
+        // froze the moment reports started going to the owner instead.
+        var copy = await Request(replicaB, "device-idle-sweep");
+        copy.NowPlayingItem = new BaseItemDto { Id = movie.Id, Name = movie.Name };
+        copy.StartAutomaticProgress(new PlaybackProgressInfo { IsPaused = true, PositionTicks = 123456789 });
+        copy.StopAutomaticProgress();
+        copy.LastPlaybackCheckIn = DateTime.UtcNow.AddHours(-1);
+
+        InvokeSweep(replicaB, "CheckForIdlePlayback");
+
+        await Assert.ThrowsAsync<TimeoutException>(() => stopped.Task.WaitAsync(TimeSpan.FromSeconds(5), cancellationToken));
+
+        Assert.Equal(movie.Name, owned.NowPlayingItem?.Name);
+        Assert.Equal(movie.Name, Single(await replicaB.GetSessions(_user.Id, null, null, null, false, cancellationToken), owned.Id).NowPlayingItem?.Name);
+        userDataManager.Verify(
+            i => i.SaveUserData(It.IsAny<User>(), It.IsAny<BaseItem>(), It.IsAny<UserItemData>(), UserDataSaveReason.PlaybackFinished, It.IsAny<CancellationToken>()),
+            Times.Never);
+
+        // The same sweep on the owner does stop it, so the assertions above are not vacuous.
+        owned.LastPlaybackCheckIn = DateTime.UtcNow.AddHours(-1);
+        InvokeSweep(replicaA, "CheckForIdlePlayback");
+
+        await stopped.Task.WaitAsync(TimeSpan.FromSeconds(30), cancellationToken);
+
+        Assert.Null(owned.NowPlayingItem);
+    }
+
+    /// <summary>
+    /// The inactive sweep decides from the local copy's paused state and then writes a real Stop to
+    /// whichever replica holds the connection. A replica that only holds a copy has no business
+    /// stopping the session the other one is serving.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    [Fact]
+    public async Task InactiveSweepOnANonOwner_DoesNotStopTheOwnersSession()
+    {
+        var cancellationToken = TestContext.Current.CancellationToken;
+        var configuration = new ServerConfiguration { InactiveSessionThreshold = 1 };
+
+        await using var replicaA = CreateReplica("pod-a", configuration: configuration);
+        await using var replicaB = CreateReplica("pod-b", configuration: configuration);
+
+        var owned = await Request(replicaA, "device-inactive-sweep");
+        var controller = new RecordingSessionController();
+        owned.AddController(controller);
+        await replicaA.OnSessionControllerConnected(owned);
+
+        owned.NowPlayingItem = new BaseItemDto { Id = Guid.NewGuid(), Name = "Paused Movie" };
+        owned.PlayState.IsPaused = true;
+        owned.LastPausedDate = DateTime.UtcNow.AddHours(-1);
+
+        var copy = await Request(replicaB, "device-inactive-sweep");
+        copy.NowPlayingItem = owned.NowPlayingItem;
+        copy.PlayState.IsPaused = true;
+        copy.LastPausedDate = DateTime.UtcNow.AddHours(-1);
+
+        InvokeSweep(replicaB, "CheckForInactiveSteams");
+
+        await Task.Delay(TimeSpan.FromSeconds(3), cancellationToken);
+
+        Assert.False(controller.HasMessage);
+
+        // The same sweep on the owner does write the Stop, so the assertion above is not vacuous.
+        InvokeSweep(replicaA, "CheckForInactiveSteams");
+
+        var (messageType, _) = await controller.WaitForMessageAsync(cancellationToken);
+        Assert.Equal(SessionMessageType.Playstate, messageType);
+    }
+
+    /// <summary>
+    /// Without sticky sessions a device posts its capabilities to either replica, and a session is only
+    /// offered for remote control while the replica that publishes it knows they support media control.
+    /// A report kept by the replica that served it drops the device from the cast list everywhere.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    [Fact]
+    public async Task CapabilitiesReportedToTheNonOwner_KeepTheDeviceControllable()
+    {
+        var cancellationToken = TestContext.Current.CancellationToken;
+        _user.SetPermission(PermissionKind.EnableAllDevices, true);
+
+        await using var replicaA = CreateReplica("pod-a");
+        await using var replicaB = CreateReplica("pod-b");
+
+        var session = await Request(replicaA, "device-capabilities");
+        session.AddController(new RecordingSessionController());
+        await replicaA.OnSessionControllerConnected(session);
+
+        // The web client posts its capabilities to whichever replica the load balancer picked.
+        await Request(replicaB, "device-capabilities");
+        await replicaB.ReportCapabilities(string.Empty, session.Id, new ClientCapabilities
+        {
+            PlayableMediaTypes = [MediaType.Video],
+            SupportedCommands = [GeneralCommandType.DisplayMessage],
+            SupportsMediaControl = true,
+            SupportsPersistentIdentifier = true
+        });
+
+        Assert.True(session.SupportsRemoteControl);
+
+        await WaitUntilAsync(
+            async () => (await ControllableIds(replicaB, cancellationToken)).Contains(session.Id),
+            cancellationToken);
+
+        Assert.Contains(session.Id, await ControllableIds(replicaA, cancellationToken));
+        Assert.Contains(session.Id, await ControllableIds(replicaB, cancellationToken));
+    }
+
+    /// <summary>
+    /// A replica that dies stops refreshing its entries, and the sessions it held have to leave the
+    /// directory rather than linger in every other replica's session list forever.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    [Fact]
+    public async Task SessionsOfAReplicaThatStopsRefreshing_LeaveTheDirectory()
+    {
+        var cancellationToken = TestContext.Current.CancellationToken;
+
+        // A never refreshes within the test, so it stands in for a replica that crashed.
+        await using var replicaA = CreateReplica("pod-a", new SessionDirectoryOptions { EntryTtlSeconds = 1, RefreshIntervalSeconds = 3600 });
+        await using var replicaB = CreateReplica("pod-b");
+
+        var session = await Request(replicaA, "device-expiring");
+
+        var listedWhileAlive = await replicaB.GetSessions(_user.Id, null, null, null, false, cancellationToken);
+        Assert.Contains(listedWhileAlive, i => string.Equals(i.Id, session.Id, StringComparison.Ordinal));
+
+        await Task.Delay(TimeSpan.FromSeconds(2), cancellationToken);
+
+        var listedAfterExpiry = await replicaB.GetSessions(_user.Id, null, null, null, false, cancellationToken);
+        Assert.DoesNotContain(listedAfterExpiry, i => string.Equals(i.Id, session.Id, StringComparison.Ordinal));
+    }
+
+    /// <summary>
+    /// A deployment without a shared store keeps the single-instance behaviour: nothing is published and
+    /// the other instance sees nothing.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    [Fact]
+    public async Task WithoutADirectory_ReplicasOnlyReportTheirOwnSessions()
+    {
+        var cancellationToken = TestContext.Current.CancellationToken;
+        await using var replicaA = CreateReplica("pod-a", directory: NullSessionDirectory.Instance, bus: NullPodMessageBus.Instance);
+        await using var replicaB = CreateReplica("pod-b", directory: NullSessionDirectory.Instance, bus: NullPodMessageBus.Instance);
+
+        var session = await Request(replicaA, "device-local");
+
+        var listedByB = await replicaB.GetSessions(_user.Id, null, null, null, false, cancellationToken);
+        Assert.DoesNotContain(listedByB, i => string.Equals(i.Id, session.Id, StringComparison.Ordinal));
+    }
+
+    private async Task<long> ReadOwnerEpoch(string sessionId, CancellationToken cancellationToken)
+    {
+        var raw = await _connection.GetDatabase().StringGetAsync("jellyfin:sessionowner:" + sessionId).WaitAsync(cancellationToken);
+
+        return long.Parse(raw.ToString().Split('|')[0], CultureInfo.InvariantCulture);
+    }
+
+    private static SessionInfoDto Single(IReadOnlyList<SessionInfoDto> sessions, string sessionId)
+        => Assert.Single(sessions, i => string.Equals(i.Id, sessionId, StringComparison.Ordinal));
+
+    private static async Task WaitUntil(Func<bool> condition, CancellationToken cancellationToken)
+    {
+        var deadline = DateTime.UtcNow.AddSeconds(10);
+        while (!condition())
+        {
+            Assert.True(DateTime.UtcNow < deadline, "The expected change never arrived.");
+            await Task.Delay(50, cancellationToken);
+        }
+    }
+
+    private static async Task WaitUntilAsync(Func<Task<bool>> condition, CancellationToken cancellationToken)
+    {
+        var deadline = DateTime.UtcNow.AddSeconds(30);
+        while (!await condition())
+        {
+            Assert.True(DateTime.UtcNow < deadline, "The expected change never arrived.");
+            await Task.Delay(100, cancellationToken);
+        }
+    }
+
+    private static void InvokeSweep(SessionManager replica, string name)
+        => typeof(SessionManager)
+            .GetMethod(name, BindingFlags.Instance | BindingFlags.NonPublic)!
+            .Invoke(replica, [null]);
+
+    private static JellyfinDbContext CreateContext(NpgsqlDataSource dataSource)
+    {
+        var optionsBuilder = new DbContextOptionsBuilder<JellyfinDbContext>();
+        var provider = new PostgreSqlDatabaseProvider(dataSource);
+        provider.Initialise(optionsBuilder, new DatabaseConfigurationOptions { DatabaseType = "PostgreSQL" });
+        return new JellyfinDbContext(
+            optionsBuilder.Options,
+            NullLogger<JellyfinDbContext>.Instance,
+            provider,
+            new NoLockBehavior(NullLogger<NoLockBehavior>.Instance));
+    }
+
+    private Task<SessionInfo> Request(SessionManager replica, string deviceId)
+        => replica.LogSessionActivity(AppName, AppVersion, deviceId, DeviceName, RemoteEndPoint, _user);
+
+    private async Task<IReadOnlyList<string>> ControllableIds(SessionManager replica, CancellationToken cancellationToken)
+        => (await replica.GetSessions(_user.Id, null, null, _user.Id, false, cancellationToken))
+            .Select(i => i.Id ?? string.Empty)
+            .ToList();
+
+    private SessionManager CreateReplica(
+        string podId,
+        SessionDirectoryOptions? options = null,
+        ILogger<SessionManager>? logger = null,
+        IUserDataManager? userDataManager = null,
+        ILibraryManager? libraryManager = null,
+        IEventManager? eventManager = null,
+        ServerConfiguration? configuration = null)
+    {
+        options ??= new SessionDirectoryOptions { EntryTtlSeconds = 60, RefreshIntervalSeconds = 3600 };
+
+        var directory = new RedisSessionDirectory(
+            _connection,
+            Options.Create(options),
+            NullLogger<RedisSessionDirectory>.Instance);
+
+        return CreateReplica(podId, options, directory, CreateBus(podId, options), logger, userDataManager, libraryManager, eventManager, configuration);
+    }
+
+    private SessionManager CreateReplica(string podId, ISessionDirectory directory)
+        => CreateReplica(podId, new SessionDirectoryOptions(), directory, CreateBus(podId, new SessionDirectoryOptions()), null);
+
+    private SessionManager CreateReplica(string podId, ISessionDirectory directory, IPodMessageBus bus)
+        => CreateReplica(podId, new SessionDirectoryOptions(), directory, bus, null);
+
+    private SessionManager CreateReplica(
+        string podId,
+        SessionDirectoryOptions options,
+        ISessionDirectory directory,
+        IPodMessageBus bus,
+        ILogger<SessionManager>? logger,
+        IUserDataManager? userDataManager = null,
+        ILibraryManager? libraryManager = null,
+        IEventManager? eventManager = null,
+        ServerConfiguration? configuration = null)
+    {
+        var userManager = new Mock<IUserManager>();
+        userManager.Setup(i => i.GetUserById(_user.Id)).Returns(_user);
+        userManager.Setup(i => i.GetUserById(_guest.Id)).Returns(_guest);
+
+        var appHost = new Mock<IServerApplicationHost>();
+        appHost.SetupGet(i => i.SystemId).Returns("server-" + podId);
+
+        var configurationManager = new Mock<IServerConfigurationManager>();
+        configurationManager.SetupGet(i => i.Configuration).Returns(configuration ?? new ServerConfiguration());
+
+        return new SessionManager(
+            logger ?? NullLogger<SessionManager>.Instance,
+            eventManager ?? Mock.Of<IEventManager>(),
+            userDataManager ?? Mock.Of<IUserDataManager>(),
+            configurationManager.Object,
+            libraryManager ?? Mock.Of<ILibraryManager>(),
+            userManager.Object,
+            Mock.Of<IMusicManager>(),
+            Mock.Of<IDtoService>(),
+            Mock.Of<IImageProcessor>(),
+            appHost.Object,
+            new DeviceManager(new DataSourceContextFactory(_dataSource), userManager.Object),
+            Mock.Of<IMediaSourceManager>(),
+            Mock.Of<IHostApplicationLifetime>(),
+            directory,
+            bus,
+            Options.Create(options));
+    }
+
+    private IPodMessageBus CreateBus(string podId, SessionDirectoryOptions options)
+        => new RedisPodMessageBus(
+            _connection,
+            Options.Create(options),
+            podId,
+            NullLogger<RedisPodMessageBus>.Instance);
+
+    /// <summary>
+    /// Hands every replica its own context over the one shared database, the way the pooled factory does
+    /// in the server.
+    /// </summary>
+    private sealed class DataSourceContextFactory : IDbContextFactory<JellyfinDbContext>
+    {
+        private readonly NpgsqlDataSource _dataSource;
+
+        public DataSourceContextFactory(NpgsqlDataSource dataSource)
+        {
+            _dataSource = dataSource;
+        }
+
+        public JellyfinDbContext CreateDbContext() => CreateContext(_dataSource);
+    }
+
+    /// <summary>
+    /// Keeps what a replica logged so a skipped route can be told apart from a silent drop.
+    /// </summary>
+    /// <typeparam name="T">The category the logger belongs to.</typeparam>
+    private sealed class CapturingLogger<T> : ILogger<T>
+    {
+        private readonly ConcurrentQueue<string> _messages = new();
+
+        public IEnumerable<string> Messages => _messages;
+
+        public IDisposable BeginScope<TState>(TState state)
+            where TState : notnull
+            => NullLogger.Instance.BeginScope(state);
+
+        public bool IsEnabled(LogLevel logLevel) => true;
+
+        public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func<TState, Exception?, string> formatter)
+            => _messages.Enqueue(formatter(state, exception));
+    }
+
+    /// <summary>
+    /// A directory whose reads can be made to fail the way an unreachable valkey does.
+    /// </summary>
+    private sealed class FailableSessionDirectory : ISessionDirectory
+    {
+        private readonly ISessionDirectory _inner;
+
+        public FailableSessionDirectory(ISessionDirectory inner)
+        {
+            _inner = inner;
+        }
+
+        public bool FailReads { get; set; }
+
+        public Task<long> AllocateConnectionEpochAsync(string sessionId, CancellationToken cancellationToken = default)
+            => _inner.AllocateConnectionEpochAsync(sessionId, cancellationToken);
+
+        public Task<bool> PublishAsync(SessionDirectoryEntry entry, long connectionEpoch, CancellationToken cancellationToken = default)
+            => _inner.PublishAsync(entry, connectionEpoch, cancellationToken);
+
+        public Task RemoveAsync(string sessionId, string ownerPod, CancellationToken cancellationToken = default)
+            => _inner.RemoveAsync(sessionId, ownerPod, cancellationToken);
+
+        public Task<SessionDirectoryEntry?> GetAsync(string sessionId, CancellationToken cancellationToken = default)
+            => FailReads
+                ? throw new RedisTimeoutException("The session directory is unreachable.", CommandStatus.Unknown)
+                : _inner.GetAsync(sessionId, cancellationToken);
+
+        public Task<IReadOnlyList<SessionDirectoryEntry>> GetAllAsync(CancellationToken cancellationToken = default)
+            => FailReads
+                ? throw new RedisTimeoutException("The session directory is unreachable.", CommandStatus.Unknown)
+                : _inner.GetAllAsync(cancellationToken);
+    }
+
+    /// <summary>
+    /// Stands in for the websocket the owning replica holds.
+    /// </summary>
+    private sealed class RecordingSessionController : ISessionController
+    {
+        private readonly TaskCompletionSource<(SessionMessageType MessageType, string Data)> _received = new();
+
+        public bool IsSessionActive { get; set; } = true;
+
+        public bool SupportsMediaControl => true;
+
+        public bool HasMessage => _received.Task.IsCompleted;
+
+        public Task SendMessage<T>(SessionMessageType name, Guid messageId, T data, CancellationToken cancellationToken)
+        {
+            _received.TrySetResult((name, JsonSerializer.Serialize(data)));
+            return Task.CompletedTask;
+        }
+
+        public Task<(SessionMessageType MessageType, string Data)> WaitForMessageAsync(CancellationToken cancellationToken)
+            => _received.Task.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken);
+    }
+}

tests/Jellyfin.Server.Tests/Jellyfin.Server.Tests.csproj

@@ -11,6 +11,7 @@
     <PackageReference Include="Microsoft.AspNetCore.Mvc.Testing" />
     <PackageReference Include="Microsoft.NET.Test.Sdk" />
     <PackageReference Include="Npgsql" />
+    <PackageReference Include="StackExchange.Redis" />
     <PackageReference Include="Testcontainers.PostgreSql" />
     <PackageReference Include="Testcontainers.Redis" />
     <PackageReference Include="xunit.v3" />