fix(session): parse the owner key without assuming the pod id

Ownership is claimed with a Lua check-and-set keyed on the instance holding
the websocket, routing prefers a live controller over a local copy, the
session list deduplicates by owner, removal is ownership-checked, undelivered
routed messages surface, single-session lookups stop scanning the keyspace and
directory writes leave the request path bounded by a timeout.

.woodpecker/ci.yaml

@@ -47,7 +47,7 @@ steps:
   # Its data directory lives on the step's ephemeral storage, not on the workspace volume.
   # Both projects attach to it through JELLYFIN_TEST_POSTGRES and give every test a database of
   # its own, so nothing here depends on a docker daemon.
-  # Valkey runs in the step for the same reason, attached through JELLYFIN_TEST_REDIS.
+  # Valkey runs in the step for the same reason, reached through JELLYFIN_TEST_REDIS.
   - name: postgres-migration-chain
     image: mcr.microsoft.com/dotnet/sdk:10.0
     depends_on:
@@ -64,8 +64,8 @@ steps:
       - PGBIN=$(ls -d /usr/lib/postgresql/*/bin | tail -1)
       - 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
+      - valkey-server --daemonize yes --bind 127.0.0.1 --port 6379 --save ""
+      - 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/QuickConnect/QuickConnectManager.cs

@@ -1,5 +1,7 @@
 using System;
+using System.Collections.Concurrent;
 using System.Globalization;
+using System.Linq;
 using System.Security.Cryptography;
 using System.Threading.Tasks;
 using MediaBrowser.Common.Extensions;
@@ -28,10 +30,12 @@ namespace Emby.Server.Implementations.QuickConnect
         /// </summary>
         private const int Timeout = 10;
 
+        private readonly ConcurrentDictionary<string, QuickConnectResult> _currentRequests = new();
+        private readonly ConcurrentDictionary<string, (DateTime Timestamp, AuthenticationResult AuthenticationResult)> _authorizedSecrets = new();
+
         private readonly IServerConfigurationManager _config;
         private readonly ILogger<QuickConnectManager> _logger;
         private readonly ISessionManager _sessionManager;
-        private readonly IQuickConnectStore _store;
 
         /// <summary>
         /// Initializes a new instance of the <see cref="QuickConnectManager"/> class.
@@ -40,17 +44,14 @@ namespace Emby.Server.Implementations.QuickConnect
         /// <param name="config">Configuration.</param>
         /// <param name="logger">Logger.</param>
         /// <param name="sessionManager">Session Manager.</param>
-        /// <param name="store">Quick connect store.</param>
         public QuickConnectManager(
             IServerConfigurationManager config,
             ILogger<QuickConnectManager> logger,
-            ISessionManager sessionManager,
-            IQuickConnectStore store)
+            ISessionManager sessionManager)
         {
             _config = config;
             _logger = logger;
             _sessionManager = sessionManager;
-            _store = store;
         }
 
         /// <inheritdoc />
@@ -68,7 +69,7 @@ namespace Emby.Server.Implementations.QuickConnect
         }
 
         /// <inheritdoc/>
-        public async Task<QuickConnectResult> TryConnect(AuthorizationInfo authorizationInfo)
+        public QuickConnectResult TryConnect(AuthorizationInfo authorizationInfo)
         {
             ArgumentException.ThrowIfNullOrEmpty(authorizationInfo.DeviceId);
             ArgumentException.ThrowIfNullOrEmpty(authorizationInfo.Device);
@@ -76,6 +77,7 @@ namespace Emby.Server.Implementations.QuickConnect
             ArgumentException.ThrowIfNullOrEmpty(authorizationInfo.Version);
 
             AssertActive();
+            ExpireRequests();
 
             var secret = GenerateSecureRandom();
             var code = GenerateCode();
@@ -88,17 +90,19 @@ namespace Emby.Server.Implementations.QuickConnect
                 authorizationInfo.Client,
                 authorizationInfo.Version);
 
-            await _store.SetRequestAsync(result, ExpiryOf(result)).ConfigureAwait(false);
+            _currentRequests[code] = result;
             return result;
         }
 
         /// <inheritdoc/>
-        public async Task<QuickConnectResult> CheckRequestStatus(string secret)
+        public QuickConnectResult CheckRequestStatus(string secret)
         {
             AssertActive();
+            ExpireRequests();
 
-            var result = await _store.GetRequestBySecretAsync(secret).ConfigureAwait(false);
-            if (result is null)
+            string code = _currentRequests.Where(x => x.Value.Secret == secret).Select(x => x.Value.Code).DefaultIfEmpty(string.Empty).First();
+
+            if (!_currentRequests.TryGetValue(code, out QuickConnectResult? result))
             {
                 throw new ResourceNotFoundException("Unable to find request with provided secret");
             }
@@ -132,9 +136,9 @@ namespace Emby.Server.Implementations.QuickConnect
         public async Task<bool> AuthorizeRequest(Guid userId, string code)
         {
             AssertActive();
+            ExpireRequests();
 
-            var result = await _store.GetRequestByCodeAsync(code).ConfigureAwait(false);
-            if (result is null)
+            if (!_currentRequests.TryGetValue(code, out QuickConnectResult? result))
             {
                 throw new ResourceNotFoundException("Unable to find request");
             }
@@ -147,12 +151,6 @@ namespace Emby.Server.Implementations.QuickConnect
             // Change the time on the request so it expires one minute into the future. It can't expire immediately as otherwise some clients wouldn't ever see that they have been authenticated.
             result.DateAdded = DateTime.UtcNow.Add(TimeSpan.FromMinutes(1));
 
-            // The guard above is a read on shared state, so it cannot settle a race between instances; the claim can.
-            if (!await _store.TryClaimAuthorizationAsync(result.Secret, ExpiryOf(result)).ConfigureAwait(false))
-            {
-                throw new InvalidOperationException("Request is already authorized");
-            }
-
             var authenticationResult = await _sessionManager.AuthenticateDirect(new AuthenticationRequest
             {
                 UserId = userId,
@@ -162,10 +160,9 @@ namespace Emby.Server.Implementations.QuickConnect
                 AppVersion = result.AppVersion
             }).ConfigureAwait(false);
 
+            _authorizedSecrets[result.Secret] = (DateTime.UtcNow, authenticationResult);
             result.Authenticated = true;
-
-            await _store.SetAuthorizationAsync(result.Secret, authenticationResult, DateTime.UtcNow.AddMinutes(Timeout)).ConfigureAwait(false);
-            await _store.SetRequestAsync(result, ExpiryOf(result)).ConfigureAwait(false);
+            _currentRequests[code] = result;
 
             _logger.LogDebug("Authorizing device with code {Code} to login as user {UserId}", code, userId);
 
@@ -173,21 +170,19 @@ namespace Emby.Server.Implementations.QuickConnect
         }
 
         /// <inheritdoc/>
-        public async Task<AuthenticationResult> GetAuthorizedRequest(string secret)
+        public AuthenticationResult GetAuthorizedRequest(string secret)
         {
             AssertActive();
+            ExpireRequests();
 
-            var result = await _store.TryConsumeAuthorizationAsync(secret).ConfigureAwait(false);
-            if (result is null)
+            if (!_authorizedSecrets.TryGetValue(secret, out var result))
             {
                 throw new ResourceNotFoundException("Unable to find request");
             }
 
-            return result;
+            return result.AuthenticationResult;
         }
 
-        private static DateTime ExpiryOf(QuickConnectResult request) => request.DateAdded.AddMinutes(Timeout);
-
         private string GenerateSecureRandom(int length = 32)
         {
             Span<byte> bytes = stackalloc byte[length];
@@ -195,5 +190,42 @@ namespace Emby.Server.Implementations.QuickConnect
 
             return Convert.ToHexString(bytes);
         }
+
+        /// <summary>
+        /// Expire quick connect requests that are over the time limit. If <paramref name="expireAll"/> is true, all requests are unconditionally expired.
+        /// </summary>
+        /// <param name="expireAll">If true, all requests will be expired.</param>
+        private void ExpireRequests(bool expireAll = false)
+        {
+            // All requests before this timestamp have expired
+            var minTime = DateTime.UtcNow.AddMinutes(-Timeout);
+
+            // Expire stale connection requests
+            foreach (var (_, currentRequest) in _currentRequests)
+            {
+                if (expireAll || currentRequest.DateAdded < minTime)
+                {
+                    var code = currentRequest.Code;
+                    _logger.LogDebug("Removing expired request {Code}", code);
+
+                    if (!_currentRequests.TryRemove(code, out _))
+                    {
+                        _logger.LogWarning("Request {Code} already expired", code);
+                    }
+                }
+            }
+
+            foreach (var (secret, (timestamp, _)) in _authorizedSecrets)
+            {
+                if (expireAll || timestamp < minTime)
+                {
+                    _logger.LogDebug("Removing expired secret {Secret}", secret);
+                    if (!_authorizedSecrets.TryRemove(secret, out _))
+                    {
+                        _logger.LogWarning("Secret {Secret} already expired", secret);
+                    }
+                }
+            }
+        }
     }
 }

Emby.Server.Implementations/QuickConnect/RedisQuickConnectStore.cs

@@ -1,175 +0,0 @@
-using System;
-using System.Text.Json;
-using System.Threading;
-using System.Threading.Tasks;
-using Jellyfin.Extensions.Json;
-using MediaBrowser.Controller.Authentication;
-using MediaBrowser.Controller.QuickConnect;
-using MediaBrowser.Model.QuickConnect;
-using Microsoft.Extensions.Logging;
-using StackExchange.Redis;
-
-namespace Emby.Server.Implementations.QuickConnect;
-
-/// <summary>
-/// A Redis-backed <see cref="IQuickConnectStore"/> that lets the initiate, authorize and exchange legs
-/// of a quick connect flow land on different instances. Expiry is the key TTL, an authorization is
-/// claimed with a Lua check-and-set and consumed with <c>GETDEL</c>, so only one instance can ever mint
-/// a given secret's access token and only one can ever hand it out.
-/// </summary>
-/// <remarks>
-/// A pending request survives an unreachable Redis through a process-local fallback, because a second
-/// copy of it is harmless. An authorization has none: a second copy of it is a second access token, and
-/// a write whose response timed out may well have been applied, so a transport failure on that path is
-/// surfaced rather than degraded.
-/// </remarks>
-public sealed class RedisQuickConnectStore : IQuickConnectStore
-{
-    private const string KeyPrefix = "jellyfin:quickconnect:";
-
-    /// <summary>
-    /// Lua script for the atomic claim of the sole right to authorize a request: the request has to
-    /// exist and not already be authorized, and the claim marker is taken with <c>SET NX</c>, so of two
-    /// instances racing on one code exactly one goes on to mint an access token.
-    /// </summary>
-    private const string ClaimAuthorizationScript = @"
-local raw = redis.call('GET', KEYS[1])
-if not raw then return 0 end
-if cjson.decode(raw)['Authenticated'] then return 0 end
-if redis.call('SET', KEYS[2], '1', 'NX', 'PX', ARGV[1]) then return 1 end
-return 0";
-
-    private readonly IDatabase _db;
-    private readonly InMemoryQuickConnectStore _fallback;
-    private readonly ILogger<RedisQuickConnectStore> _logger;
-
-    /// <summary>
-    /// Initializes a new instance of the <see cref="RedisQuickConnectStore"/> class.
-    /// </summary>
-    /// <param name="redis">The Redis connection multiplexer.</param>
-    /// <param name="logger">The logger.</param>
-    public RedisQuickConnectStore(IConnectionMultiplexer redis, ILogger<RedisQuickConnectStore> logger)
-    {
-        ArgumentNullException.ThrowIfNull(redis);
-
-        _db = redis.GetDatabase();
-        _fallback = new InMemoryQuickConnectStore();
-        _logger = logger;
-    }
-
-    /// <inheritdoc />
-    public async Task<QuickConnectResult?> GetRequestBySecretAsync(string secret, CancellationToken cancellationToken = default)
-    {
-        RedisValue raw;
-        try
-        {
-            raw = await _db.StringGetAsync(RequestKey(secret)).ConfigureAwait(false);
-        }
-        catch (Exception ex) when (IsTransportFailure(ex))
-        {
-            LogDegraded(ex);
-            return await _fallback.GetRequestBySecretAsync(secret, cancellationToken).ConfigureAwait(false);
-        }
-
-        // A miss is an answer rather than a transport failure, so the fallback is not consulted for it.
-        return raw.HasValue ? JsonSerializer.Deserialize<QuickConnectResult>(raw.ToString(), JsonDefaults.Options) : null;
-    }
-
-    /// <inheritdoc />
-    public async Task<QuickConnectResult?> GetRequestByCodeAsync(string code, CancellationToken cancellationToken = default)
-    {
-        RedisValue secret;
-        try
-        {
-            secret = await _db.StringGetAsync(CodeKey(code)).ConfigureAwait(false);
-        }
-        catch (Exception ex) when (IsTransportFailure(ex))
-        {
-            LogDegraded(ex);
-            return await _fallback.GetRequestByCodeAsync(code, cancellationToken).ConfigureAwait(false);
-        }
-
-        return secret.HasValue
-            ? await GetRequestBySecretAsync(secret.ToString(), cancellationToken).ConfigureAwait(false)
-            : null;
-    }
-
-    /// <inheritdoc />
-    public async Task SetRequestAsync(QuickConnectResult request, DateTime expiresUtc, CancellationToken cancellationToken = default)
-    {
-        ArgumentNullException.ThrowIfNull(request);
-
-        var ttl = expiresUtc - DateTime.UtcNow;
-        if (ttl <= TimeSpan.Zero)
-        {
-            return;
-        }
-
-        try
-        {
-            var json = JsonSerializer.Serialize(request, JsonDefaults.Options);
-            await _db.StringSetAsync(RequestKey(request.Secret), json, ttl).ConfigureAwait(false);
-            await _db.StringSetAsync(CodeKey(request.Code), request.Secret, ttl).ConfigureAwait(false);
-        }
-        catch (Exception ex) when (IsTransportFailure(ex))
-        {
-            LogDegraded(ex);
-            await _fallback.SetRequestAsync(request, expiresUtc, cancellationToken).ConfigureAwait(false);
-        }
-    }
-
-    /// <inheritdoc />
-    public async Task<bool> TryClaimAuthorizationAsync(string secret, DateTime expiresUtc, CancellationToken cancellationToken = default)
-    {
-        var ttl = expiresUtc - DateTime.UtcNow;
-        if (ttl <= TimeSpan.Zero)
-        {
-            return false;
-        }
-
-        var claimed = (long?)await _db.ScriptEvaluateAsync(
-            ClaimAuthorizationScript,
-            keys: new RedisKey[] { RequestKey(secret), ClaimKey(secret) },
-            values: new RedisValue[] { (long)ttl.TotalMilliseconds }).ConfigureAwait(false);
-
-        return claimed == 1;
-    }
-
-    /// <inheritdoc />
-    public async Task SetAuthorizationAsync(string secret, AuthenticationResult authenticationResult, DateTime expiresUtc, CancellationToken cancellationToken = default)
-    {
-        var ttl = expiresUtc - DateTime.UtcNow;
-        if (ttl <= TimeSpan.Zero)
-        {
-            return;
-        }
-
-        var json = JsonSerializer.Serialize(authenticationResult, JsonDefaults.Options);
-        await _db.StringSetAsync(AuthorizationKey(secret), json, ttl).ConfigureAwait(false);
-    }
-
-    /// <inheritdoc />
-    public async Task<AuthenticationResult?> TryConsumeAuthorizationAsync(string secret, CancellationToken cancellationToken = default)
-    {
-        var raw = await _db.StringGetDeleteAsync(AuthorizationKey(secret)).ConfigureAwait(false);
-
-        return raw.HasValue
-            ? JsonSerializer.Deserialize<AuthenticationResult>(raw.ToString(), JsonDefaults.Options)
-            : null;
-    }
-
-    // Deliberately excludes a malformed stored value, which is a fault of its own rather than a reason
-    // to answer from this instance.
-    private static bool IsTransportFailure(Exception exception) => exception is RedisException or TimeoutException;
-
-    private static string RequestKey(string secret) => KeyPrefix + "request:" + secret;
-
-    private static string CodeKey(string code) => KeyPrefix + "code:" + code;
-
-    private static string ClaimKey(string secret) => KeyPrefix + "claim:" + secret;
-
-    private static string AuthorizationKey(string secret) => KeyPrefix + "auth:" + secret;
-
-    private void LogDegraded(Exception exception)
-        => _logger.LogWarning(exception, "Quick connect request state could not be shared through Redis; falling back to this instance only.");
-}

Emby.Server.Implementations/Session/RedisPodMessageBus.cs

@@ -0,0 +1,95 @@
+using System;
+using System.Text.Json;
+using System.Threading;
+using System.Threading.Tasks;
+using Jellyfin.Extensions.Json;
+using MediaBrowser.Controller.Session;
+using Microsoft.Extensions.Logging;
+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.
+/// </summary>
+public sealed class RedisPodMessageBus : IPodMessageBus
+{
+    private const string ChannelPrefix = "jellyfin:pod:";
+
+    private static readonly TimeSpan _publishTimeout = TimeSpan.FromSeconds(5);
+
+    private static readonly JsonSerializerOptions _jsonOptions = JsonDefaults.Options;
+
+    private readonly ISubscriber _subscriber;
+    private readonly ILogger<RedisPodMessageBus> _logger;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="RedisPodMessageBus"/> class.
+    /// </summary>
+    /// <param name="redis">The Redis connection multiplexer.</param>
+    /// <param name="logger">The logger.</param>
+    public RedisPodMessageBus(IConnectionMultiplexer redis, ILogger<RedisPodMessageBus> logger)
+    {
+        ArgumentNullException.ThrowIfNull(redis);
+
+        _subscriber = redis.GetSubscriber();
+        _logger = logger;
+        PodId = PodIdentity.Current;
+    }
+
+    /// <inheritdoc />
+    public string PodId { get; }
+
+    /// <inheritdoc />
+    public async Task<long> PublishAsync(string targetPod, PodMessage message, CancellationToken cancellationToken = default)
+    {
+        ArgumentException.ThrowIfNullOrEmpty(targetPod);
+        ArgumentNullException.ThrowIfNull(message);
+
+        message.OriginPod = PodId;
+
+        try
+        {
+            return await _subscriber.PublishAsync(
+                RedisChannel.Literal(ChannelPrefix + targetPod),
+                JsonSerializer.Serialize(message, _jsonOptions)).WaitAsync(_publishTimeout, cancellationToken).ConfigureAwait(false);
+        }
+        catch (Exception ex)
+        {
+            _logger.LogWarning(ex, "Failed to send a {Kind} message to {TargetPod}.", message.Kind, targetPod);
+            return 0;
+        }
+    }
+
+    /// <inheritdoc />
+    public void Subscribe(Func<PodMessage, Task> handler)
+    {
+        ArgumentNullException.ThrowIfNull(handler);
+
+        try
+        {
+            _subscriber.Subscribe(RedisChannel.Literal(ChannelPrefix + PodId), (_, value) => Dispatch(handler, value));
+        }
+        catch (Exception ex)
+        {
+            _logger.LogWarning(ex, "Failed to subscribe to {PodId}; messages routed here are dropped.", PodId);
+        }
+    }
+
+    private async void Dispatch(Func<PodMessage, Task> handler, RedisValue value)
+    {
+        try
+        {
+            var message = JsonSerializer.Deserialize<PodMessage>(value.ToString(), _jsonOptions);
+            if (message is not null)
+            {
+                await handler(message).ConfigureAwait(false);
+            }
+        }
+        catch (Exception ex)
+        {
+            _logger.LogWarning(ex, "Failed to handle a message routed to this instance.");
+        }
+    }
+}

Emby.Server.Implementations/Session/RedisSessionDirectory.cs

@@ -0,0 +1,215 @@
+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, so an instance that only served a request
+/// for the session cannot take it from the instance the device is actually connected to. Each entry is a
+/// key with an expiry, so the sessions of an instance that stops refreshing them disappear on their own.
+/// </summary>
+public sealed class RedisSessionDirectory : ISessionDirectory
+{
+    private const string KeyPrefix = "jellyfin:session:";
+    private const string OwnerKeyPrefix = "jellyfin:sessionowner:";
+
+    /// <summary>
+    /// Lua script for an atomic ownership claim. The owner key holds <c>connectedTicks|pod</c>, where
+    /// the ticks are zero for an instance that holds no connection. A claim by another instance is
+    /// refused unless its connection is newer than the recorded one, so the instance holding the live
+    /// connection keeps ownership however many requests the others serve.
+    /// </summary>
+    private const string ClaimScript = @"
+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)
+    if owner ~= ARGV[1] and connected > 0 and tonumber(ARGV[2]) <= 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";
+
+    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;
+
+    private TimeSpan OperationTimeout => TimeSpan.FromSeconds(Math.Max(1, _options.OperationTimeoutSeconds));
+
+    /// <inheritdoc />
+    public async Task<bool> PublishAsync(SessionDirectoryEntry entry, long connectedUtcTicks, 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 },
+                values: new RedisValue[]
+                {
+                    entry.OwnerPod,
+                    connectedUtcTicks.ToString(CultureInfo.InvariantCulture),
+                    JsonSerializer.Serialize(entry, _jsonOptions),
+                    EntryTtlMs
+                }).WaitAsync(OperationTimeout, cancellationToken).ConfigureAwait(false);
+
+            return claimed == 1;
+        }
+        catch (Exception ex)
+        {
+            _logger.LogWarning(ex, "Failed to publish session {SessionId}; it stays invisible to the other instances.", sessionId);
+            return false;
+        }
+    }
+
+    /// <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)
+    {
+        try
+        {
+            var raw = await _db.StringGetAsync(KeyPrefix + sessionId).WaitAsync(OperationTimeout, cancellationToken).ConfigureAwait(false);
+
+            return raw.HasValue ? Deserialize(raw) : null;
+        }
+        catch (Exception ex)
+        {
+            _logger.LogWarning(ex, "Failed to read session {SessionId} from the directory.", sessionId);
+            return null;
+        }
+    }
+
+    /// <inheritdoc />
+    public async Task<IReadOnlyList<SessionDirectoryEntry>> GetAllAsync(CancellationToken cancellationToken = default)
+    {
+        var entries = new List<SessionDirectoryEntry>();
+
+        try
+        {
+            foreach (var server in _redis.GetServers())
+            {
+                if (!server.IsConnected)
+                {
+                    continue;
+                }
+
+                var keys = new List<RedisKey>();
+                await foreach (var key in server.KeysAsync(database: _db.Database, pattern: KeyPrefix + "*", pageSize: 1000).WithCancellation(cancellationToken).ConfigureAwait(false))
+                {
+                    keys.Add(key);
+                }
+
+                var values = await Task.WhenAll(keys.Select(key => _db.StringGetAsync(key)))
+                    .WaitAsync(OperationTimeout, cancellationToken).ConfigureAwait(false);
+
+                foreach (var raw in values)
+                {
+                    if (!raw.HasValue)
+                    {
+                        continue;
+                    }
+
+                    var entry = Deserialize(raw);
+                    if (entry?.Session is not null)
+                    {
+                        entries.Add(entry);
+                    }
+                }
+            }
+        }
+        catch (Exception ex)
+        {
+            // Degrade to the sessions this instance holds rather than failing the request outright.
+            _logger.LogWarning(ex, "Failed to read the session directory; only local sessions are reported.");
+            return Array.Empty<SessionDirectoryEntry>();
+        }
+
+        return entries;
+    }
+
+    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,76 @@
+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.
+/// </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>
+    public RemoteSessionController(IPodMessageBus bus, ILogger logger, string ownerPod, string sessionId, bool supportsMediaControl)
+    {
+        _bus = bus;
+        _logger = logger;
+        _ownerPod = ownerPod;
+        _sessionId = sessionId;
+        SupportsMediaControl = supportsMediaControl;
+    }
+
+    /// <inheritdoc />
+    public bool IsSessionActive => true;
+
+    /// <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.PublishAsync(
+            _ownerPod,
+            new PodMessage
+            {
+                Kind = RoutedSessionMessage.Kind,
+                Payload = JsonSerializer.Serialize(routed, JsonDefaults.Options)
+            },
+            cancellationToken).ConfigureAwait(false);
+
+        if (delivered == 0)
+        {
+            _logger.LogWarning("Instance {OwnerPod} holds session {SessionId} but is not listening; the {MessageType} message was not delivered.", _ownerPod, _sessionId, name);
+
+            throw new ResourceNotFoundException(
+                string.Format(CultureInfo.InvariantCulture, "The instance holding session {0} is unreachable.", _sessionId));
+        }
+    }
+}

Emby.Server.Implementations/Session/SessionManager.cs

@@ -5,6 +5,7 @@ using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Globalization;
 using System.Linq;
+using System.Text.Json;
 using System.Threading;
 using System.Threading.Tasks;
 using Jellyfin.Data;
@@ -15,6 +16,7 @@ using Jellyfin.Database.Implementations.Entities;
 using Jellyfin.Database.Implementations.Entities.Security;
 using Jellyfin.Database.Implementations.Enums;
 using Jellyfin.Extensions;
+using Jellyfin.Extensions.Json;
 using MediaBrowser.Common.Events;
 using MediaBrowser.Common.Extensions;
 using MediaBrowser.Controller;
@@ -39,6 +41,7 @@ using MediaBrowser.Model.SyncPlay;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
 using Episode = MediaBrowser.Controller.Entities.TV.Episode;
 
 namespace Emby.Server.Implementations.Session
@@ -60,15 +63,23 @@ namespace Emby.Server.Implementations.Session
         private readonly IMediaSourceManager _mediaSourceManager;
         private readonly IServerApplicationHost _appHost;
         private readonly IDeviceManager _deviceManager;
+        private readonly ISessionDirectory _sessionDirectory;
+        private readonly IPodMessageBus _podMessageBus;
+        private readonly SessionDirectoryOptions _sessionDirectoryOptions;
+        private readonly bool _directoryEnabled;
         private readonly CancellationTokenRegistration _shutdownCallback;
         private readonly ConcurrentDictionary<string, SessionInfo> _activeConnections
             = new(StringComparer.OrdinalIgnoreCase);
 
+        private readonly ConcurrentDictionary<string, long> _connectionEpochs = new(StringComparer.Ordinal);
+        private readonly ConcurrentDictionary<string, long> _lastDirectoryPublish = new(StringComparer.Ordinal);
+
         private readonly ConcurrentDictionary<string, ConcurrentDictionary<string, string>> _activeLiveStreamSessions
             = new(StringComparer.OrdinalIgnoreCase);
 
         private Timer _idleTimer;
         private Timer _inactiveTimer;
+        private Timer _directoryTimer;
 
         private DtoOptions _itemInfoDtoOptions;
         private bool _disposed;
@@ -89,6 +100,9 @@ namespace Emby.Server.Implementations.Session
         /// <param name="deviceManager">Instance of <see cref="IDeviceManager"/> interface.</param>
         /// <param name="mediaSourceManager">Instance of <see cref="IMediaSourceManager"/> interface.</param>
         /// <param name="hostApplicationLifetime">Instance of <see cref="IHostApplicationLifetime"/> interface.</param>
+        /// <param name="sessionDirectory">Instance of <see cref="ISessionDirectory"/> interface.</param>
+        /// <param name="podMessageBus">Instance of <see cref="IPodMessageBus"/> interface.</param>
+        /// <param name="sessionDirectoryOptions">The session directory options.</param>
         public SessionManager(
             ILogger<SessionManager> logger,
             IEventManager eventManager,
@@ -102,7 +116,10 @@ namespace Emby.Server.Implementations.Session
             IServerApplicationHost appHost,
             IDeviceManager deviceManager,
             IMediaSourceManager mediaSourceManager,
-            IHostApplicationLifetime hostApplicationLifetime)
+            IHostApplicationLifetime hostApplicationLifetime,
+            ISessionDirectory sessionDirectory,
+            IPodMessageBus podMessageBus,
+            IOptions<SessionDirectoryOptions> sessionDirectoryOptions)
         {
             _logger = logger;
             _eventManager = eventManager;
@@ -116,9 +133,21 @@ namespace Emby.Server.Implementations.Session
             _appHost = appHost;
             _deviceManager = deviceManager;
             _mediaSourceManager = mediaSourceManager;
+            _sessionDirectory = sessionDirectory;
+            _podMessageBus = podMessageBus;
+            _sessionDirectoryOptions = sessionDirectoryOptions.Value;
             _shutdownCallback = hostApplicationLifetime.ApplicationStopping.Register(OnApplicationStopping);
 
             _deviceManager.DeviceOptionsUpdated += OnDeviceManagerDeviceOptionsUpdated;
+
+            _directoryEnabled = _sessionDirectory is not NullSessionDirectory;
+
+            if (_directoryEnabled)
+            {
+                _podMessageBus.Subscribe(OnPodMessage);
+                var interval = TimeSpan.FromSeconds(Math.Max(1, _sessionDirectoryOptions.RefreshIntervalSeconds));
+                _directoryTimer = new Timer(RefreshSessionDirectory, null, interval, interval);
+            }
         }
 
         /// <summary>
@@ -218,6 +247,8 @@ namespace Emby.Server.Implementations.Session
 
             _eventManager.Publish(new SessionEndedEventArgs(info));
 
+            await RemoveFromDirectoryAsync(info).ConfigureAwait(false);
+
             await info.DisposeAsync().ConfigureAwait(false);
         }
 
@@ -288,11 +319,13 @@ namespace Emby.Server.Implementations.Session
                     });
             }
 
+            QueueDirectoryPublish(session);
+
             return session;
         }
 
         /// <inheritdoc />
-        public void OnSessionControllerConnected(SessionInfo session)
+        public async Task OnSessionControllerConnected(SessionInfo session)
         {
             EventHelper.QueueEventIfNotNull(
                 SessionControllerConnected,
@@ -302,6 +335,219 @@ namespace Emby.Server.Implementations.Session
                     SessionInfo = session
                 },
                 _logger);
+
+            // Ownership of the session belongs to whichever instance holds its connection, so this one
+            // claims it before the connection is used.
+            _lastDirectoryPublish[session.Id] = Environment.TickCount64;
+            await PublishToDirectoryAsync(session).ConfigureAwait(false);
+        }
+
+        // Keeps the directory write off the request path: the caller does not wait for Redis, and a
+        // session reporting playback every few seconds does not write on every report.
+        private void QueueDirectoryPublish(SessionInfo session)
+        {
+            if (!_directoryEnabled || string.IsNullOrEmpty(session.Id))
+            {
+                return;
+            }
+
+            var now = Environment.TickCount64;
+            var throttleMs = Math.Max(1000L, _sessionDirectoryOptions.RefreshIntervalSeconds * 500L);
+            var scheduled = _lastDirectoryPublish.AddOrUpdate(
+                session.Id,
+                now,
+                (_, last) => now - last >= throttleMs ? now : last);
+
+            if (scheduled == now)
+            {
+                _ = PublishToDirectoryAsync(session);
+            }
+        }
+
+        private async Task PublishToDirectoryAsync(SessionInfo session)
+        {
+            if (!_directoryEnabled || string.IsNullOrEmpty(session.Id))
+            {
+                return;
+            }
+
+            try
+            {
+                var connectedUtcTicks = GetConnectionEpoch(session);
+
+                await _sessionDirectory.PublishAsync(
+                    new SessionDirectoryEntry
+                    {
+                        OwnerPod = _podMessageBus.PodId,
+                        HoldsConnection = connectedUtcTicks > 0,
+                        Session = ToSessionInfoDto(session)
+                    },
+                    connectedUtcTicks).ConfigureAwait(false);
+            }
+            catch (Exception ex)
+            {
+                _logger.LogDebug(ex, "Error publishing session {Session} to the directory.", session.Id);
+            }
+        }
+
+        // Ownership follows the connection, not the last request served: an instance without a live
+        // controller claims with epoch zero, which never displaces an instance that has one.
+        private long GetConnectionEpoch(SessionInfo session)
+        {
+            if (!session.SessionControllers.Any(i => i.IsSessionActive))
+            {
+                _connectionEpochs.TryRemove(session.Id, out _);
+                return 0;
+            }
+
+            return _connectionEpochs.GetOrAdd(session.Id, _ => DateTime.UtcNow.Ticks);
+        }
+
+        private async ValueTask RemoveFromDirectoryAsync(SessionInfo session)
+        {
+            _connectionEpochs.TryRemove(session.Id, out _);
+            _lastDirectoryPublish.TryRemove(session.Id, out _);
+
+            if (!_directoryEnabled || string.IsNullOrEmpty(session.Id))
+            {
+                return;
+            }
+
+            await _sessionDirectory.RemoveAsync(session.Id, _podMessageBus.PodId).ConfigureAwait(false);
+        }
+
+        private async void RefreshSessionDirectory(object state)
+        {
+            try
+            {
+                foreach (var session in _activeConnections.Values)
+                {
+                    await PublishToDirectoryAsync(session).ConfigureAwait(false);
+                }
+            }
+            catch (Exception ex)
+            {
+                _logger.LogDebug(ex, "Error refreshing the session directory.");
+            }
+        }
+
+        private async Task<IReadOnlyList<SessionDirectoryEntry>> GetRemoteEntriesAsync(CancellationToken cancellationToken)
+        {
+            if (!_directoryEnabled)
+            {
+                return Array.Empty<SessionDirectoryEntry>();
+            }
+
+            var entries = await _sessionDirectory.GetAllAsync(cancellationToken).ConfigureAwait(false);
+
+            return entries
+                .Where(entry => entry.Session is not null
+                    && !string.Equals(entry.OwnerPod, _podMessageBus.PodId, StringComparison.Ordinal))
+                .ToList();
+        }
+
+        private async Task<SessionInfo> GetRemoteSession(string sessionId)
+        {
+            if (!_directoryEnabled)
+            {
+                return null;
+            }
+
+            var entry = await _sessionDirectory.GetAsync(sessionId).ConfigureAwait(false);
+
+            if (entry?.Session is null
+                || string.Equals(entry.OwnerPod, _podMessageBus.PodId, StringComparison.Ordinal))
+            {
+                return null;
+            }
+
+            var dto = entry.Session;
+            var session = new SessionInfo(this, _logger)
+            {
+                Id = dto.Id,
+                UserId = dto.UserId,
+                UserName = dto.UserName,
+                Client = dto.Client,
+                DeviceId = dto.DeviceId,
+                DeviceName = dto.DeviceName,
+                DeviceType = dto.DeviceType,
+                ApplicationVersion = dto.ApplicationVersion,
+                RemoteEndPoint = dto.RemoteEndPoint,
+                LastActivityDate = dto.LastActivityDate,
+                ServerId = dto.ServerId,
+                AdditionalUsers = dto.AdditionalUsers ?? [],
+                Capabilities = dto.Capabilities?.ToClientCapabilities()
+            };
+
+            if (entry.HoldsConnection)
+            {
+                session.AddController(new RemoteSessionController(_podMessageBus, _logger, entry.OwnerPod, dto.Id, dto.SupportsMediaControl));
+            }
+
+            return session;
+        }
+
+        private Task OnPodMessage(PodMessage message)
+        {
+            switch (message.Kind)
+            {
+                case RoutedSessionMessage.Kind:
+                    return OnRoutedSessionMessage(message);
+                case RoutedAdditionalUserChange.Kind:
+                    OnRoutedAdditionalUserChange(message);
+                    return Task.CompletedTask;
+                default:
+                    return Task.CompletedTask;
+            }
+        }
+
+        private async Task OnRoutedSessionMessage(PodMessage message)
+        {
+            var routed = JsonSerializer.Deserialize<RoutedSessionMessage>(message.Payload, JsonDefaults.Options);
+            if (routed is null)
+            {
+                return;
+            }
+
+            var session = Sessions.FirstOrDefault(i => string.Equals(i.Id, routed.SessionId, StringComparison.Ordinal));
+            var controllers = session?.SessionControllers.Where(i => i.IsSessionActive).ToList();
+
+            if (controllers is null || controllers.Count == 0)
+            {
+                _logger.LogWarning(
+                    "A {MessageType} message for session {Session} was routed to this instance, which no longer holds its connection.",
+                    routed.MessageType,
+                    routed.SessionId);
+                return;
+            }
+
+            using var data = JsonDocument.Parse(routed.Data);
+            foreach (var controller in controllers)
+            {
+                await controller.SendMessage(routed.MessageType, routed.MessageId, data.RootElement, CancellationToken.None).ConfigureAwait(false);
+            }
+        }
+
+        private void OnRoutedAdditionalUserChange(PodMessage message)
+        {
+            var routed = JsonSerializer.Deserialize<RoutedAdditionalUserChange>(message.Payload, JsonDefaults.Options);
+            var session = routed is null
+                ? null
+                : Sessions.FirstOrDefault(i => string.Equals(i.Id, routed.SessionId, StringComparison.Ordinal));
+
+            if (session is null)
+            {
+                return;
+            }
+
+            if (routed.Add)
+            {
+                AttachAdditionalUser(session, routed.UserId, _userManager.GetUserById(routed.UserId)?.Username);
+            }
+            else
+            {
+                DetachAdditionalUser(session, routed.UserId);
+            }
         }
 
         /// <inheritdoc />
@@ -1219,10 +1465,18 @@ namespace Emby.Server.Implementations.Session
             return session;
         }
 
-        private SessionInfo GetSessionToRemoteControl(string sessionId)
+        // A local SessionInfo without a live controller is a copy left behind by a request this instance
+        // happened to serve, not the connection: prefer the owner named by the directory over it.
+        private async Task<SessionInfo> GetSessionToRemoteControl(string sessionId)
         {
-            // Accept either device id or session id
-            var session = Sessions.FirstOrDefault(i => string.Equals(i.Id, sessionId, StringComparison.Ordinal));
+            var local = Sessions.FirstOrDefault(i => string.Equals(i.Id, sessionId, StringComparison.Ordinal));
+
+            if (local is not null && local.SessionControllers.Any(i => i.IsSessionActive))
+            {
+                return local;
+            }
+
+            var session = await GetRemoteSession(sessionId).ConfigureAwait(false) ?? local;
 
             if (session is null)
             {
@@ -1291,19 +1545,19 @@ namespace Emby.Server.Implementations.Session
         }
 
         /// <inheritdoc />
-        public Task SendGeneralCommand(string controllingSessionId, string sessionId, GeneralCommand command, CancellationToken cancellationToken)
+        public async Task SendGeneralCommand(string controllingSessionId, string sessionId, GeneralCommand command, CancellationToken cancellationToken)
         {
             CheckDisposed();
 
-            var session = GetSessionToRemoteControl(sessionId);
+            var session = await GetSessionToRemoteControl(sessionId).ConfigureAwait(false);
 
             if (!string.IsNullOrEmpty(controllingSessionId))
             {
-                var controllingSession = GetSession(controllingSessionId);
+                var controllingSession = await GetSessionToRemoteControl(controllingSessionId).ConfigureAwait(false);
                 AssertCanControl(session, controllingSession);
             }
 
-            return SendMessageToSession(session, SessionMessageType.GeneralCommand, command, cancellationToken);
+            await SendMessageToSession(session, SessionMessageType.GeneralCommand, command, cancellationToken).ConfigureAwait(false);
         }
 
         private static async Task SendMessageToSession<T>(SessionInfo session, SessionMessageType name, T data, CancellationToken cancellationToken)
@@ -1340,7 +1594,7 @@ namespace Emby.Server.Implementations.Session
         {
             CheckDisposed();
 
-            var session = GetSessionToRemoteControl(sessionId);
+            var session = await GetSessionToRemoteControl(sessionId).ConfigureAwait(false);
 
             var user = session.UserId.IsEmpty() ? null : _userManager.GetUserById(session.UserId);
 
@@ -1410,7 +1664,7 @@ namespace Emby.Server.Implementations.Session
 
             if (!string.IsNullOrEmpty(controllingSessionId))
             {
-                var controllingSession = GetSession(controllingSessionId);
+                var controllingSession = await GetSessionToRemoteControl(controllingSessionId).ConfigureAwait(false);
                 AssertCanControl(session, controllingSession);
                 if (!controllingSession.UserId.IsEmpty())
                 {
@@ -1425,7 +1679,16 @@ namespace Emby.Server.Implementations.Session
         public async Task SendSyncPlayCommand(string sessionId, SendCommand command, CancellationToken cancellationToken)
         {
             CheckDisposed();
-            var session = GetSession(sessionId);
+
+            // SyncPlay group membership is instance-local, so a session listed by another instance is not
+            // reachable from here. It is skipped rather than reported as missing.
+            var session = GetSession(sessionId, false);
+            if (session is null)
+            {
+                _logger.LogDebug("SyncPlay command for session {Session} dropped; it is not held by this instance.", sessionId);
+                return;
+            }
+
             await SendMessageToSession(session, SessionMessageType.SyncPlayCommand, command, cancellationToken).ConfigureAwait(false);
         }
 
@@ -1433,7 +1696,14 @@ namespace Emby.Server.Implementations.Session
         public async Task SendSyncPlayGroupUpdate<T>(string sessionId, GroupUpdate<T> command, CancellationToken cancellationToken)
         {
             CheckDisposed();
-            var session = GetSession(sessionId);
+
+            var session = GetSession(sessionId, false);
+            if (session is null)
+            {
+                _logger.LogDebug("SyncPlay group update for session {Session} dropped; it is not held by this instance.", sessionId);
+                return;
+            }
+
             await SendMessageToSession(session, SessionMessageType.SyncPlayGroupUpdate, command, cancellationToken).ConfigureAwait(false);
         }
 
@@ -1521,15 +1791,15 @@ namespace Emby.Server.Implementations.Session
         }
 
         /// <inheritdoc />
-        public Task SendPlaystateCommand(string controllingSessionId, string sessionId, PlaystateRequest command, CancellationToken cancellationToken)
+        public async Task SendPlaystateCommand(string controllingSessionId, string sessionId, PlaystateRequest command, CancellationToken cancellationToken)
         {
             CheckDisposed();
 
-            var session = GetSessionToRemoteControl(sessionId);
+            var session = await GetSessionToRemoteControl(sessionId).ConfigureAwait(false);
 
             if (!string.IsNullOrEmpty(controllingSessionId))
             {
-                var controllingSession = GetSession(controllingSessionId);
+                var controllingSession = await GetSessionToRemoteControl(controllingSessionId).ConfigureAwait(false);
                 AssertCanControl(session, controllingSession);
                 if (!controllingSession.UserId.IsEmpty())
                 {
@@ -1537,7 +1807,7 @@ namespace Emby.Server.Implementations.Session
                 }
             }
 
-            return SendMessageToSession(session, SessionMessageType.Playstate, command, cancellationToken);
+            await SendMessageToSession(session, SessionMessageType.Playstate, command, cancellationToken).ConfigureAwait(false);
         }
 
         private void AssertCanControl(SessionInfo session, SessionInfo controllingSession)
@@ -1606,17 +1876,18 @@ namespace Emby.Server.Implementations.Session
         /// <param name="controllingSessionId">The controlling session identifier.</param>
         /// <param name="sessionId">The session identifier.</param>
         /// <param name="userId">The user identifier.</param>
+        /// <returns>A task representing the operation.</returns>
         /// <exception cref="SecurityException">The controlling user is not allowed to attach the user to the session.</exception>
         /// <exception cref="ArgumentException">The requested user is already the primary user of the session.</exception>
-        public void AddAdditionalUser(string controllingSessionId, string sessionId, Guid userId)
+        public async Task AddAdditionalUser(string controllingSessionId, string sessionId, Guid userId)
         {
             CheckDisposed();
 
-            var session = GetSession(sessionId);
+            var session = await GetSessionToRemoteControl(sessionId).ConfigureAwait(false);
 
             if (!string.IsNullOrEmpty(controllingSessionId))
             {
-                var controllingSession = GetSession(controllingSessionId);
+                var controllingSession = await GetSessionToRemoteControl(controllingSessionId).ConfigureAwait(false);
                 AssertCanControl(session, controllingSession);
                 AssertCanAttachUser(controllingSession, userId);
             }
@@ -1626,18 +1897,16 @@ namespace Emby.Server.Implementations.Session
                 throw new ArgumentException("The requested user is already the primary user of the session.");
             }
 
-            if (session.AdditionalUsers.All(i => !i.UserId.Equals(userId)))
-            {
-                var user = _userManager.GetUserById(userId)
-                    ?? throw new ArgumentException("The requested user does not exist.");
-                var newUser = new SessionUserInfo
-                {
-                    UserId = userId,
-                    UserName = user.Username
-                };
+            var user = _userManager.GetUserById(userId)
+                ?? throw new ArgumentException("The requested user does not exist.");
 
-                session.AdditionalUsers = [.. session.AdditionalUsers, newUser];
+            var local = GetSession(sessionId, false);
+            if (local is not null)
+            {
+                AttachAdditionalUser(local, userId, user.Username);
             }
+
+            await RouteAdditionalUserChange(sessionId, userId, true).ConfigureAwait(false);
         }
 
         /// <summary>
@@ -1646,17 +1915,18 @@ namespace Emby.Server.Implementations.Session
         /// <param name="controllingSessionId">The controlling session identifier.</param>
         /// <param name="sessionId">The session identifier.</param>
         /// <param name="userId">The user identifier.</param>
+        /// <returns>A task representing the operation.</returns>
         /// <exception cref="SecurityException">The controlling user is not allowed to control the session.</exception>
         /// <exception cref="ArgumentException">The requested user is already the primary user of the session.</exception>
-        public void RemoveAdditionalUser(string controllingSessionId, string sessionId, Guid userId)
+        public async Task RemoveAdditionalUser(string controllingSessionId, string sessionId, Guid userId)
         {
             CheckDisposed();
 
-            var session = GetSession(sessionId);
+            var session = await GetSessionToRemoteControl(sessionId).ConfigureAwait(false);
 
             if (!string.IsNullOrEmpty(controllingSessionId))
             {
-                AssertCanControl(session, GetSession(controllingSessionId));
+                AssertCanControl(session, await GetSessionToRemoteControl(controllingSessionId).ConfigureAwait(false));
             }
 
             if (session.UserId.Equals(userId))
@@ -1664,17 +1934,73 @@ namespace Emby.Server.Implementations.Session
                 throw new ArgumentException("The requested user is already the primary user of the session.");
             }
 
-            var user = session.AdditionalUsers.FirstOrDefault(i => i.UserId.Equals(userId));
+            var local = GetSession(sessionId, false);
+            if (local is not null)
+            {
+                DetachAdditionalUser(local, userId);
+            }
 
-            if (user is not null)
+            await RouteAdditionalUserChange(sessionId, userId, false).ConfigureAwait(false);
+        }
+
+        private static void AttachAdditionalUser(SessionInfo session, Guid userId, string userName)
+        {
+            if (session.AdditionalUsers.All(i => !i.UserId.Equals(userId)))
+            {
+                session.AdditionalUsers = [.. session.AdditionalUsers, new SessionUserInfo { UserId = userId, UserName = userName }];
+            }
+        }
+
+        private static void DetachAdditionalUser(SessionInfo session, Guid userId)
+        {
+            var existing = session.AdditionalUsers.FirstOrDefault(i => i.UserId.Equals(userId));
+
+            if (existing is not null)
             {
                 var list = session.AdditionalUsers.ToList();
-                list.Remove(user);
+                list.Remove(existing);
 
                 session.AdditionalUsers = list.ToArray();
             }
         }
 
+        // The owner is the instance whose copy of the session is the one everyone else is shown, so the
+        // change has to be applied there as well as on whichever instance served the request.
+        private async Task RouteAdditionalUserChange(string sessionId, Guid userId, bool add)
+        {
+            if (!_directoryEnabled)
+            {
+                return;
+            }
+
+            var entry = await _sessionDirectory.GetAsync(sessionId).ConfigureAwait(false);
+
+            if (entry is null || string.Equals(entry.OwnerPod, _podMessageBus.PodId, StringComparison.Ordinal))
+            {
+                return;
+            }
+
+            var payload = new RoutedAdditionalUserChange
+            {
+                SessionId = sessionId,
+                UserId = userId,
+                Add = add
+            };
+
+            var delivered = await _podMessageBus.PublishAsync(
+                entry.OwnerPod,
+                new PodMessage
+                {
+                    Kind = RoutedAdditionalUserChange.Kind,
+                    Payload = JsonSerializer.Serialize(payload, JsonDefaults.Options)
+                }).ConfigureAwait(false);
+
+            if (delivered == 0)
+            {
+                _logger.LogWarning("Instance {OwnerPod} holds session {Session} but is not listening; the additional user change was not applied there.", entry.OwnerPod, sessionId);
+            }
+        }
+
         /// <summary>
         /// Authenticates the new session.
         /// </summary>
@@ -2060,14 +2386,25 @@ namespace Emby.Server.Implementations.Session
         }
 
         /// <inheritdoc/>
-        public IReadOnlyList<SessionInfoDto> GetSessions(
+        public async Task<IReadOnlyList<SessionInfoDto>> GetSessions(
             Guid userId,
             string deviceId,
             int? activeWithinSeconds,
             Guid? controllableUserToCheck,
-            bool isApiKey)
+            bool isApiKey,
+            CancellationToken cancellationToken)
         {
-            var result = Sessions;
+            var remote = await GetRemoteEntriesAsync(cancellationToken).ConfigureAwait(false);
+            var ownedElsewhere = remote.Select(entry => entry.Session.Id).ToHashSet(StringComparer.Ordinal);
+
+            // A session this instance only holds a copy of is reported by its owner, whose controllers are
+            // the ones that decide whether it is active and controllable.
+            IEnumerable<SessionInfoDto> result = Sessions
+                .Where(i => !ownedElsewhere.Contains(i.Id))
+                .Select(ToSessionInfoDto)
+                .Concat(remote.Select(entry => entry.Session))
+                .OrderByDescending(i => i.LastActivityDate);
+
             if (!string.IsNullOrEmpty(deviceId))
             {
                 result = result.Where(i => string.Equals(i.DeviceId, deviceId, StringComparison.OrdinalIgnoreCase));
@@ -2115,7 +2452,7 @@ namespace Emby.Server.Implementations.Session
                 if (!userCanControlOthers)
                 {
                     // User cannot control other user's sessions, validate user id.
-                    result = result.Where(i => i.UserId.IsEmpty() || i.ContainsUser(userId));
+                    result = result.Where(i => i.UserId.IsEmpty() || ContainsUser(i, userId));
                 }
 
                 result = result.Where(i =>
@@ -2136,7 +2473,7 @@ namespace Emby.Server.Implementations.Session
             else if (!userIsAdmin)
             {
                 // Request isn't from administrator, limit to "own" sessions.
-                result = result.Where(i => i.UserId.IsEmpty() || i.ContainsUser(userId));
+                result = result.Where(i => i.UserId.IsEmpty() || ContainsUser(i, userId));
             }
 
             if (!userIsAdmin)
@@ -2159,7 +2496,18 @@ namespace Emby.Server.Implementations.Session
                 result = result.Where(i => i.LastActivityDate >= minActiveDate);
             }
 
-            return result.Select(ToSessionInfoDto).ToList();
+            return result.ToList();
+        }
+
+        private static bool ContainsUser(SessionInfoDto session, Guid userId)
+        {
+            if (session.UserId.Equals(userId))
+            {
+                return true;
+            }
+
+            return session.AdditionalUsers is not null
+                && session.AdditionalUsers.Any(i => i.UserId.Equals(userId));
         }
 
         /// <inheritdoc />
@@ -2234,6 +2582,12 @@ namespace Emby.Server.Implementations.Session
                 _inactiveTimer = null;
             }
 
+            if (_directoryTimer is not null)
+            {
+                await _directoryTimer.DisposeAsync().ConfigureAwait(false);
+                _directoryTimer = null;
+            }
+
             await _shutdownCallback.DisposeAsync().ConfigureAwait(false);
 
             _deviceManager.DeviceOptionsUpdated -= OnDeviceManagerDeviceOptionsUpdated;
@@ -2257,6 +2611,7 @@ namespace Emby.Server.Implementations.Session
             // Close open websockets to allow Kestrel to shut down cleanly
             foreach (var session in _activeConnections.Values)
             {
+                await RemoveFromDirectoryAsync(session).ConfigureAwait(false);
                 await session.DisposeAsync().ConfigureAwait(false);
             }
 

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/QuickConnectController.cs

@@ -59,7 +59,7 @@ public class QuickConnectController : BaseJellyfinApiController
         try
         {
             var auth = await _authContext.GetAuthorizationInfo(Request).ConfigureAwait(false);
-            return await _quickConnect.TryConnect(auth).ConfigureAwait(false);
+            return _quickConnect.TryConnect(auth);
         }
         catch (AuthenticationException)
         {
@@ -77,11 +77,11 @@ public class QuickConnectController : BaseJellyfinApiController
     [HttpGet("Connect")]
     [ProducesResponseType(StatusCodes.Status200OK)]
     [ProducesResponseType(StatusCodes.Status404NotFound)]
-    public async Task<ActionResult<QuickConnectResult>> GetQuickConnectState([FromQuery, Required] string secret)
+    public ActionResult<QuickConnectResult> GetQuickConnectState([FromQuery, Required] string secret)
     {
         try
         {
-            return await _quickConnect.CheckRequestStatus(secret).ConfigureAwait(false);
+            return _quickConnect.CheckRequestStatus(secret);
         }
         catch (ResourceNotFoundException)
         {

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

Jellyfin.Api/Controllers/UserController.cs

@@ -245,11 +245,11 @@ public class UserController : BaseJellyfinApiController
     [HttpPost("AuthenticateWithQuickConnect")]
     [ProducesResponseType(StatusCodes.Status200OK)]
     [Tags("Authentication")]
-    public async Task<ActionResult<AuthenticationResult>> AuthenticateWithQuickConnect([FromBody, Required] QuickConnectDto request)
+    public ActionResult<AuthenticationResult> AuthenticateWithQuickConnect([FromBody, Required] QuickConnectDto request)
     {
         try
         {
-            return await _quickConnectManager.GetAuthorizedRequest(request.Secret).ConfigureAwait(false);
+            return _quickConnectManager.GetAuthorizedRequest(request.Secret);
         }
         catch (SecurityException e)
         {

Jellyfin.Server/CoreAppHost.cs

@@ -116,9 +116,9 @@ namespace Jellyfin.Server
             // to the other instances. Redis-backed when configured, no-op otherwise.
             serviceCollection.AddConfigurationInvalidationBus(_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);
+            // 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);
 
             foreach (var type in GetExportTypes<ILyricProvider>())
             {

Jellyfin.Server/Extensions/QuickConnectStoreServiceCollectionExtensions.cs

@@ -1,72 +0,0 @@
-using System;
-using Emby.Server.Implementations.QuickConnect;
-using MediaBrowser.Controller.MediaEncoding;
-using MediaBrowser.Controller.QuickConnect;
-using Microsoft.Extensions.Configuration;
-using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.Logging;
-using StackExchange.Redis;
-
-namespace Jellyfin.Server.Extensions;
-
-/// <summary>
-/// Extensions for registering the quick connect store.
-/// </summary>
-public static class QuickConnectStoreServiceCollectionExtensions
-{
-    /// <summary>
-    /// Registers the quick connect store, Redis-backed when a connection string is configured and
-    /// process-local otherwise, and reports the selected store at <see cref="LogLevel.Information"/>.
-    /// </summary>
-    /// <remarks>
-    /// The connection string is only set for a multi-instance deployment, which is the only shape where
-    /// the initiate, authorize and exchange legs of one flow can land on different instances.
-    /// </remarks>
-    /// <param name="serviceCollection">The service collection.</param>
-    /// <param name="configuration">The configuration to read the Redis connection string from.</param>
-    /// <param name="logger">The logger to report the selected store on.</param>
-    /// <returns>The updated service collection.</returns>
-    public static IServiceCollection AddQuickConnectStore(
-        this IServiceCollection serviceCollection,
-        IConfiguration configuration,
-        ILogger logger)
-    {
-        ArgumentNullException.ThrowIfNull(configuration);
-        ArgumentNullException.ThrowIfNull(logger);
-
-        if (string.IsNullOrEmpty(configuration[TranscodeStoreOptions.RedisConnectionStringKey]))
-        {
-            logger.LogInformation(
-                "Quick connect store: {Store}. A quick connect flow has to complete against one instance; set {Key} to share it.",
-                nameof(InMemoryQuickConnectStore),
-                TranscodeStoreOptions.RedisConnectionStringKey);
-
-            return serviceCollection.AddSingleton<IQuickConnectStore, InMemoryQuickConnectStore>();
-        }
-
-        logger.LogInformation(
-            "Quick connect store: {Store}. Quick connect flows complete across any instance.",
-            nameof(RedisQuickConnectStore));
-
-        return serviceCollection.AddSingleton<IQuickConnectStore>(sp =>
-        {
-            try
-            {
-                return new RedisQuickConnectStore(
-                    sp.GetRequiredService<IConnectionMultiplexer>(),
-                    sp.GetRequiredService<ILogger<RedisQuickConnectStore>>());
-            }
-            catch (Exception ex)
-            {
-                // Fail open: an unreachable Redis degrades to the single-instance behaviour of a flow
-                // having to complete against one instance, rather than taking quick connect down.
-                sp.GetRequiredService<ILogger<CoreAppHost>>().LogError(
-                    ex,
-                    "Redis is configured but unavailable, so quick connect flows will not complete across instances. Check {Key}.",
-                    TranscodeStoreOptions.RedisConnectionStringKey);
-
-                return new InMemoryQuickConnectStore();
-            }
-        });
-    }
-}

Jellyfin.Server/Extensions/SessionDirectoryServiceCollectionExtensions.cs

@@ -0,0 +1,83 @@
+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 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);
+
+            serviceCollection.AddSingleton<ISessionDirectory>(NullSessionDirectory.Instance);
+            return serviceCollection.AddSingleton<IPodMessageBus>(NullPodMessageBus.Instance);
+        }
+
+        logger.LogInformation(
+            "Session directory: {Directory}. Sessions are visible to, and controllable from, every instance.",
+            nameof(RedisSessionDirectory));
+
+        serviceCollection.AddSingleton<IPodMessageBus>(sp => Create<IPodMessageBus>(
+            sp,
+            () => new RedisPodMessageBus(
+                sp.GetRequiredService<IConnectionMultiplexer>(),
+                sp.GetRequiredService<ILogger<RedisPodMessageBus>>()),
+            NullPodMessageBus.Instance));
+
+        return serviceCollection.AddSingleton<ISessionDirectory>(sp => Create<ISessionDirectory>(
+            sp,
+            () => new RedisSessionDirectory(
+                sp.GetRequiredService<IConnectionMultiplexer>(),
+                sp.GetRequiredService<Microsoft.Extensions.Options.IOptions<SessionDirectoryOptions>>(),
+                sp.GetRequiredService<ILogger<RedisSessionDirectory>>()),
+            NullSessionDirectory.Instance));
+    }
+
+    // Fail open: an unreachable Redis degrades to the single-instance behaviour rather than aborting startup.
+    private static T Create<T>(IServiceProvider serviceProvider, Func<T> factory, T fallback)
+    {
+        try
+        {
+            return factory();
+        }
+        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 fallback;
+        }
+    }
+}

MediaBrowser.Controller/QuickConnect/IQuickConnect.cs

@@ -21,14 +21,14 @@ namespace MediaBrowser.Controller.QuickConnect
         /// </summary>
         /// <param name="authorizationInfo">The initiator authorization info.</param>
         /// <returns>A quick connect result with tokens to proceed or throws an exception if not active.</returns>
-        Task<QuickConnectResult> TryConnect(AuthorizationInfo authorizationInfo);
+        QuickConnectResult TryConnect(AuthorizationInfo authorizationInfo);
 
         /// <summary>
         /// Checks the status of an individual request.
         /// </summary>
         /// <param name="secret">Unique secret identifier of the request.</param>
         /// <returns>Quick connect result.</returns>
-        Task<QuickConnectResult> CheckRequestStatus(string secret);
+        QuickConnectResult CheckRequestStatus(string secret);
 
         /// <summary>
         /// Authorizes a quick connect request to connect as the calling user.
@@ -43,6 +43,6 @@ namespace MediaBrowser.Controller.QuickConnect
         /// </summary>
         /// <param name="secret">The secret.</param>
         /// <returns>The authentication result.</returns>
-        Task<AuthenticationResult> GetAuthorizedRequest(string secret);
+        AuthenticationResult GetAuthorizedRequest(string secret);
     }
 }

MediaBrowser.Controller/QuickConnect/IQuickConnectStore.cs

@@ -1,70 +0,0 @@
-using System;
-using System.Threading;
-using System.Threading.Tasks;
-using MediaBrowser.Controller.Authentication;
-using MediaBrowser.Model.QuickConnect;
-
-namespace MediaBrowser.Controller.QuickConnect;
-
-/// <summary>
-/// Holds the state of in-flight quick connect requests. The three legs of a quick connect flow -
-/// initiate, authorize and exchange - can each land on a different instance, so the state has to be
-/// reachable from all of them.
-/// </summary>
-public interface IQuickConnectStore
-{
-    /// <summary>
-    /// Looks up a pending request by the secret handed to the initiating client.
-    /// </summary>
-    /// <param name="secret">The request secret.</param>
-    /// <param name="cancellationToken">A cancellation token.</param>
-    /// <returns>The request, or <c>null</c> when it is unknown or has expired.</returns>
-    Task<QuickConnectResult?> GetRequestBySecretAsync(string secret, CancellationToken cancellationToken = default);
-
-    /// <summary>
-    /// Looks up a pending request by the code shown to the user.
-    /// </summary>
-    /// <param name="code">The user facing code.</param>
-    /// <param name="cancellationToken">A cancellation token.</param>
-    /// <returns>The request, or <c>null</c> when it is unknown or has expired.</returns>
-    Task<QuickConnectResult?> GetRequestByCodeAsync(string code, CancellationToken cancellationToken = default);
-
-    /// <summary>
-    /// Stores a new or updated request until <paramref name="expiresUtc"/>.
-    /// </summary>
-    /// <param name="request">The request to store.</param>
-    /// <param name="expiresUtc">The instant the request stops being resolvable.</param>
-    /// <param name="cancellationToken">A cancellation token.</param>
-    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
-    Task SetRequestAsync(QuickConnectResult request, DateTime expiresUtc, CancellationToken cancellationToken = default);
-
-    /// <summary>
-    /// Atomically claims the sole right to authorize the request behind <paramref name="secret"/>, so
-    /// that two instances racing on one code cannot both mint an access token. The claim is never
-    /// released: a mint that failed after writing its token would otherwise be retried into a second one.
-    /// </summary>
-    /// <param name="secret">The request secret.</param>
-    /// <param name="expiresUtc">The instant the claim lapses, after which the request can be authorized again.</param>
-    /// <param name="cancellationToken">A cancellation token.</param>
-    /// <returns><c>true</c> when this caller may go on to authorize the request; <c>false</c> when it is unknown, expired, already authorized or being authorized elsewhere.</returns>
-    Task<bool> TryClaimAuthorizationAsync(string secret, DateTime expiresUtc, CancellationToken cancellationToken = default);
-
-    /// <summary>
-    /// Stores the authentication minted for an authorized request until <paramref name="expiresUtc"/>.
-    /// </summary>
-    /// <param name="secret">The request secret the client exchanges.</param>
-    /// <param name="authenticationResult">The authentication to hand out.</param>
-    /// <param name="expiresUtc">The instant the authentication stops being exchangeable.</param>
-    /// <param name="cancellationToken">A cancellation token.</param>
-    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
-    Task SetAuthorizationAsync(string secret, AuthenticationResult authenticationResult, DateTime expiresUtc, CancellationToken cancellationToken = default);
-
-    /// <summary>
-    /// Atomically takes the authentication for <paramref name="secret"/> and removes it, so that two
-    /// instances racing on the same secret cannot both hand out an access token.
-    /// </summary>
-    /// <param name="secret">The request secret.</param>
-    /// <param name="cancellationToken">A cancellation token.</param>
-    /// <returns>The authentication, or <c>null</c> when the secret is unknown, expired or already exchanged.</returns>
-    Task<AuthenticationResult?> TryConsumeAuthorizationAsync(string secret, CancellationToken cancellationToken = default);
-}

MediaBrowser.Controller/QuickConnect/InMemoryQuickConnectStore.cs

@@ -1,110 +0,0 @@
-using System;
-using System.Collections.Concurrent;
-using System.Linq;
-using System.Threading;
-using System.Threading.Tasks;
-using MediaBrowser.Controller.Authentication;
-using MediaBrowser.Model.QuickConnect;
-
-namespace MediaBrowser.Controller.QuickConnect;
-
-/// <summary>
-/// A process-local <see cref="IQuickConnectStore"/>. It is the single-instance default, and the
-/// fallback a shared store degrades to while its backend is unreachable, so quick connect keeps
-/// working for clients whose three legs happen to land on one instance.
-/// </summary>
-public sealed class InMemoryQuickConnectStore : IQuickConnectStore
-{
-    private readonly ConcurrentDictionary<string, Entry<QuickConnectResult>> _requests = new(StringComparer.Ordinal);
-    private readonly ConcurrentDictionary<string, Entry<AuthenticationResult>> _authorizations = new(StringComparer.Ordinal);
-    private readonly ConcurrentDictionary<string, DateTime> _authorizationClaims = new(StringComparer.Ordinal);
-
-    /// <inheritdoc />
-    public Task<QuickConnectResult?> GetRequestBySecretAsync(string secret, CancellationToken cancellationToken = default)
-    {
-        Expire();
-        return Task.FromResult(_requests.TryGetValue(secret, out var entry) ? entry.Value : null);
-    }
-
-    /// <inheritdoc />
-    public Task<QuickConnectResult?> GetRequestByCodeAsync(string code, CancellationToken cancellationToken = default)
-    {
-        Expire();
-        return Task.FromResult(_requests.Values
-            .Select(entry => entry.Value)
-            .FirstOrDefault(request => string.Equals(request.Code, code, StringComparison.Ordinal)));
-    }
-
-    /// <inheritdoc />
-    public Task SetRequestAsync(QuickConnectResult request, DateTime expiresUtc, CancellationToken cancellationToken = default)
-    {
-        ArgumentNullException.ThrowIfNull(request);
-
-        Expire();
-        _requests[request.Secret] = new Entry<QuickConnectResult>(expiresUtc, request);
-        return Task.CompletedTask;
-    }
-
-    /// <inheritdoc />
-    public Task<bool> TryClaimAuthorizationAsync(string secret, DateTime expiresUtc, CancellationToken cancellationToken = default)
-    {
-        Expire();
-        if (!_requests.TryGetValue(secret, out var entry) || entry.Value.Authenticated)
-        {
-            return Task.FromResult(false);
-        }
-
-        return Task.FromResult(_authorizationClaims.TryAdd(secret, expiresUtc));
-    }
-
-    /// <inheritdoc />
-    public Task SetAuthorizationAsync(string secret, AuthenticationResult authenticationResult, DateTime expiresUtc, CancellationToken cancellationToken = default)
-    {
-        Expire();
-        _authorizations[secret] = new Entry<AuthenticationResult>(expiresUtc, authenticationResult);
-        return Task.CompletedTask;
-    }
-
-    /// <inheritdoc />
-    public Task<AuthenticationResult?> TryConsumeAuthorizationAsync(string secret, CancellationToken cancellationToken = default)
-    {
-        Expire();
-        if (!_authorizations.TryRemove(secret, out var entry) || entry.ExpiresUtc <= DateTime.UtcNow)
-        {
-            return Task.FromResult<AuthenticationResult?>(null);
-        }
-
-        return Task.FromResult<AuthenticationResult?>(entry.Value);
-    }
-
-    private void Expire()
-    {
-        var now = DateTime.UtcNow;
-
-        foreach (var (secret, entry) in _requests)
-        {
-            if (entry.ExpiresUtc <= now)
-            {
-                _requests.TryRemove(secret, out _);
-            }
-        }
-
-        foreach (var (secret, entry) in _authorizations)
-        {
-            if (entry.ExpiresUtc <= now)
-            {
-                _authorizations.TryRemove(secret, out _);
-            }
-        }
-
-        foreach (var (secret, expiresUtc) in _authorizationClaims)
-        {
-            if (expiresUtc <= now)
-            {
-                _authorizationClaims.TryRemove(secret, out _);
-            }
-        }
-    }
-
-    private sealed record Entry<T>(DateTime ExpiresUtc, T Value);
-}

MediaBrowser.Controller/Session/IPodMessageBus.cs

@@ -0,0 +1,33 @@
+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 reports how many listeners took it, so that a message
+    /// addressed to an instance that is no longer there is not mistaken for a delivered one.
+    /// </summary>
+    /// <param name="targetPod">The instance to deliver to.</param>
+    /// <param name="message">The message.</param>
+    /// <param name="cancellationToken">The cancellation token.</param>
+    /// <returns>The number of instances the message reached.</returns>
+    Task<long> PublishAsync(string targetPod, PodMessage message, CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// Registers a handler for the messages addressed to this instance.
+    /// </summary>
+    /// <param name="handler">The handler.</param>
+    void Subscribe(Func<PodMessage, Task> handler);
+}

MediaBrowser.Controller/Session/ISessionDirectory.cs

@@ -0,0 +1,47 @@
+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>
+    /// 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="connectedUtcTicks">When the publishing instance's connection to the session was established, 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 connectedUtcTicks, 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>
+    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.
@@ -306,8 +309,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<long> PublishAsync(string targetPod, PodMessage message, CancellationToken cancellationToken = default)
+        => Task.FromResult(0L);
+
+    /// <inheritdoc />
+    public void Subscribe(Func<PodMessage, Task> handler)
+    {
+    }
+}

MediaBrowser.Controller/Session/NullSessionDirectory.cs

@@ -0,0 +1,34 @@
+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<bool> PublishAsync(SessionDirectoryEntry entry, long connectedUtcTicks, 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,23 @@
+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>
+    /// 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 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/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/QuickConnect/QuickConnectManagerTests.cs

@@ -8,7 +8,6 @@ using MediaBrowser.Common.Extensions;
 using MediaBrowser.Controller.Authentication;
 using MediaBrowser.Controller.Configuration;
 using MediaBrowser.Controller.Net;
-using MediaBrowser.Controller.QuickConnect;
 using MediaBrowser.Model.Configuration;
 using Moq;
 using Xunit;
@@ -41,8 +40,6 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect
                 ConfigureMembers = true
             }).Inject(configManager.Object);
 
-            _fixture.Inject<IQuickConnectStore>(new InMemoryQuickConnectStore());
-
             // User object contains circular references.
             _fixture.Behaviors.OfType<ThrowingRecursionBehavior>().ToList()
                 .ForEach(b => _fixture.Behaviors.Remove(b));
@@ -63,8 +60,8 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect
         [InlineData("Device", "", "Client", "1.0.0")]
         [InlineData("Device", "DeviceId", "", "1.0.0")]
         [InlineData("Device", "DeviceId", "Client", "")]
-        public async Task TryConnect_InvalidAuthorizationInfo_ThrowsArgumentException(string device, string deviceId, string client, string version)
-            => await Assert.ThrowsAsync<ArgumentException>(() => _quickConnectManager.TryConnect(
+        public void TryConnect_InvalidAuthorizationInfo_ThrowsArgumentException(string device, string deviceId, string client, string version)
+            => Assert.Throws<ArgumentException>(() => _quickConnectManager.TryConnect(
                 new AuthorizationInfo
                 {
                     Device = device,
@@ -74,17 +71,17 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect
                 }));
 
         [Fact]
-        public async Task TryConnect_QuickConnectUnavailable_ThrowsAuthenticationException()
+        public void TryConnect_QuickConnectUnavailable_ThrowsAuthenticationException()
         {
             _config.QuickConnectAvailable = false;
-            await Assert.ThrowsAsync<AuthenticationException>(() => _quickConnectManager.TryConnect(_quickConnectAuthInfo));
+            Assert.Throws<AuthenticationException>(() => _quickConnectManager.TryConnect(_quickConnectAuthInfo));
         }
 
         [Fact]
-        public async Task CheckRequestStatus_QuickConnectUnavailable_ThrowsAuthenticationException()
+        public void CheckRequestStatus_QuickConnectUnavailable_ThrowsAuthenticationException()
         {
             _config.QuickConnectAvailable = false;
-            await Assert.ThrowsAsync<AuthenticationException>(() => _quickConnectManager.CheckRequestStatus(string.Empty));
+            Assert.Throws<AuthenticationException>(() => _quickConnectManager.CheckRequestStatus(string.Empty));
         }
 
         [Fact]
@@ -95,10 +92,10 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect
         }
 
         [Fact]
-        public async Task GetAuthorizedRequest_QuickConnectUnavailable_ThrowsAuthenticationException()
+        public void GetAuthorizedRequest_QuickConnectUnavailable_ThrowsAuthenticationException()
         {
             _config.QuickConnectAvailable = false;
-            await Assert.ThrowsAsync<AuthenticationException>(() => _quickConnectManager.GetAuthorizedRequest(string.Empty));
+            Assert.Throws<AuthenticationException>(() => _quickConnectManager.GetAuthorizedRequest(string.Empty));
         }
 
         [Fact]
@@ -109,71 +106,34 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect
         }
 
         [Fact]
-        public async Task CheckRequestStatus_QuickConnectAvailable_Success()
+        public void CheckRequestStatus_QuickConnectAvailable_Success()
         {
             _config.QuickConnectAvailable = true;
-            var res1 = await _quickConnectManager.TryConnect(_quickConnectAuthInfo);
-            var res2 = await _quickConnectManager.CheckRequestStatus(res1.Secret);
-            Assert.Equal(res1.Secret, res2.Secret);
-            Assert.Equal(res1.Code, res2.Code);
+            var res1 = _quickConnectManager.TryConnect(_quickConnectAuthInfo);
+            var res2 = _quickConnectManager.CheckRequestStatus(res1.Secret);
+            Assert.Equal(res1, res2);
         }
 
         [Fact]
-        public async Task CheckRequestStatus_UnknownSecret_ThrowsResourceNotFoundException()
+        public void CheckRequestStatus_UnknownSecret_ThrowsResourceNotFoundException()
         {
             _config.QuickConnectAvailable = true;
-            await Assert.ThrowsAsync<ResourceNotFoundException>(() => _quickConnectManager.CheckRequestStatus("Unknown secret"));
+            Assert.Throws<ResourceNotFoundException>(() => _quickConnectManager.CheckRequestStatus("Unknown secret"));
         }
 
         [Fact]
-        public async Task GetAuthorizedRequest_UnknownSecret_ThrowsResourceNotFoundException()
+        public void GetAuthorizedRequest_UnknownSecret_ThrowsResourceNotFoundException()
         {
             _config.QuickConnectAvailable = true;
-            await Assert.ThrowsAsync<ResourceNotFoundException>(() => _quickConnectManager.GetAuthorizedRequest("Unknown secret"));
+            Assert.Throws<ResourceNotFoundException>(() => _quickConnectManager.GetAuthorizedRequest("Unknown secret"));
         }
 
         [Fact]
         public async Task AuthorizeRequest_QuickConnectAvailable_Success()
         {
             _config.QuickConnectAvailable = true;
-            var res = await _quickConnectManager.TryConnect(_quickConnectAuthInfo);
+            var res = _quickConnectManager.TryConnect(_quickConnectAuthInfo);
             Assert.True(await _quickConnectManager.AuthorizeRequest(Guid.Empty, res.Code));
         }
-
-        [Fact]
-        public async Task AuthorizeRequest_RacedOnOneCode_SucceedsOnce()
-        {
-            _config.QuickConnectAvailable = true;
-            var res = await _quickConnectManager.TryConnect(_quickConnectAuthInfo);
-
-            var outcomes = await Task.WhenAll(
-                Task.Run(() => AuthorizeAsync(res.Code)),
-                Task.Run(() => AuthorizeAsync(res.Code)));
-
-            Assert.Single(outcomes, authorized => authorized);
-        }
-
-        [Fact]
-        public async Task GetAuthorizedRequest_SecondExchange_ThrowsResourceNotFoundException()
-        {
-            _config.QuickConnectAvailable = true;
-            var res = await _quickConnectManager.TryConnect(_quickConnectAuthInfo);
-            await _quickConnectManager.AuthorizeRequest(Guid.Empty, res.Code);
-
-            Assert.NotNull(await _quickConnectManager.GetAuthorizedRequest(res.Secret));
-            await Assert.ThrowsAsync<ResourceNotFoundException>(() => _quickConnectManager.GetAuthorizedRequest(res.Secret));
-        }
-
-        private async Task<bool> AuthorizeAsync(string code)
-        {
-            try
-            {
-                return await _quickConnectManager.AuthorizeRequest(Guid.Empty, code).ConfigureAwait(false);
-            }
-            catch (InvalidOperationException)
-            {
-                return false;
-            }
-        }
     }
 }

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));
     }
@@ -173,7 +180,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 +193,7 @@ public class SessionManagerTests
 
         var adminSession = await LogSessionActivity(sessionManager, admin);
 
-        sessionManager.AddAdditionalUser(adminSession.Id, adminSession.Id, guest.Id);
+        await sessionManager.AddAdditionalUser(adminSession.Id, adminSession.Id, guest.Id);
 
         Assert.Contains(adminSession.AdditionalUsers, i => i.UserId.Equals(guest.Id));
     }
@@ -201,7 +208,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]
@@ -238,7 +245,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()));
     }
 
     // All sessions are logged with the same client and device id on purpose, those values are taken

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

@@ -13,7 +13,6 @@ namespace Jellyfin.Server.Tests.HighAvailability;
 /// startup configuration the host reads them from must therefore accept that form; when it does not,
 /// a correctly set variable is dropped and the feature it configures stays off without any error.
 /// </summary>
-[Collection("JellyfinSectionConfiguration")]
 public sealed class JellyfinSectionConfigurationTests : IDisposable
 {
     private const string RedisKey = "Jellyfin:TranscodeStore:RedisConnectionString";

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

@@ -1,170 +0,0 @@
-using System;
-using System.Collections.Concurrent;
-using System.Globalization;
-using System.IO;
-using System.Net;
-using System.Net.Sockets;
-using System.Threading;
-using System.Threading.Tasks;
-using StackExchange.Redis;
-
-namespace Jellyfin.Server.Tests.HighAvailability;
-
-/// <summary>
-/// A loopback TCP proxy in front of a Redis server. Cutting it drops every connection through it and
-/// refuses new ones, so a test can take Redis away from one instance mid-flow - and give it back - the
-/// way a restarted valkey does, and watch what a real StackExchange.Redis client makes of it.
-/// </summary>
-public sealed class RedisFaultProxy : IAsyncDisposable
-{
-    private readonly ConcurrentDictionary<TcpClient, byte> _live = new();
-    private readonly CancellationTokenSource _cts = new();
-    private readonly TcpListener _listener;
-    private readonly string _targetHost;
-    private readonly int _targetPort;
-    private readonly int _port;
-
-    private volatile bool _cut;
-
-    private RedisFaultProxy(TcpListener listener, int port, string targetHost, int targetPort)
-    {
-        _listener = listener;
-        _port = port;
-        _targetHost = targetHost;
-        _targetPort = targetPort;
-    }
-
-    /// <summary>
-    /// Gets a connection string pointing at the proxy. The timeouts are short so a cut surfaces as a
-    /// failure in seconds rather than in the library's minute-scale defaults.
-    /// </summary>
-    public string ConnectionString => string.Create(
-        CultureInfo.InvariantCulture,
-        $"127.0.0.1:{_port},abortConnect=false,connectTimeout=500,syncTimeout=2000,connectRetry=1");
-
-    /// <summary>
-    /// Starts a proxy in front of the server named by <paramref name="target"/>.
-    /// </summary>
-    /// <param name="target">The connection string of the server to forward to.</param>
-    /// <returns>The running proxy.</returns>
-    public static RedisFaultProxy Start(string target)
-    {
-        var endpoint = ConfigurationOptions.Parse(target).EndPoints[0];
-        var (host, port) = endpoint switch
-        {
-            DnsEndPoint dns => (dns.Host, dns.Port),
-            IPEndPoint ip => (ip.Address.ToString(), ip.Port),
-            _ => throw new NotSupportedException("Unsupported endpoint " + endpoint)
-        };
-
-        var listener = new TcpListener(IPAddress.Loopback, 0);
-        listener.Start();
-
-        var proxy = new RedisFaultProxy(listener, ((IPEndPoint)listener.LocalEndpoint).Port, host, port);
-        _ = Task.Run(proxy.AcceptAsync);
-        return proxy;
-    }
-
-    /// <summary>
-    /// Takes Redis away from everything connected through the proxy.
-    /// </summary>
-    public void Cut()
-    {
-        _cut = true;
-        DropLiveConnections();
-    }
-
-    /// <summary>
-    /// Lets connections through again. Clients reconnect on their own schedule, so callers have to wait
-    /// for the connection to come back rather than assume it already has.
-    /// </summary>
-    public void Restore() => _cut = false;
-
-    /// <inheritdoc/>
-    public async ValueTask DisposeAsync()
-    {
-        _cut = true;
-        await _cts.CancelAsync().ConfigureAwait(false);
-        _listener.Stop();
-        DropLiveConnections();
-        _cts.Dispose();
-    }
-
-    private void DropLiveConnections()
-    {
-        foreach (var client in _live.Keys)
-        {
-            if (_live.TryRemove(client, out _))
-            {
-                client.Dispose();
-            }
-        }
-    }
-
-    private async Task AcceptAsync()
-    {
-        while (!_cts.IsCancellationRequested)
-        {
-            TcpClient client;
-            try
-            {
-                client = await _listener.AcceptTcpClientAsync(_cts.Token).ConfigureAwait(false);
-            }
-            catch (Exception exception) when (exception is OperationCanceledException or SocketException or ObjectDisposedException)
-            {
-                return;
-            }
-
-            if (_cut)
-            {
-                client.Dispose();
-                continue;
-            }
-
-            _ = Task.Run(() => ForwardAsync(client));
-        }
-    }
-
-    private async Task ForwardAsync(TcpClient client)
-    {
-        TcpClient? upstream = null;
-        try
-        {
-            upstream = new TcpClient();
-            await upstream.ConnectAsync(_targetHost, _targetPort, _cts.Token).ConfigureAwait(false);
-
-            _live[client] = 0;
-            _live[upstream] = 0;
-
-            var clientStream = client.GetStream();
-            var upstreamStream = upstream.GetStream();
-            await Task.WhenAny(
-                CopyAsync(clientStream, upstreamStream),
-                CopyAsync(upstreamStream, clientStream)).ConfigureAwait(false);
-        }
-        catch (Exception exception) when (exception is IOException or SocketException or OperationCanceledException or ObjectDisposedException)
-        {
-        }
-        finally
-        {
-            _live.TryRemove(client, out _);
-            client.Dispose();
-            if (upstream is not null)
-            {
-                _live.TryRemove(upstream, out _);
-                upstream.Dispose();
-            }
-        }
-    }
-
-    private async Task CopyAsync(NetworkStream from, NetworkStream to)
-    {
-        try
-        {
-            await from.CopyToAsync(to, _cts.Token).ConfigureAwait(false);
-        }
-        catch (Exception exception) when (exception is IOException or SocketException or OperationCanceledException or ObjectDisposedException)
-        {
-        }
-    }
-}

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 connects to it.
     /// </summary>
     /// <returns>The running server.</returns>
     public static async Task<RedisTestServer> StartAsync()
@@ -39,24 +39,19 @@ public sealed class RedisTestServer : IAsyncDisposable
         var provided = Environment.GetEnvironmentVariable(ConnectionStringVariable);
         if (!string.IsNullOrWhiteSpace(provided))
         {
-            var attached = new RedisTestServer(null, provided);
-            await attached.WaitUntilReadyAsync().ConfigureAwait(false);
-            return attached;
+            return new RedisTestServer(null, provided);
         }
 
-        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());
-        await started.WaitUntilReadyAsync().ConfigureAwait(false);
-        return started;
+        return new RedisTestServer(container, container.GetConnectionString());
     }
 
     /// <summary>
-    /// Opens a connection of its own, so each in-process stand-in for a replica talks to the server the
-    /// way a separate pod would.
+    /// Opens a connection to the server.
     /// </summary>
-    /// <returns>A new multiplexer.</returns>
+    /// <returns>The connection.</returns>
     public async Task<IConnectionMultiplexer> ConnectAsync()
         => await ConnectionMultiplexer.ConnectAsync(ConnectionString).ConfigureAwait(false);
 
@@ -68,24 +63,4 @@ public sealed class RedisTestServer : IAsyncDisposable
             await _container.DisposeAsync().ConfigureAwait(false);
         }
     }
-
-    private async Task WaitUntilReadyAsync()
-    {
-        for (var attempt = 1; ; attempt++)
-        {
-            try
-            {
-                var connection = await ConnectionMultiplexer.ConnectAsync(ConnectionString).ConfigureAwait(false);
-                await using (connection.ConfigureAwait(false))
-                {
-                    await connection.GetDatabase().PingAsync().ConfigureAwait(false);
-                    return;
-                }
-            }
-            catch (RedisException) when (attempt < 60)
-            {
-                await Task.Delay(TimeSpan.FromSeconds(1)).ConfigureAwait(false);
-            }
-        }
-    }
 }

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

@@ -0,0 +1,449 @@
+using System;
+using System.Linq;
+using System.Text.Json;
+using System.Threading;
+using System.Threading.Tasks;
+using Jellyfin.Database.Implementations;
+using Jellyfin.Database.Implementations.DbConfiguration;
+using Jellyfin.Database.Implementations.Entities;
+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.Events;
+using MediaBrowser.Controller.Library;
+using MediaBrowser.Controller.Session;
+using MediaBrowser.Model.Session;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.Hosting;
+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>
+    /// 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>
+    /// 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 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 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 SessionManager CreateReplica(string podId, SessionDirectoryOptions? options = 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));
+    }
+
+    private SessionManager CreateReplica(string podId, ISessionDirectory directory, IPodMessageBus bus)
+        => CreateReplica(podId, new SessionDirectoryOptions(), directory, bus);
+
+    private SessionManager CreateReplica(string podId, SessionDirectoryOptions options, ISessionDirectory directory, IPodMessageBus bus)
+    {
+        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);
+
+        return new SessionManager(
+            NullLogger<SessionManager>.Instance,
+            Mock.Of<IEventManager>(),
+            Mock.Of<IUserDataManager>(),
+            Mock.Of<IServerConfigurationManager>(),
+            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));
+    }
+
+    // The bus reads the instance identity from the environment, so the two replicas are built one at a time.
+    private IPodMessageBus CreateBus(string podId)
+    {
+        var previous = Environment.GetEnvironmentVariable("JELLYFIN_INSTANCE_ID");
+        Environment.SetEnvironmentVariable("JELLYFIN_INSTANCE_ID", podId);
+        try
+        {
+            return new RedisPodMessageBus(
+                _connection,
+                NullLogger<RedisPodMessageBus>.Instance);
+        }
+        finally
+        {
+            Environment.SetEnvironmentVariable("JELLYFIN_INSTANCE_ID", previous);
+        }
+    }
+
+    /// <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>
+    /// 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 => true;
+
+        public bool SupportsMediaControl => true;
+
+        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" />

tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectReplicaTests.cs

@@ -1,369 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.Globalization;
-using System.Threading;
-using System.Threading.Tasks;
-using Emby.Server.Implementations.QuickConnect;
-using Jellyfin.Data.Queries;
-using Jellyfin.Database.Implementations;
-using Jellyfin.Database.Implementations.DbConfiguration;
-using Jellyfin.Database.Implementations.Entities;
-using Jellyfin.Database.Implementations.Entities.Security;
-using Jellyfin.Database.Implementations.Locking;
-using Jellyfin.Database.Providers.PostgreSQL;
-using Jellyfin.Server.Implementations.Devices;
-using Jellyfin.Server.Tests.HighAvailability;
-using Jellyfin.Server.Tests.Migrations;
-using MediaBrowser.Common.Extensions;
-using MediaBrowser.Controller.Authentication;
-using MediaBrowser.Controller.Configuration;
-using MediaBrowser.Controller.Devices;
-using MediaBrowser.Controller.Library;
-using MediaBrowser.Controller.Net;
-using MediaBrowser.Controller.QuickConnect;
-using MediaBrowser.Controller.Session;
-using MediaBrowser.Model.Configuration;
-using MediaBrowser.Model.Dto;
-using MediaBrowser.Model.QuickConnect;
-using Microsoft.EntityFrameworkCore;
-using Microsoft.Extensions.Logging.Abstractions;
-using Moq;
-using Npgsql;
-using StackExchange.Redis;
-using Xunit;
-
-namespace Jellyfin.Server.Tests.QuickConnect;
-
-/// <summary>
-/// Three independently constructed <see cref="QuickConnectManager"/> instances over one PostgreSQL
-/// database and one Redis are the in-process stand-in for three replicas without sticky sessions: the
-/// initiate, authorize and exchange legs of one flow each land on a different one.
-/// </summary>
-[Trait("Category", "RequiresDocker")]
-public sealed class QuickConnectReplicaTests : IAsyncLifetime
-{
-    private static readonly AuthorizationInfo _authorizationInfo = new AuthorizationInfo
-    {
-        Device = "Living Room TV",
-        DeviceId = "device-1",
-        Client = "Jellyfin Web",
-        Version = "1.0.0"
-    };
-
-    private readonly List<IConnectionMultiplexer> _connections = new();
-
-    private PostgreSqlTestServer _postgres = null!;
-    private RedisTestServer _redis = null!;
-
-    /// <inheritdoc/>
-    public async ValueTask InitializeAsync()
-    {
-        _postgres = await PostgreSqlTestServer.StartAsync().ConfigureAwait(false);
-        _redis = await RedisTestServer.StartAsync().ConfigureAwait(false);
-    }
-
-    /// <inheritdoc/>
-    public async ValueTask DisposeAsync()
-    {
-        foreach (var connection in _connections)
-        {
-            await connection.DisposeAsync().ConfigureAwait(false);
-        }
-
-        await _redis.DisposeAsync().ConfigureAwait(false);
-        await _postgres.DisposeAsync().ConfigureAwait(false);
-    }
-
-    /// <summary>
-    /// The three legs of a quick connect flow land on three different replicas, and the token the third
-    /// one hands out is the one the second one minted into the shared database.
-    /// </summary>
-    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
-    [Fact]
-    public async Task InitiateAuthorizeExchange_AcrossThreeReplicas_Succeeds()
-    {
-        var cancellationToken = TestContext.Current.CancellationToken;
-        var connectionString = await _postgres.CreateDatabaseAsync("quickconnect_replica_flow", cancellationToken);
-
-        await using var dataSource = new NpgsqlDataSourceBuilder(connectionString).Build();
-        var user = await CreateSchemaWithUserAsync(dataSource, cancellationToken);
-
-        var replicaA = await CreateReplicaAsync(dataSource, user);
-        var replicaB = await CreateReplicaAsync(dataSource, user);
-        var replicaC = await CreateReplicaAsync(dataSource, user);
-
-        var initiated = await replicaA.Manager.TryConnect(_authorizationInfo);
-
-        // The code is shown to the user on whichever replica serves the dashboard.
-        Assert.True(await replicaB.Manager.AuthorizeRequest(user.Id, initiated.Code));
-
-        var polled = await replicaC.Manager.CheckRequestStatus(initiated.Secret);
-        Assert.True(polled.Authenticated);
-        Assert.Equal(initiated.Code, polled.Code);
-        Assert.Equal(_authorizationInfo.DeviceId, polled.DeviceId);
-
-        var exchanged = await replicaC.Manager.GetAuthorizedRequest(initiated.Secret);
-
-        Assert.False(string.IsNullOrEmpty(exchanged.AccessToken));
-        Assert.Equal(user.Id, exchanged.User.Id);
-
-        var devices = await replicaA.Devices.GetDevices(new DeviceQuery { AccessToken = exchanged.AccessToken });
-        Assert.Equal(user.Id, Assert.Single(devices.Items).UserId);
-    }
-
-    /// <summary>
-    /// A secret is single use across the whole deployment: two replicas racing to exchange it must not
-    /// both hand out an access token. One scheduling of one race settles nothing either way, so the race
-    /// is run repeatedly.
-    /// </summary>
-    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
-    [Fact]
-    public async Task Exchange_RacedOnTwoReplicas_SucceedsOnce()
-    {
-        var cancellationToken = TestContext.Current.CancellationToken;
-        var connectionString = await _postgres.CreateDatabaseAsync("quickconnect_replica_race", cancellationToken);
-
-        await using var dataSource = new NpgsqlDataSourceBuilder(connectionString).Build();
-        var user = await CreateSchemaWithUserAsync(dataSource, cancellationToken);
-
-        var replicaA = await CreateReplicaAsync(dataSource, user);
-        var replicaB = await CreateReplicaAsync(dataSource, user);
-        var replicaC = await CreateReplicaAsync(dataSource, user);
-
-        for (var attempt = 0; attempt < 25; attempt++)
-        {
-            var initiated = await replicaA.Manager.TryConnect(AuthorizationInfoFor(attempt));
-            await replicaB.Manager.AuthorizeRequest(user.Id, initiated.Code);
-
-            var outcomes = await Task.WhenAll(
-                Task.Run(() => ExchangeAsync(replicaA.Manager, initiated.Secret), cancellationToken),
-                Task.Run(() => ExchangeAsync(replicaC.Manager, initiated.Secret), cancellationToken));
-
-            Assert.Single(outcomes, outcome => outcome is not null);
-
-            // And it stays consumed for every later attempt, on any replica.
-            await Assert.ThrowsAsync<ResourceNotFoundException>(() => replicaB.Manager.GetAuthorizedRequest(initiated.Secret));
-        }
-    }
-
-    /// <summary>
-    /// Two replicas authorizing one code at the same time mint one access token between them. A second
-    /// one would be live, attached to the same device and reachable by nobody.
-    /// </summary>
-    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
-    [Fact]
-    public async Task Authorize_RacedOnTwoReplicas_MintsOneAccessToken()
-    {
-        var cancellationToken = TestContext.Current.CancellationToken;
-        var connectionString = await _postgres.CreateDatabaseAsync("quickconnect_replica_authorize_race", cancellationToken);
-
-        await using var dataSource = new NpgsqlDataSourceBuilder(connectionString).Build();
-        var user = await CreateSchemaWithUserAsync(dataSource, cancellationToken);
-
-        var replicaA = await CreateReplicaAsync(dataSource, user);
-        var replicaB = await CreateReplicaAsync(dataSource, user);
-        var replicaC = await CreateReplicaAsync(dataSource, user);
-
-        for (var attempt = 0; attempt < 20; attempt++)
-        {
-            var authorizationInfo = AuthorizationInfoFor(attempt);
-            var initiated = await replicaA.Manager.TryConnect(authorizationInfo);
-
-            var outcomes = await Task.WhenAll(
-                Task.Run(() => AuthorizeAsync(replicaB.Manager, user.Id, initiated.Code), cancellationToken),
-                Task.Run(() => AuthorizeAsync(replicaC.Manager, user.Id, initiated.Code), cancellationToken));
-
-            Assert.Single(outcomes, authorized => authorized);
-
-            var devices = await replicaA.Devices.GetDevices(new DeviceQuery { DeviceId = authorizationInfo.DeviceId });
-            var device = Assert.Single(devices.Items);
-
-            var exchanged = await replicaA.Manager.GetAuthorizedRequest(initiated.Secret);
-            Assert.Equal(device.AccessToken, exchanged.AccessToken);
-        }
-    }
-
-    /// <summary>
-    /// An expired request is rejected on a replica that never saw it created, rather than resolving to a
-    /// stale authorization.
-    /// </summary>
-    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
-    [Fact]
-    public async Task ExpiredRequest_IsRejectedOnEveryReplica()
-    {
-        var cancellationToken = TestContext.Current.CancellationToken;
-        var connectionString = await _postgres.CreateDatabaseAsync("quickconnect_replica_expiry", cancellationToken);
-
-        await using var dataSource = new NpgsqlDataSourceBuilder(connectionString).Build();
-        var user = await CreateSchemaWithUserAsync(dataSource, cancellationToken);
-
-        var replicaA = await CreateReplicaAsync(dataSource, user);
-        var replicaB = await CreateReplicaAsync(dataSource, user);
-
-        var initiated = await replicaA.Manager.TryConnect(_authorizationInfo);
-        Assert.NotNull(await replicaB.Manager.CheckRequestStatus(initiated.Secret));
-
-        // Shorten the stored expiry instead of waiting out the ten minute timeout.
-        await replicaA.Store.SetRequestAsync(initiated, DateTime.UtcNow.AddSeconds(1), cancellationToken);
-        await Task.Delay(TimeSpan.FromSeconds(2), cancellationToken);
-
-        await Assert.ThrowsAsync<ResourceNotFoundException>(() => replicaB.Manager.CheckRequestStatus(initiated.Secret));
-        await Assert.ThrowsAsync<ResourceNotFoundException>(() => replicaB.Manager.AuthorizeRequest(user.Id, initiated.Code));
-    }
-
-    /// <summary>
-    /// An authorization that was never exchanged expires too, so a code authorized and then abandoned
-    /// cannot be redeemed later from another replica.
-    /// </summary>
-    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
-    [Fact]
-    public async Task ExpiredAuthorization_IsRejectedOnEveryReplica()
-    {
-        var cancellationToken = TestContext.Current.CancellationToken;
-        var connectionString = await _postgres.CreateDatabaseAsync("quickconnect_replica_auth_expiry", cancellationToken);
-
-        await using var dataSource = new NpgsqlDataSourceBuilder(connectionString).Build();
-        var user = await CreateSchemaWithUserAsync(dataSource, cancellationToken);
-
-        var replicaA = await CreateReplicaAsync(dataSource, user);
-        var replicaB = await CreateReplicaAsync(dataSource, user);
-
-        var initiated = await replicaA.Manager.TryConnect(_authorizationInfo);
-        await replicaA.Manager.AuthorizeRequest(user.Id, initiated.Code);
-
-        var stored = await replicaA.Store.GetRequestBySecretAsync(initiated.Secret, cancellationToken);
-        Assert.True(stored?.Authenticated);
-
-        await replicaA.Store.SetAuthorizationAsync(
-            initiated.Secret,
-            new AuthenticationResult { AccessToken = "stale" },
-            DateTime.UtcNow.AddSeconds(1),
-            cancellationToken);
-        await Task.Delay(TimeSpan.FromSeconds(2), cancellationToken);
-
-        await Assert.ThrowsAsync<ResourceNotFoundException>(() => replicaB.Manager.GetAuthorizedRequest(initiated.Secret));
-    }
-
-    private static AuthorizationInfo AuthorizationInfoFor(int attempt) => new AuthorizationInfo
-    {
-        Device = _authorizationInfo.Device,
-        DeviceId = string.Create(CultureInfo.InvariantCulture, $"device-{attempt}"),
-        Client = _authorizationInfo.Client,
-        Version = _authorizationInfo.Version
-    };
-
-    private static async Task<bool> AuthorizeAsync(IQuickConnect manager, Guid userId, string code)
-    {
-        try
-        {
-            return await manager.AuthorizeRequest(userId, code).ConfigureAwait(false);
-        }
-        catch (InvalidOperationException)
-        {
-            return false;
-        }
-    }
-
-    private static async Task<AuthenticationResult?> ExchangeAsync(IQuickConnect manager, string secret)
-    {
-        try
-        {
-            return await manager.GetAuthorizedRequest(secret).ConfigureAwait(false);
-        }
-        catch (ResourceNotFoundException)
-        {
-            return null;
-        }
-    }
-
-    private static async Task<User> CreateSchemaWithUserAsync(NpgsqlDataSource dataSource, CancellationToken cancellationToken)
-    {
-        var context = CreateContext(dataSource);
-        await using (context.ConfigureAwait(false))
-        {
-            await context.Database.EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
-
-            var user = new User("quickconnect-user", "provider", "provider");
-            context.Users.Add(user);
-            await context.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
-
-            return user;
-        }
-    }
-
-    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 async Task<Replica> CreateReplicaAsync(NpgsqlDataSource dataSource, User user)
-    {
-        var connection = await _redis.ConnectAsync().ConfigureAwait(false);
-        _connections.Add(connection);
-
-        var userManager = new Mock<IUserManager>();
-        userManager.Setup(manager => manager.GetUserById(user.Id)).Returns(user);
-        var deviceManager = new DeviceManager(new DataSourceContextFactory(dataSource), userManager.Object);
-
-        var configManager = new Mock<IServerConfigurationManager>();
-        configManager.Setup(manager => manager.Configuration).Returns(new ServerConfiguration { QuickConnectAvailable = true });
-
-        // Stands in for SessionManager.AuthenticateDirect: the token has to be minted into the shared
-        // database, because the replica that exchanges the secret is not the one that authorized it.
-        var sessionManager = new Mock<ISessionManager>();
-        sessionManager
-            .Setup(manager => manager.AuthenticateDirect(It.IsAny<AuthenticationRequest>()))
-            .Returns<AuthenticationRequest>(async request =>
-            {
-                var device = await deviceManager.CreateDevice(
-                    new Device(request.UserId, request.App, request.AppVersion, request.DeviceName, request.DeviceId)).ConfigureAwait(false);
-
-                return new AuthenticationResult
-                {
-                    AccessToken = device.AccessToken,
-                    ServerId = "server-1",
-                    User = new UserDto { Id = user.Id, Name = user.Username, ServerId = "server-1" },
-                    SessionInfo = new SessionInfoDto
-                    {
-                        Id = device.Id.ToString(CultureInfo.InvariantCulture),
-                        UserId = user.Id,
-                        UserName = user.Username,
-                        Client = request.App,
-                        DeviceId = request.DeviceId,
-                        DeviceName = request.DeviceName,
-                        ApplicationVersion = request.AppVersion
-                    }
-                };
-            });
-
-        var store = new RedisQuickConnectStore(connection, NullLogger<RedisQuickConnectStore>.Instance);
-        var manager = new QuickConnectManager(
-            configManager.Object,
-            NullLogger<QuickConnectManager>.Instance,
-            sessionManager.Object,
-            store);
-
-        return new Replica(manager, store, deviceManager);
-    }
-
-    private sealed record Replica(IQuickConnect Manager, IQuickConnectStore Store, IDeviceManager Devices);
-
-    private sealed class DataSourceContextFactory : IDbContextFactory<JellyfinDbContext>
-    {
-        private readonly NpgsqlDataSource _dataSource;
-
-        public DataSourceContextFactory(NpgsqlDataSource dataSource)
-        {
-            _dataSource = dataSource;
-        }
-
-        public JellyfinDbContext CreateDbContext() => CreateContext(_dataSource);
-    }
-}

tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectStoreWiringTests.cs

@@ -1,142 +0,0 @@
-using System;
-using System.IO;
-using System.Threading.Tasks;
-using Emby.Server.Implementations.QuickConnect;
-using Jellyfin.Server.Extensions;
-using Jellyfin.Server.Tests.HighAvailability;
-using MediaBrowser.Common.Configuration;
-using MediaBrowser.Controller.Authentication;
-using MediaBrowser.Controller.QuickConnect;
-using MediaBrowser.Model.QuickConnect;
-using Microsoft.Extensions.Configuration;
-using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.Logging.Abstractions;
-using Moq;
-using StackExchange.Redis;
-using Xunit;
-
-namespace Jellyfin.Server.Tests.QuickConnect;
-
-/// <summary>
-/// Drives the whole configuration path a deployment uses: a bare
-/// <c>Jellyfin__TranscodeStore__RedisConnectionString</c> environment variable, the server's own
-/// configuration builder, the store registration, and a quick connect flow against a real valkey.
-/// </summary>
-[Trait("Category", "RequiresDocker")]
-[Collection("JellyfinSectionConfiguration")]
-public sealed class QuickConnectStoreWiringTests : IAsyncLifetime
-{
-    private const string RedisConnectionStringVariable = "Jellyfin__TranscodeStore__RedisConnectionString";
-
-    private RedisTestServer _redis = null!;
-    private string _configDirectory = string.Empty;
-
-    /// <inheritdoc/>
-    public async ValueTask InitializeAsync()
-    {
-        _redis = await RedisTestServer.StartAsync().ConfigureAwait(false);
-        _configDirectory = Directory.CreateTempSubdirectory("jellyfin-quickconnect-wiring").FullName;
-        await File.WriteAllTextAsync(Path.Combine(_configDirectory, "logging.default.json"), "{}").ConfigureAwait(false);
-    }
-
-    /// <inheritdoc/>
-    public async ValueTask DisposeAsync()
-    {
-        Environment.SetEnvironmentVariable(RedisConnectionStringVariable, null);
-
-        if (_configDirectory.Length > 0)
-        {
-            Directory.Delete(_configDirectory, true);
-        }
-
-        await _redis.DisposeAsync().ConfigureAwait(false);
-    }
-
-    /// <summary>
-    /// The variable form deployments set selects the shared store, and that store really talks to valkey.
-    /// </summary>
-    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
-    [Fact]
-    public async Task ManifestStyleEnvironmentVariable_SelectsTheSharedStore()
-    {
-        Environment.SetEnvironmentVariable(RedisConnectionStringVariable, _redis.ConnectionString + ",abortConnect=false");
-
-        await using var provider = BuildProvider();
-
-        var store = provider.GetRequiredService<IQuickConnectStore>();
-        Assert.IsType<RedisQuickConnectStore>(store);
-
-        var request = NewRequest();
-        await store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), TestContext.Current.CancellationToken);
-
-        Assert.True(await store.TryClaimAuthorizationAsync(request.Secret, DateTime.UtcNow.AddMinutes(10), TestContext.Current.CancellationToken));
-
-        var redis = provider.GetRequiredService<IConnectionMultiplexer>();
-        Assert.True(await redis.GetDatabase().KeyExistsAsync("jellyfin:quickconnect:request:" + request.Secret));
-    }
-
-    /// <summary>
-    /// Without the variable the deployment is single-instance and gets the process-local store.
-    /// </summary>
-    [Fact]
-    public void NoEnvironmentVariable_SelectsTheProcessLocalStore()
-    {
-        Environment.SetEnvironmentVariable(RedisConnectionStringVariable, null);
-
-        using var provider = BuildProvider();
-
-        Assert.IsType<InMemoryQuickConnectStore>(provider.GetRequiredService<IQuickConnectStore>());
-    }
-
-    /// <summary>
-    /// A configured but unreachable Redis degrades to the single-instance behaviour of a flow having to
-    /// complete against one instance, rather than taking quick connect down at startup.
-    /// </summary>
-    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
-    [Fact]
-    public async Task UnreachableRedisAtStartup_DegradesToTheProcessLocalStore()
-    {
-        Environment.SetEnvironmentVariable(RedisConnectionStringVariable, "127.0.0.1:1,connectTimeout=250,connectRetry=0");
-
-        await using var provider = BuildProvider();
-
-        var store = provider.GetRequiredService<IQuickConnectStore>();
-        Assert.IsType<InMemoryQuickConnectStore>(store);
-
-        // Quick connect still works, it just cannot span instances.
-        var request = NewRequest();
-        await store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), TestContext.Current.CancellationToken);
-        Assert.True(await store.TryClaimAuthorizationAsync(request.Secret, DateTime.UtcNow.AddMinutes(10), TestContext.Current.CancellationToken));
-        await store.SetAuthorizationAsync(
-            request.Secret,
-            new AuthenticationResult { AccessToken = "token-1" },
-            DateTime.UtcNow.AddMinutes(10),
-            TestContext.Current.CancellationToken);
-
-        Assert.Equal("token-1", (await store.TryConsumeAuthorizationAsync(request.Secret, TestContext.Current.CancellationToken))?.AccessToken);
-        Assert.Null(await store.TryConsumeAuthorizationAsync(request.Secret, TestContext.Current.CancellationToken));
-    }
-
-    private static QuickConnectResult NewRequest() => new QuickConnectResult(
-        Guid.NewGuid().ToString("N"),
-        Guid.NewGuid().ToString("N").Substring(0, 6),
-        DateTime.UtcNow,
-        "device-1",
-        "Living Room TV",
-        "Jellyfin Web",
-        "1.0.0");
-
-    private ServiceProvider BuildProvider()
-    {
-        var appPaths = new Mock<IApplicationPaths>();
-        appPaths.Setup(paths => paths.ConfigurationDirectoryPath).Returns(_configDirectory);
-        IConfiguration configuration = Jellyfin.Server.Program.CreateAppConfiguration(new StartupOptions(), appPaths.Object);
-
-        var services = new ServiceCollection();
-        services.AddLogging();
-        services.AddTranscodeSessionStore(configuration, NullLogger.Instance);
-        services.AddQuickConnectStore(configuration, NullLogger.Instance);
-
-        return services.BuildServiceProvider();
-    }
-}

tests/Jellyfin.Server.Tests/QuickConnect/RedisQuickConnectStoreDegradedTests.cs

@@ -1,308 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.Text.Json;
-using System.Threading;
-using System.Threading.Tasks;
-using Emby.Server.Implementations.QuickConnect;
-using Jellyfin.Server.Tests.HighAvailability;
-using MediaBrowser.Controller.Authentication;
-using MediaBrowser.Model.QuickConnect;
-using Microsoft.Extensions.Logging.Abstractions;
-using StackExchange.Redis;
-using Xunit;
-
-namespace Jellyfin.Server.Tests.QuickConnect;
-
-/// <summary>
-/// What a <see cref="RedisQuickConnectStore"/> does while its Redis is unreachable. Each instance talks
-/// to the one real server through a proxy of its own, so an outage can be given to one instance and not
-/// the others, and then taken back.
-/// </summary>
-[Trait("Category", "RequiresDocker")]
-public sealed class RedisQuickConnectStoreDegradedTests : IAsyncLifetime
-{
-    private readonly List<RedisFaultProxy> _proxies = new();
-    private readonly List<IConnectionMultiplexer> _connections = new();
-
-    private RedisTestServer _redis = null!;
-
-    private static CancellationToken CancellationToken => TestContext.Current.CancellationToken;
-
-    /// <inheritdoc/>
-    public async ValueTask InitializeAsync()
-    {
-        _redis = await RedisTestServer.StartAsync().ConfigureAwait(false);
-    }
-
-    /// <inheritdoc/>
-    public async ValueTask DisposeAsync()
-    {
-        foreach (var connection in _connections)
-        {
-            await connection.DisposeAsync().ConfigureAwait(false);
-        }
-
-        foreach (var proxy in _proxies)
-        {
-            await proxy.DisposeAsync().ConfigureAwait(false);
-        }
-
-        await _redis.DisposeAsync().ConfigureAwait(false);
-    }
-
-    /// <summary>
-    /// A request stored while Redis is unreachable is still resolvable on the instance that stored it,
-    /// so a flow whose three legs happen to land on one instance keeps working through the outage.
-    /// </summary>
-    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
-    [Fact]
-    public async Task PendingRequest_SurvivesAnOutage_OnTheInstanceThatStoredIt()
-    {
-        var instance = await CreateInstanceAsync();
-        var request = NewRequest();
-
-        instance.Proxy.Cut();
-        await instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken);
-
-        Assert.Equal(request.Secret, (await instance.Store.GetRequestBySecretAsync(request.Secret, CancellationToken))?.Secret);
-        Assert.Equal(request.Secret, (await instance.Store.GetRequestByCodeAsync(request.Code, CancellationToken))?.Secret);
-    }
-
-    /// <summary>
-    /// Once Redis answers again it is the only authority: a miss is a miss, not a reason to serve the
-    /// copy this instance kept while it was unreachable.
-    /// </summary>
-    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
-    [Fact]
-    public async Task PendingRequest_StoredDuringAnOutage_IsNotServedOnceRedisAnswersAgain()
-    {
-        var instance = await CreateInstanceAsync();
-        var request = NewRequest();
-
-        instance.Proxy.Cut();
-        await instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken);
-        Assert.NotNull(await instance.Store.GetRequestBySecretAsync(request.Secret, CancellationToken));
-
-        await RestoreAsync(instance);
-
-        Assert.Null(await instance.Store.GetRequestBySecretAsync(request.Secret, CancellationToken));
-        Assert.Null(await instance.Store.GetRequestByCodeAsync(request.Code, CancellationToken));
-    }
-
-    /// <summary>
-    /// A malformed stored value is a fault of its own, not a transport failure, so it is surfaced rather
-    /// than answered from the copy this instance happens to hold.
-    /// </summary>
-    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
-    [Fact]
-    public async Task PendingRequest_ThatIsMalformedInRedis_SurfacesInsteadOfDegrading()
-    {
-        var instance = await CreateInstanceAsync();
-        var request = NewRequest();
-
-        instance.Proxy.Cut();
-        await instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken);
-        await RestoreAsync(instance);
-
-        await instance.Connection.GetDatabase().StringSetAsync(
-            "jellyfin:quickconnect:request:" + request.Secret,
-            "{ not json",
-            TimeSpan.FromMinutes(10));
-
-        await Assert.ThrowsAsync<JsonException>(() => instance.Store.GetRequestBySecretAsync(request.Secret, CancellationToken));
-    }
-
-    /// <summary>
-    /// An authorization write that failed leaves nothing behind on the instance, because the response
-    /// that never arrived may still have been applied and a second copy of an authorization is a second
-    /// access token.
-    /// </summary>
-    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
-    [Fact]
-    public async Task Authorization_ThatFailedToStore_LeavesNothingOnTheInstance()
-    {
-        var instance = await CreateInstanceAsync();
-        var request = NewRequest();
-        await instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken);
-
-        instance.Proxy.Cut();
-        await AssertTransportFailureAsync(() => instance.Store.SetAuthorizationAsync(
-            request.Secret,
-            new AuthenticationResult { AccessToken = "token-1" },
-            DateTime.UtcNow.AddMinutes(10),
-            CancellationToken));
-
-        await RestoreAsync(instance);
-
-        Assert.Null(await instance.Store.TryConsumeAuthorizationAsync(request.Secret, CancellationToken));
-    }
-
-    /// <summary>
-    /// The instance whose authorization write failed while the write landed anyway still hands the token
-    /// out exactly once, rather than once from Redis and again from a copy of its own.
-    /// </summary>
-    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
-    [Fact]
-    public async Task Authorization_IsHandedOutOnce_EvenAfterAFailedWriteOnTheSameInstance()
-    {
-        var instance = await CreateInstanceAsync();
-        var request = NewRequest();
-        var authentication = new AuthenticationResult { AccessToken = "token-1" };
-        await instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken);
-
-        instance.Proxy.Cut();
-        await AssertTransportFailureAsync(() => instance.Store.SetAuthorizationAsync(
-            request.Secret,
-            authentication,
-            DateTime.UtcNow.AddMinutes(10),
-            CancellationToken));
-
-        await RestoreAsync(instance);
-
-        // Stands in for that write having been applied before the response was lost.
-        await instance.Store.SetAuthorizationAsync(request.Secret, authentication, DateTime.UtcNow.AddMinutes(10), CancellationToken);
-
-        Assert.Equal("token-1", (await instance.Store.TryConsumeAuthorizationAsync(request.Secret, CancellationToken))?.AccessToken);
-        Assert.Null(await instance.Store.TryConsumeAuthorizationAsync(request.Secret, CancellationToken));
-    }
-
-    /// <summary>
-    /// Exchanging during an outage fails loudly and spends nothing, so the token is still there to be
-    /// handed out once when Redis comes back.
-    /// </summary>
-    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
-    [Fact]
-    public async Task Exchange_DuringAnOutage_SurfacesTheFailureAndLeavesTheTokenUnspent()
-    {
-        var instance = await CreateInstanceAsync();
-        var request = NewRequest();
-        await instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken);
-        await instance.Store.SetAuthorizationAsync(
-            request.Secret,
-            new AuthenticationResult { AccessToken = "token-1" },
-            DateTime.UtcNow.AddMinutes(10),
-            CancellationToken);
-
-        instance.Proxy.Cut();
-        await AssertTransportFailureAsync(() => instance.Store.TryConsumeAuthorizationAsync(request.Secret, CancellationToken));
-
-        await RestoreAsync(instance);
-
-        Assert.Equal("token-1", (await instance.Store.TryConsumeAuthorizationAsync(request.Secret, CancellationToken))?.AccessToken);
-        Assert.Null(await instance.Store.TryConsumeAuthorizationAsync(request.Secret, CancellationToken));
-    }
-
-    /// <summary>
-    /// Authorizing during an outage fails loudly rather than claiming locally, because a claim only this
-    /// instance knows about does not stop another one minting a second access token.
-    /// </summary>
-    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
-    [Fact]
-    public async Task Claim_DuringAnOutage_SurfacesTheFailure()
-    {
-        var instance = await CreateInstanceAsync();
-        var request = NewRequest();
-        await instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken);
-
-        instance.Proxy.Cut();
-        await AssertTransportFailureAsync(() => instance.Store.TryClaimAuthorizationAsync(
-            request.Secret,
-            DateTime.UtcNow.AddMinutes(10),
-            CancellationToken));
-    }
-
-    /// <summary>
-    /// Two instances racing to authorize one request: exactly one of them may go on to mint a token.
-    /// </summary>
-    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
-    [Fact]
-    public async Task Claim_RacedOnTwoInstances_SucceedsOnce()
-    {
-        var first = await CreateInstanceAsync();
-        var second = await CreateInstanceAsync();
-
-        for (var attempt = 0; attempt < 25; attempt++)
-        {
-            var request = NewRequest();
-            var expiresUtc = DateTime.UtcNow.AddMinutes(10);
-            await first.Store.SetRequestAsync(request, expiresUtc, CancellationToken);
-
-            var claims = await Task.WhenAll(
-                Task.Run(() => first.Store.TryClaimAuthorizationAsync(request.Secret, expiresUtc, CancellationToken), CancellationToken),
-                Task.Run(() => second.Store.TryClaimAuthorizationAsync(request.Secret, expiresUtc, CancellationToken), CancellationToken));
-
-            Assert.Single(claims, claimed => claimed);
-        }
-    }
-
-    /// <summary>
-    /// A request that is unknown, already claimed or already authorized cannot be claimed.
-    /// </summary>
-    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
-    [Fact]
-    public async Task Claim_IsRefused_ForUnknownClaimedAndAuthorizedRequests()
-    {
-        var instance = await CreateInstanceAsync();
-        var expiresUtc = DateTime.UtcNow.AddMinutes(10);
-
-        Assert.False(await instance.Store.TryClaimAuthorizationAsync("unknown-secret", expiresUtc, CancellationToken));
-
-        var request = NewRequest();
-        await instance.Store.SetRequestAsync(request, expiresUtc, CancellationToken);
-        Assert.True(await instance.Store.TryClaimAuthorizationAsync(request.Secret, expiresUtc, CancellationToken));
-        Assert.False(await instance.Store.TryClaimAuthorizationAsync(request.Secret, expiresUtc, CancellationToken));
-
-        var authorized = NewRequest();
-        authorized.Authenticated = true;
-        await instance.Store.SetRequestAsync(authorized, expiresUtc, CancellationToken);
-        Assert.False(await instance.Store.TryClaimAuthorizationAsync(authorized.Secret, expiresUtc, CancellationToken));
-    }
-
-    private static QuickConnectResult NewRequest() => new QuickConnectResult(
-        Guid.NewGuid().ToString("N"),
-        Guid.NewGuid().ToString("N").Substring(0, 6),
-        DateTime.UtcNow,
-        "device-1",
-        "Living Room TV",
-        "Jellyfin Web",
-        "1.0.0");
-
-    private static async Task AssertTransportFailureAsync(Func<Task> operation)
-    {
-        var exception = await Record.ExceptionAsync(operation);
-
-        Assert.NotNull(exception);
-        Assert.True(exception is RedisException or TimeoutException, exception.ToString());
-    }
-
-    private static async Task RestoreAsync(Instance instance)
-    {
-        instance.Proxy.Restore();
-
-        for (var attempt = 1; ; attempt++)
-        {
-            try
-            {
-                await instance.Connection.GetDatabase().PingAsync();
-                return;
-            }
-            catch (Exception exception) when (exception is RedisException or TimeoutException && attempt < 60)
-            {
-                await Task.Delay(TimeSpan.FromMilliseconds(500), CancellationToken);
-            }
-        }
-    }
-
-    private async Task<Instance> CreateInstanceAsync()
-    {
-        var proxy = RedisFaultProxy.Start(_redis.ConnectionString);
-        _proxies.Add(proxy);
-
-        var connection = await ConnectionMultiplexer.ConnectAsync(proxy.ConnectionString).ConfigureAwait(false);
-        _connections.Add(connection);
-
-        return new Instance(proxy, connection, new RedisQuickConnectStore(connection, NullLogger<RedisQuickConnectStore>.Instance));
-    }
-
-    private sealed record Instance(RedisFaultProxy Proxy, IConnectionMultiplexer Connection, RedisQuickConnectStore Store);
-}