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

@@ -45,9 +45,9 @@ steps:
   # testcontainers, and a postgres service container deadlocks the step because the backend mounts
   # the ReadWriteOnce workspace volume into service pods and schedules them on another node.
   # Its data directory lives on the step's ephemeral storage, not on the workspace volume.
-  # Scoped to Jellyfin.Server.Tests: the three classes in Jellyfin.Database.Tests.PostgreSQL still
-  # start their own container, and PostgreSqlProviderTests already fails on main - on an EF 10
-  # scalar query and on its own data - which a third test in the class then inherits.
+  # 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, reached through JELLYFIN_TEST_REDIS.
   - name: postgres-migration-chain
     image: mcr.microsoft.com/dotnet/sdk:10.0
     depends_on:
@@ -56,15 +56,20 @@ steps:
       DOTNET_CLI_TELEMETRY_OPTOUT: "1"
       DOTNET_NOLOGO: "1"
       JELLYFIN_TEST_POSTGRES: "Host=127.0.0.1;Port=5432;Database=postgres;Username=postgres"
+      JELLYFIN_TEST_REDIS: "127.0.0.1:6379"
     commands:
       - apt-get -o Acquire::Retries=3 update || apt-get -o Acquire::Retries=3 update
-      - DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends postgresql
+      - DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends postgresql valkey-server
       - install -d -o postgres -g postgres /tmp/pgdata /tmp/pgrun
       - 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 ""
+      - 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"
+      - dotnet test tests/Jellyfin.Database.Tests.PostgreSQL/Jellyfin.Database.Tests.PostgreSQL.csproj -c Release --no-build --verbosity minimal --filter "Category=RequiresDocker"
     backend_options:
       kubernetes:
         serviceAccountName: jellyfin-ha-src

Emby.Server.Implementations/Library/LibraryManager.cs

@@ -2332,7 +2332,10 @@ namespace Emby.Server.Implementations.Library
         {
             IOrderedEnumerable<BaseItem>? orderedItems = null;
 
-            foreach (var orderBy in sortBy.Select(o => GetComparer(o, user)).Where(c => c is not null))
+            var comparers = sortBy.Select(o => GetComparer(o, user)).Where(c => c is not null).ToList();
+            items = PrefetchUserData(items, user, comparers);
+
+            foreach (var orderBy in comparers)
             {
                 if (orderBy is RandomComparer)
                 {
@@ -2364,14 +2367,14 @@ namespace Emby.Server.Implementations.Library
         {
             IOrderedEnumerable<BaseItem>? orderedItems = null;
 
-            foreach (var (name, sortOrder) in orderBy)
-            {
-                var comparer = GetComparer(name, user);
-                if (comparer is null)
-                {
-                    continue;
-                }
+            var comparers = orderBy
+                .Select(o => (Comparer: GetComparer(o.OrderBy, user), o.SortOrder))
+                .Where(c => c.Comparer is not null)
+                .ToList();
+            items = PrefetchUserData(items, user, comparers.Select(c => c.Comparer).ToList());
 
+            foreach (var (comparer, sortOrder) in comparers)
+            {
                 if (comparer is RandomComparer)
                 {
                     var randomItems = items.ToArray();
@@ -2397,6 +2400,31 @@ namespace Emby.Server.Implementations.Library
             return orderedItems ?? items;
         }
 
+        // The user comparers read user data per item, so without one batched read up front an
+        // in-memory sort would issue a database round trip per comparison.
+        private IEnumerable<BaseItem> PrefetchUserData(IEnumerable<BaseItem> items, User? user, IReadOnlyList<IBaseItemComparer?> comparers)
+        {
+            if (user is null)
+            {
+                return items;
+            }
+
+            var userComparers = comparers.OfType<IUserBaseItemComparer>().ToList();
+            if (userComparers.Count == 0)
+            {
+                return items;
+            }
+
+            var itemList = items as IReadOnlyList<BaseItem> ?? items.ToList();
+            var userData = _userDataManager.GetUserDataBatch(itemList, user);
+            foreach (var comparer in userComparers)
+            {
+                comparer.PrefetchedUserData = userData;
+            }
+
+            return itemList;
+        }
+
         /// <summary>
         /// Gets the comparer.
         /// </summary>

Emby.Server.Implementations/Library/UserDataManager.cs

@@ -2,10 +2,8 @@
 
 using System;
 using System.Collections.Generic;
-using System.Globalization;
 using System.Linq;
 using System.Threading;
-using BitFaster.Caching.Lru;
 using Jellyfin.Database.Implementations;
 using Jellyfin.Database.Implementations.Entities;
 using MediaBrowser.Controller.Configuration;
@@ -27,7 +25,6 @@ namespace Emby.Server.Implementations.Library
     {
         private readonly IServerConfigurationManager _config;
         private readonly IDbContextFactory<JellyfinDbContext> _repository;
-        private readonly FastConcurrentLru<string, UserItemData> _cache;
 
         /// <summary>
         /// Initializes a new instance of the <see cref="UserDataManager"/> class.
@@ -40,7 +37,6 @@ namespace Emby.Server.Implementations.Library
         {
             _config = config;
             _repository = repository;
-            _cache = new FastConcurrentLru<string, UserItemData>(Environment.ProcessorCount, _config.Configuration.CacheSize, StringComparer.OrdinalIgnoreCase);
         }
 
         /// <inheritdoc />
@@ -77,11 +73,6 @@ namespace Emby.Server.Implementations.Library
             dbContext.SaveChanges();
             transaction.Commit();
 
-            var userId = user.InternalId;
-            var cacheKey = GetCacheKey(userId, item.Id);
-            _cache.AddOrUpdate(cacheKey, userData);
-            item.UserData = dbContext.UserData.Where(e => e.ItemId == item.Id).AsNoTracking().ToArray(); // rehydrate the cached userdata
-
             UserDataSaved?.Invoke(this, new UserDataSaveEventArgs
             {
                 Keys = keys,
@@ -180,64 +171,41 @@ namespace Emby.Server.Implementations.Library
         /// <inheritdoc />
         public Dictionary<Guid, UserItemData> GetUserDataBatch(IReadOnlyList<BaseItem> items, User user)
         {
+            ArgumentNullException.ThrowIfNull(items);
+            ArgumentNullException.ThrowIfNull(user);
+
             var result = new Dictionary<Guid, UserItemData>(items.Count);
-            var itemsNeedingQuery = new List<(BaseItem Item, List<string> Keys)>();
-
-            foreach (var item in items)
-            {
-                var cacheKey = GetCacheKey(user.InternalId, item.Id);
-                if (_cache.TryGet(cacheKey, out var cachedData))
-                {
-                    result[item.Id] = cachedData;
-                }
-                else
-                {
-                    var userDataRow = ResolveUserDataRow(item, item.UserData?.Where(e => e.UserId.Equals(user.Id)));
-                    var userData = userDataRow is not null ? Map(userDataRow) : null;
-                    if (userData is not null)
-                    {
-                        result[item.Id] = userData;
-                        _cache.AddOrUpdate(cacheKey, userData);
-                    }
-                    else
-                    {
-                        var keys = item.GetUserDataKeys();
-                        itemsNeedingQuery.Add((item, keys));
-                    }
-                }
-            }
-
-            if (itemsNeedingQuery.Count == 0)
+            if (items.Count == 0)
             {
                 return result;
             }
 
-            // Build a single query for all missing items. Fetch rows by item alone so rows kept
-            // under keys from older metadata resolve the same way as the in-memory path.
-            var allItemIds = itemsNeedingQuery.Select(x => x.Item.Id).ToList();
+            // Fetch rows by item alone so rows kept under keys from older metadata resolve the same
+            // way as the single item path.
+            var itemIds = items.Select(e => e.Id).Distinct().ToList();
             using var context = _repository.CreateDbContext();
-            var userDataArray = context.UserData
+            var userDataByItem = context.UserData
                 .AsNoTracking()
                 .Where(e => e.UserId.Equals(user.Id))
-                .WhereOneOrMany(allItemIds, e => e.ItemId)
-                .ToArray();
+                .WhereOneOrMany(itemIds, e => e.ItemId)
+                .ToArray()
+                .GroupBy(e => e.ItemId)
+                .ToDictionary(g => g.Key, g => g.ToArray());
 
-            var userDataByItem = userDataArray.GroupBy(e => e.ItemId).ToDictionary(g => g.Key, g => g.ToArray());
-            foreach (var (item, keys) in itemsNeedingQuery)
+            foreach (var item in items)
             {
-                UserItemData userData;
-                if (userDataByItem.TryGetValue(item.Id, out var itemUserData) && itemUserData.Length > 0)
+                if (result.ContainsKey(item.Id))
                 {
-                    userData = Map(ResolveUserDataRow(item, itemUserData)!);
-                }
-                else
-                {
-                    userData = new UserItemData { Key = keys.Count > 0 ? keys[0] : string.Empty };
+                    continue;
                 }
 
-                result[item.Id] = userData;
-                var cacheKey = GetCacheKey(user.InternalId, item.Id);
-                _cache.AddOrUpdate(cacheKey, userData);
+                var row = userDataByItem.TryGetValue(item.Id, out var itemUserData)
+                    ? ResolveUserDataRow(item, itemUserData)
+                    : null;
+
+                result[item.Id] = row is not null
+                    ? Map(row)
+                    : new UserItemData { Key = item.GetUserDataKeys().FirstOrDefault() ?? string.Empty };
             }
 
             return result;
@@ -340,20 +308,19 @@ namespace Emby.Server.Implementations.Library
             return result;
         }
 
-        /// <summary>
-        /// Gets the internal key.
-        /// </summary>
-        /// <returns>System.String.</returns>
-        private static string GetCacheKey(long internalUserId, Guid itemId)
-        {
-            return internalUserId.ToString(CultureInfo.InvariantCulture) + "-" + itemId.ToString("N", CultureInfo.InvariantCulture);
-        }
-
         /// <inheritdoc />
         public UserItemData? GetUserData(User user, BaseItem item)
         {
             ArgumentNullException.ThrowIfNull(user);
-            var row = ResolveUserDataRow(item, item.UserData?.Where(e => e.UserId.Equals(user.Id)));
+            ArgumentNullException.ThrowIfNull(item);
+
+            using var dbContext = _repository.CreateDbContext();
+            var rows = dbContext.UserData
+                .AsNoTracking()
+                .Where(e => e.ItemId == item.Id && e.UserId == user.Id)
+                .ToArray();
+
+            var row = ResolveUserDataRow(item, rows);
             return row is not null ? Map(row) : new UserItemData()
             {
                 Key = item.GetUserDataKeys()[0],
@@ -536,16 +503,6 @@ namespace Emby.Server.Implementations.Library
             }
 
             dbContext.SaveChanges();
-
-            var cacheKey = GetCacheKey(user.InternalId, item.Id);
-            if (_cache.TryGet(cacheKey, out var cached))
-            {
-                cached.AudioStreamIndex = null;
-                cached.SubtitleStreamIndex = null;
-                _cache.AddOrUpdate(cacheKey, cached);
-            }
-
-            item.UserData = dbContext.UserData.Where(e => e.ItemId == item.Id).AsNoTracking().ToArray();
         }
     }
 }

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>

Emby.Server.Implementations/Sorting/DatePlayedComparer.cs

@@ -1,6 +1,7 @@
 #nullable disable
 
 using System;
+using System.Collections.Generic;
 using Jellyfin.Data.Enums;
 using Jellyfin.Database.Implementations.Entities;
 using MediaBrowser.Controller.Entities;
@@ -27,6 +28,12 @@ namespace Emby.Server.Implementations.Sorting
         /// <value>The user manager.</value>
         public IUserManager UserManager { get; set; }
 
+        /// <summary>
+        /// Gets or sets the prefetched user data.
+        /// </summary>
+        /// <value>The prefetched user data.</value>
+        public IReadOnlyDictionary<Guid, UserItemData> PrefetchedUserData { get; set; }
+
         /// <summary>
         /// Gets or sets the user data manager.
         /// </summary>
@@ -57,7 +64,7 @@ namespace Emby.Server.Implementations.Sorting
         /// <returns>DateTime.</returns>
         private DateTime GetDate(BaseItem x)
         {
-            var userdata = UserDataManager.GetUserData(User, x);
+            var userdata = this.GetUserData(x);
 
             if (userdata is not null && userdata.LastPlayedDate.HasValue)
             {

Emby.Server.Implementations/Sorting/IsFavoriteOrLikeComparer.cs

@@ -1,6 +1,8 @@
 #nullable disable
 #pragma warning disable CS1591
 
+using System;
+using System.Collections.Generic;
 using Jellyfin.Data.Enums;
 using Jellyfin.Database.Implementations.Entities;
 using MediaBrowser.Controller.Entities;
@@ -35,6 +37,12 @@ namespace Emby.Server.Implementations.Sorting
         /// <value>The user manager.</value>
         public IUserManager UserManager { get; set; }
 
+        /// <summary>
+        /// Gets or sets the prefetched user data.
+        /// </summary>
+        /// <value>The prefetched user data.</value>
+        public IReadOnlyDictionary<Guid, UserItemData> PrefetchedUserData { get; set; }
+
         /// <summary>
         /// Compares the specified x.
         /// </summary>
@@ -53,7 +61,7 @@ namespace Emby.Server.Implementations.Sorting
         /// <returns>DateTime.</returns>
         private int GetValue(BaseItem x)
         {
-            return x.IsFavoriteOrLiked(User, userItemData: null) ? 0 : 1;
+            return x.IsFavoriteOrLiked(User, this.GetUserData(x)) ? 0 : 1;
         }
     }
 }

Emby.Server.Implementations/Sorting/IsPlayedComparer.cs

@@ -2,6 +2,8 @@
 
 #pragma warning disable CS1591
 
+using System;
+using System.Collections.Generic;
 using Jellyfin.Data.Enums;
 using Jellyfin.Database.Implementations.Entities;
 using MediaBrowser.Controller.Entities;
@@ -36,6 +38,12 @@ namespace Emby.Server.Implementations.Sorting
         /// <value>The user manager.</value>
         public IUserManager UserManager { get; set; }
 
+        /// <summary>
+        /// Gets or sets the prefetched user data.
+        /// </summary>
+        /// <value>The prefetched user data.</value>
+        public IReadOnlyDictionary<Guid, UserItemData> PrefetchedUserData { get; set; }
+
         /// <summary>
         /// Compares the specified x.
         /// </summary>
@@ -54,7 +62,7 @@ namespace Emby.Server.Implementations.Sorting
         /// <returns>DateTime.</returns>
         private int GetValue(BaseItem x)
         {
-            return x.IsPlayed(User, userItemData: null) ? 0 : 1;
+            return x.IsPlayed(User, this.GetUserData(x)) ? 0 : 1;
         }
     }
 }

Emby.Server.Implementations/Sorting/IsUnplayedComparer.cs

@@ -2,6 +2,8 @@
 
 #pragma warning disable CS1591
 
+using System;
+using System.Collections.Generic;
 using Jellyfin.Data.Enums;
 using Jellyfin.Database.Implementations.Entities;
 using MediaBrowser.Controller.Entities;
@@ -36,6 +38,12 @@ namespace Emby.Server.Implementations.Sorting
         /// <value>The user manager.</value>
         public IUserManager UserManager { get; set; }
 
+        /// <summary>
+        /// Gets or sets the prefetched user data.
+        /// </summary>
+        /// <value>The prefetched user data.</value>
+        public IReadOnlyDictionary<Guid, UserItemData> PrefetchedUserData { get; set; }
+
         /// <summary>
         /// Compares the specified x.
         /// </summary>
@@ -54,7 +62,7 @@ namespace Emby.Server.Implementations.Sorting
         /// <returns>DateTime.</returns>
         private int GetValue(BaseItem x)
         {
-            return x.IsUnplayed(User, userItemData: null) ? 0 : 1;
+            return x.IsUnplayed(User, this.GetUserData(x)) ? 0 : 1;
         }
     }
 }

Emby.Server.Implementations/Sorting/PlayCountComparer.cs

@@ -1,5 +1,7 @@
 #nullable disable
 
+using System;
+using System.Collections.Generic;
 using Jellyfin.Data.Enums;
 using Jellyfin.Database.Implementations.Entities;
 using MediaBrowser.Controller.Entities;
@@ -38,6 +40,12 @@ namespace Emby.Server.Implementations.Sorting
         /// <value>The user manager.</value>
         public IUserManager UserManager { get; set; }
 
+        /// <summary>
+        /// Gets or sets the prefetched user data.
+        /// </summary>
+        /// <value>The prefetched user data.</value>
+        public IReadOnlyDictionary<Guid, UserItemData> PrefetchedUserData { get; set; }
+
         /// <summary>
         /// Compares the specified x.
         /// </summary>
@@ -56,7 +64,7 @@ namespace Emby.Server.Implementations.Sorting
         /// <returns>DateTime.</returns>
         private int GetValue(BaseItem x)
         {
-            var userdata = UserDataManager.GetUserData(User, x);
+            var userdata = this.GetUserData(x);
 
             return userdata is null ? 0 : userdata.PlayCount;
         }

Emby.Server.Implementations/TV/TVSeriesManager.cs

@@ -13,6 +13,7 @@ using MediaBrowser.Controller.Configuration;
 using MediaBrowser.Controller.Dto;
 using MediaBrowser.Controller.Entities;
 using MediaBrowser.Controller.Library;
+using MediaBrowser.Controller.Persistence;
 using MediaBrowser.Controller.TV;
 using MediaBrowser.Model.Querying;
 using Episode = MediaBrowser.Controller.Entities.TV.Episode;
@@ -124,53 +125,100 @@ namespace Emby.Server.Implementations.TV
 
             var batchResult = _libraryManager.GetNextUpEpisodesBatch(query, seriesKeys, includeSpecials, includeRewatching);
 
-            var nextUpList = new List<(DateTime LastWatchedDate, Episode Episode)>();
-
+            var results = new List<NextUpEpisodeBatchResult>(seriesKeys.Count);
             foreach (var seriesKey in seriesKeys)
             {
-                if (!batchResult.TryGetValue(seriesKey, out var result))
+                if (batchResult.TryGetValue(seriesKey, out var result))
                 {
-                    continue;
+                    results.Add(result);
                 }
+            }
 
-                var nextEpisode = DetermineNextEpisode(result, user, includeSpecials, request.EnableResumable, false);
+            // The selection below tests the played state of every episode it considers, so read the whole
+            // series batch in one query rather than one query per series.
+            var selectionCandidates = new List<BaseItem>();
+            foreach (var result in results)
+            {
+                AddCandidate(selectionCandidates, result.NextUp);
+                AddCandidate(selectionCandidates, result.LastWatched);
+                AddCandidate(selectionCandidates, result.NextPlayedForRewatching);
+                AddCandidate(selectionCandidates, result.LastWatchedForRewatching);
 
+                if (result.Specials is not null)
+                {
+                    selectionCandidates.AddRange(result.Specials);
+                }
+            }
+
+            var selectionUserData = _userDataManager.GetUserDataBatch(selectionCandidates, user);
+
+            var candidates = new List<NextUpCandidate>();
+            foreach (var result in results)
+            {
+                var nextEpisode = SelectNextEpisode(result, user, includeSpecials, includePlayed: false, selectionUserData);
                 if (nextEpisode is not null)
                 {
-                    // The last played date and the version that was actually played live on the version item's user data
-                    // The played state propagated to the sibling versions carries no date
-                    var (playedVersion, lastPlayedDate) = GetMostRecentlyPlayedVersion(result.LastWatched, user);
-                    nextEpisode = GetPreferredVersion(nextEpisode, result.LastWatched, playedVersion);
-
-                    DateTime lastWatchedDate = DateTime.MinValue;
-                    if (result.LastWatched is not null)
-                    {
-                        lastWatchedDate = lastPlayedDate ?? DateTime.MinValue.AddDays(1);
-                    }
-
-                    nextUpList.Add((lastWatchedDate, nextEpisode));
+                    candidates.Add(new NextUpCandidate(nextEpisode, result.LastWatched, !request.EnableResumable));
                 }
 
                 if (includeRewatching)
                 {
-                    var nextPlayedEpisode = DetermineNextEpisodeForRewatching(result, user, includeSpecials);
-
+                    var nextPlayedEpisode = SelectNextEpisode(result, user, includeSpecials, includePlayed: true, selectionUserData);
                     if (nextPlayedEpisode is not null)
                     {
-                        var (playedVersion, lastPlayedDate) = GetMostRecentlyPlayedVersion(result.LastWatchedForRewatching, user);
-                        nextPlayedEpisode = GetPreferredVersion(nextPlayedEpisode, result.LastWatchedForRewatching, playedVersion);
-
-                        DateTime rewatchLastWatchedDate = DateTime.MinValue;
-                        if (result.LastWatchedForRewatching is not null)
-                        {
-                            rewatchLastWatchedDate = lastPlayedDate ?? DateTime.MinValue.AddDays(1);
-                        }
-
-                        nextUpList.Add((rewatchLastWatchedDate, nextPlayedEpisode));
+                        // A rewatch suggestion is dropped once it has been resumed, whatever the request asked for.
+                        candidates.Add(new NextUpCandidate(nextPlayedEpisode, result.LastWatchedForRewatching, true));
                     }
                 }
             }
 
+            // The resume progress may live on an alternate version, so read every version in one query.
+            var episodeVersions = new List<BaseItem>();
+            foreach (var candidate in candidates)
+            {
+                if (candidate.DropWhenResumed)
+                {
+                    candidate.EpisodeVersions = candidate.Episode.GetAllVersions();
+                    episodeVersions.AddRange(candidate.EpisodeVersions);
+                }
+            }
+
+            if (episodeVersions.Count > 0)
+            {
+                var resumeUserData = _userDataManager.GetUserDataBatch(episodeVersions, user);
+                candidates.RemoveAll(candidate => candidate.EpisodeVersions
+                    .Any(version => GetUserData(user, version, resumeUserData)?.PlaybackPositionTicks > 0));
+            }
+
+            // The last played date and the version that was actually played live on the version item's user data
+            // The played state propagated to the sibling versions carries no date
+            var lastWatchedVersions = new List<BaseItem>();
+            foreach (var candidate in candidates)
+            {
+                if (candidate.LastWatched is Video lastWatchedVideo)
+                {
+                    candidate.LastWatchedVersions = lastWatchedVideo.GetAllVersions();
+                    lastWatchedVersions.AddRange(candidate.LastWatchedVersions);
+                }
+            }
+
+            var lastWatchedUserData = _userDataManager.GetUserDataBatch(lastWatchedVersions, user);
+
+            var nextUpList = new List<(DateTime LastWatchedDate, Episode Episode)>(candidates.Count);
+            foreach (var candidate in candidates)
+            {
+                var (playedVersion, lastPlayedDate) = GetMostRecentlyPlayedVersion(candidate.LastWatchedVersions, user, lastWatchedUserData);
+                var nextEpisode = GetPreferredVersion(candidate.Episode, candidate.LastWatched, playedVersion);
+
+                DateTime lastWatchedDate = DateTime.MinValue;
+                if (candidate.LastWatched is not null)
+                {
+                    lastWatchedDate = lastPlayedDate ?? DateTime.MinValue.AddDays(1);
+                }
+
+                nextUpList.Add((lastWatchedDate, nextEpisode));
+            }
+
             var sortedEpisodes = nextUpList
                 .OrderByDescending(x => x.LastWatchedDate)
                 .Select(x => (BaseItem)x.Episode);
@@ -178,12 +226,25 @@ namespace Emby.Server.Implementations.TV
             return GetResult(sortedEpisodes, request);
         }
 
-        private Episode? DetermineNextEpisode(
-            MediaBrowser.Controller.Persistence.NextUpEpisodeBatchResult result,
+        private static void AddCandidate(List<BaseItem> candidates, BaseItem? item)
+        {
+            if (item is not null)
+            {
+                candidates.Add(item);
+            }
+        }
+
+        private UserItemData? GetUserData(User user, BaseItem item, IReadOnlyDictionary<Guid, UserItemData> prefetchedUserData)
+            => prefetchedUserData.TryGetValue(item.Id, out var userData)
+                ? userData
+                : _userDataManager.GetUserData(user, item);
+
+        private Episode? SelectNextEpisode(
+            NextUpEpisodeBatchResult result,
             User user,
             bool includeSpecials,
-            bool includeResumable,
-            bool includePlayed)
+            bool includePlayed,
+            IReadOnlyDictionary<Guid, UserItemData> prefetchedUserData)
         {
             var nextEpisode = (includePlayed ? result.NextPlayedForRewatching : result.NextUp) as Episode;
             var lastWatchedEpisode = (includePlayed ? result.LastWatchedForRewatching : result.LastWatched) as Episode;
@@ -217,60 +278,41 @@ namespace Emby.Server.Implementations.TV
 
                     if (!includePlayed)
                     {
-                        sortedEpisodes = sortedEpisodes.Where(episode => _userDataManager.GetUserData(user, episode) is not { Played: true });
+                        sortedEpisodes = sortedEpisodes.Where(episode => GetUserData(user, episode, prefetchedUserData) is not { Played: true });
                     }
 
                     nextEpisode = sortedEpisodes.FirstOrDefault();
                 }
             }
 
-            if (nextEpisode is not null && !includeResumable)
-            {
-                // The resume progress may live on an alternate version
-                foreach (var version in nextEpisode.GetAllVersions())
-                {
-                    if (_userDataManager.GetUserData(user, version)?.PlaybackPositionTicks > 0)
-                    {
-                        return null;
-                    }
-                }
-            }
-
             return nextEpisode;
         }
 
-        private Episode? DetermineNextEpisodeForRewatching(
-            MediaBrowser.Controller.Persistence.NextUpEpisodeBatchResult result,
-            User user,
-            bool includeSpecials)
-        {
-            return DetermineNextEpisode(result, user, includeSpecials, includeResumable: false, includePlayed: true);
-        }
-
         /// <summary>
         /// Gets the version of the last watched episode that was actually played, together with its last played date.
         /// The version that was played carries the most recent LastPlayedDate.
         /// dates.
         /// </summary>
-        /// <param name="lastWatched">The last watched episode (any version).</param>
+        /// <param name="versions">The versions of the last watched episode.</param>
         /// <param name="user">The user.</param>
+        /// <param name="prefetchedUserData">User data read for every version up front.</param>
         /// <returns>The played version and its last played date.</returns>
-        private (Video? PlayedVersion, DateTime? LastPlayedDate) GetMostRecentlyPlayedVersion(BaseItem? lastWatched, User user)
+        private (Video? PlayedVersion, DateTime? LastPlayedDate) GetMostRecentlyPlayedVersion(
+            IReadOnlyList<Video> versions,
+            User user,
+            IReadOnlyDictionary<Guid, UserItemData> prefetchedUserData)
         {
-            if (lastWatched is not Video lastWatchedVideo)
+            if (versions.Count == 0)
             {
                 return (null, null);
             }
 
-            var versions = lastWatchedVideo.GetAllVersions();
-            var userDataByVersion = _userDataManager.GetUserDataBatch(versions, user);
-
             var playedVersion = VersionPlaybackSelector.SelectMostRecentlyPlayed(
                 versions,
-                version => userDataByVersion.GetValueOrDefault(version.Id),
+                version => GetUserData(user, version, prefetchedUserData),
                 data => data.LastPlayedDate.HasValue);
 
-            return (playedVersion, playedVersion is null ? null : userDataByVersion[playedVersion.Id].LastPlayedDate);
+            return (playedVersion, playedVersion is null ? null : GetUserData(user, playedVersion, prefetchedUserData)?.LastPlayedDate);
         }
 
         /// <summary>
@@ -346,5 +388,28 @@ namespace Emby.Server.Implementations.TV
                 totalCount,
                 items.ToArray());
         }
+
+        /// <summary>
+        /// An episode picked for Next Up, together with the versions its user data is read from.
+        /// </summary>
+        private sealed class NextUpCandidate
+        {
+            public NextUpCandidate(Episode episode, BaseItem? lastWatched, bool dropWhenResumed)
+            {
+                Episode = episode;
+                LastWatched = lastWatched;
+                DropWhenResumed = dropWhenResumed;
+            }
+
+            public Episode Episode { get; }
+
+            public BaseItem? LastWatched { get; }
+
+            public bool DropWhenResumed { get; }
+
+            public IReadOnlyList<Video> EpisodeVersions { get; set; } = [];
+
+            public IReadOnlyList<Video> LastWatchedVersions { get; set; } = [];
+        }
     }
 }

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

@@ -108,7 +108,7 @@ public class TvShowsController : BaseJellyfinApiController
                 StartIndex = startIndex,
                 User = user,
                 EnableTotalRecordCount = enableTotalRecordCount,
-                NextUpDateCutoff = nextUpDateCutoff ?? DateTime.MinValue,
+                NextUpDateCutoff = nextUpDateCutoff?.ToUniversalTime() ?? DateTime.SpecifyKind(DateTime.MinValue, DateTimeKind.Utc),
                 EnableResumable = enableResumable,
                 EnableRewatching = enableRewatching
             },

Jellyfin.Server/CoreAppHost.cs

@@ -116,6 +116,10 @@ namespace Jellyfin.Server
             // to the other instances. Redis-backed when configured, no-op otherwise.
             serviceCollection.AddConfigurationInvalidationBus(_startupConfig, Logger);
 
+            // Session directory: publishes which instance holds which session and routes remote-control
+            // messages to it. Redis-backed when configured, no-op otherwise.
+            serviceCollection.AddSessionDirectory(_startupConfig, Logger);
+
             foreach (var type in GetExportTypes<ILyricProvider>())
             {
                 serviceCollection.AddSingleton(typeof(ILyricProvider), type);

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/Entities/UserViewBuilder.cs

@@ -449,19 +449,26 @@ namespace MediaBrowser.Controller.Entities
             IUserDataManager userDataManager,
             ILibraryManager libraryManager)
         {
-            var filtered = items.Where(i => Filter(i, user, query, userDataManager, libraryManager));
+            var itemList = items as IReadOnlyList<BaseItem> ?? items.ToList();
+
+            // The user data checks below run per item, so read them all in one query up front.
+            var userDataBatch = user is not null && RequiresUserData(query)
+                ? userDataManager.GetUserDataBatch(itemList, user)
+                : null;
+
+            var filtered = itemList.Where(i => Filter(i, user, query, userDataManager, libraryManager, userDataBatch));
 
             if (query.IsPlayed.HasValue && user is not null)
             {
-                var itemList = filtered.ToList();
-                var folderIds = itemList.OfType<Folder>().Select(f => f.Id).ToList();
+                var filteredList = filtered.ToList();
+                var folderIds = filteredList.OfType<Folder>().Select(f => f.Id).ToList();
 
                 if (folderIds.Count > 0)
                 {
                     var counts = libraryManager.GetPlayedAndTotalCountBatch(folderIds, user);
                     var isPlayedValue = query.IsPlayed.Value;
 
-                    return itemList.Where(item =>
+                    return filteredList.Where(item =>
                     {
                         if (item is Folder)
                         {
@@ -473,7 +480,7 @@ namespace MediaBrowser.Controller.Entities
                     });
                 }
 
-                return itemList;
+                return filteredList;
             }
 
             return filtered;
@@ -515,12 +522,29 @@ namespace MediaBrowser.Controller.Entities
                 itemsArray);
         }
 
+        private static bool RequiresUserData(InternalItemsQuery query)
+            => query.IsLiked.HasValue
+                || query.IsFavoriteOrLiked.HasValue
+                || query.IsFavorite.HasValue
+                || query.IsResumable.HasValue
+                || query.IsPlayed.HasValue;
+
+        private static UserItemData GetUserData(
+            IUserDataManager userDataManager,
+            User user,
+            BaseItem item,
+            Dictionary<Guid, UserItemData> userDataBatch)
+            => userDataBatch is not null && userDataBatch.TryGetValue(item.Id, out var userData)
+                ? userData
+                : userDataManager.GetUserData(user, item);
+
         private static bool Filter(
             BaseItem item,
             User user,
             InternalItemsQuery query,
             IUserDataManager userDataManager,
-            ILibraryManager libraryManager)
+            ILibraryManager libraryManager,
+            Dictionary<Guid, UserItemData> userDataBatch)
         {
             if (!string.IsNullOrEmpty(query.NameStartsWith) && !item.SortName.StartsWith(query.NameStartsWith, StringComparison.InvariantCultureIgnoreCase))
             {
@@ -568,7 +592,7 @@ namespace MediaBrowser.Controller.Entities
 
             if (query.IsLiked.HasValue)
             {
-                userData = userDataManager.GetUserData(user, item);
+                userData = GetUserData(userDataManager, user, item, userDataBatch);
                 if (!userData.Likes.HasValue || userData.Likes != query.IsLiked.Value)
                 {
                     return false;
@@ -577,7 +601,7 @@ namespace MediaBrowser.Controller.Entities
 
             if (query.IsFavoriteOrLiked.HasValue)
             {
-                userData ??= userDataManager.GetUserData(user, item);
+                userData ??= GetUserData(userDataManager, user, item, userDataBatch);
                 var isFavoriteOrLiked = userData.IsFavorite || (userData.Likes ?? false);
 
                 if (isFavoriteOrLiked != query.IsFavoriteOrLiked.Value)
@@ -588,7 +612,7 @@ namespace MediaBrowser.Controller.Entities
 
             if (query.IsFavorite.HasValue)
             {
-                userData ??= userDataManager.GetUserData(user, item);
+                userData ??= GetUserData(userDataManager, user, item, userDataBatch);
                 if (userData.IsFavorite != query.IsFavorite.Value)
                 {
                     return false;
@@ -597,7 +621,7 @@ namespace MediaBrowser.Controller.Entities
 
             if (query.IsResumable.HasValue)
             {
-                userData ??= userDataManager.GetUserData(user, item);
+                userData ??= GetUserData(userDataManager, user, item, userDataBatch);
                 var isResumable = userData.PlaybackPositionTicks > 0;
 
                 if (isResumable != query.IsResumable.Value)
@@ -612,7 +636,7 @@ namespace MediaBrowser.Controller.Entities
                 // Folders are batch-filtered by the collection Filter() overload.
                 if (!item.IsFolder)
                 {
-                    userData ??= userDataManager.GetUserData(user, item);
+                    userData ??= GetUserData(userDataManager, user, item, userDataBatch);
                     if (item.IsPlayed(user, userData) != query.IsPlayed.Value)
                     {
                         return false;

MediaBrowser.Controller/ScheduledTasks/ScanLeaderOptions.cs

@@ -43,6 +43,14 @@ public sealed class ScanLeaderOptions
         "TaskExtractMediaSegments",
         "KeyframeExtraction",
         "CleanupUserDataTask",
-        "OptimizeDatabaseTask"
+        "OptimizeDatabaseTask",
+        "DownloadLyrics",
+        "DownloadSubtitles",
+        "TmdbRefreshUpcomingEpisodes",
+        "RefreshTrickplayImages",
+        "MoveTrickplayImages",
+        "RefreshInternetChannels",
+        "RefreshGuide",
+        "PluginUpdates"
     };
 }

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;
+}

MediaBrowser.Controller/Sorting/IUserBaseItemComparer.cs

@@ -1,6 +1,9 @@
 #nullable disable
 
+using System;
+using System.Collections.Generic;
 using Jellyfin.Database.Implementations.Entities;
+using MediaBrowser.Controller.Entities;
 using MediaBrowser.Controller.Library;
 
 namespace MediaBrowser.Controller.Sorting
@@ -27,5 +30,16 @@ namespace MediaBrowser.Controller.Sorting
         /// </summary>
         /// <value>The user data repository.</value>
         IUserDataManager UserDataManager { get; set; }
+
+        /// <summary>
+        /// Gets or sets user data for the items being sorted, keyed by item id, read once up front.
+        /// A comparer that does not store it reads its user data one item at a time instead.
+        /// </summary>
+        /// <value>The prefetched user data, or <c>null</c> when none was prefetched.</value>
+        IReadOnlyDictionary<Guid, UserItemData> PrefetchedUserData
+        {
+            get => null;
+            set { }
+        }
     }
 }

MediaBrowser.Controller/Sorting/UserBaseItemComparerExtensions.cs

@@ -0,0 +1,28 @@
+#nullable disable
+
+using MediaBrowser.Controller.Entities;
+
+namespace MediaBrowser.Controller.Sorting
+{
+    /// <summary>
+    /// Helpers shared by the comparers that sort on user data.
+    /// </summary>
+    public static class UserBaseItemComparerExtensions
+    {
+        /// <summary>
+        /// Gets the user data for an item, preferring the batch the sort prefetched.
+        /// </summary>
+        /// <param name="comparer">The comparer.</param>
+        /// <param name="item">The item.</param>
+        /// <returns>The item's user data.</returns>
+        public static UserItemData GetUserData(this IUserBaseItemComparer comparer, BaseItem item)
+        {
+            if (comparer.PrefetchedUserData is not null && comparer.PrefetchedUserData.TryGetValue(item.Id, out var userData))
+            {
+                return userData;
+            }
+
+            return comparer.UserDataManager.GetUserData(comparer.User, item);
+        }
+    }
+}

MediaBrowser.Model/Querying/NextUpQuery.cs

@@ -12,7 +12,7 @@ public class NextUpQuery
     {
         EnableImageTypes = Array.Empty<ImageType>();
         EnableTotalRecordCount = true;
-        NextUpDateCutoff = DateTime.MinValue;
+        NextUpDateCutoff = DateTime.SpecifyKind(DateTime.MinValue, DateTimeKind.Utc);
         EnableResumable = false;
         EnableRewatching = false;
     }
@@ -56,7 +56,7 @@ public class NextUpQuery
     public bool EnableTotalRecordCount { get; set; }
 
     /// <summary>
-    /// Gets or sets a value indicating the oldest date for a show to appear in Next Up.
+    /// Gets or sets a value indicating the oldest date, in UTC, for a show to appear in Next Up.
     /// </summary>
     public DateTime NextUpDateCutoff { get; set; }
 

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
 

src/Jellyfin.LiveTv/Channels/ChannelManager.cs

@@ -212,7 +212,8 @@ namespace Jellyfin.LiveTv.Channels
             if (query.IsFavorite.HasValue)
             {
                 var val = query.IsFavorite.Value;
-                channels = channels.Where(i => _userDataManager.GetUserData(user, i).IsFavorite == val)
+                var userData = _userDataManager.GetUserDataBatch(channels, user);
+                channels = channels.Where(i => userData.TryGetValue(i.Id, out var data) && data.IsFavorite == val)
                     .ToList();
             }
 

src/Jellyfin.LiveTv/LiveTvManager.cs

@@ -304,8 +304,17 @@ namespace Jellyfin.LiveTv
 
             if (query.IsAiring ?? false)
             {
+                // Scoring reads the channel's user data per program, so read every channel's in one query.
+                var channels = programList
+                    .Cast<LiveTvProgram>()
+                    .Select(i => _libraryManager.GetItemById(i.ChannelId))
+                    .OfType<BaseItem>()
+                    .DistinctBy(i => i.Id)
+                    .ToList();
+                var channelUserData = _userDataManager.GetUserDataBatch(channels, user);
+
                 orderedPrograms = orderedPrograms
-                    .ThenByDescending(i => GetRecommendationScore(i, user, true));
+                    .ThenByDescending(i => GetRecommendationScore(i, user, true, channelUserData));
             }
 
             IEnumerable<BaseItem> programs = orderedPrograms;
@@ -338,7 +347,11 @@ namespace Jellyfin.LiveTv
                 _dtoService.GetBaseItemDtos(internalResult.Items, options, query.User)));
         }
 
-        private int GetRecommendationScore(LiveTvProgram program, User user, bool factorChannelWatchCount)
+        private int GetRecommendationScore(
+            LiveTvProgram program,
+            User user,
+            bool factorChannelWatchCount,
+            IReadOnlyDictionary<Guid, UserItemData> channelUserData)
         {
             var score = 0;
 
@@ -359,7 +372,9 @@ namespace Jellyfin.LiveTv
                 return score;
             }
 
-            var channelUserdata = _userDataManager.GetUserData(user, channel);
+            var channelUserdata = channelUserData.TryGetValue(channel.Id, out var cached)
+                ? cached
+                : _userDataManager.GetUserData(user, channel);
 
             if (channelUserdata.Likes.HasValue)
             {

tests/Jellyfin.Database.Tests.PostgreSQL/Jellyfin.Database.Tests.PostgreSQL.csproj

@@ -18,6 +18,11 @@
     <PackageReference Include="coverlet.collector" />
   </ItemGroup>
 
+  <ItemGroup>
+    <!-- Linked, not project-referenced: Jellyfin.Server.Tests drags the whole server into this output. -->
+    <Compile Include="..\Jellyfin.Server.Tests\Migrations\PostgreSqlTestServer.cs" Link="Migrations\PostgreSqlTestServer.cs" />
+  </ItemGroup>
+
   <ItemGroup>
     <ProjectReference Include="..\..\src\Jellyfin.Database\Jellyfin.Database.Providers.PostgreSQL\Jellyfin.Database.Providers.PostgreSQL.csproj" />
     <ProjectReference Include="..\..\src\Jellyfin.Database\Jellyfin.Database.Implementations\Jellyfin.Database.Implementations.csproj" />

tests/Jellyfin.Database.Tests.PostgreSQL/PostgreSqlConcurrencyTests.cs

@@ -1,71 +1,68 @@
 using System;
 using System.Collections.Generic;
+using System.Threading;
 using System.Threading.Tasks;
-using DotNet.Testcontainers.Builders;
 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.Tests.Migrations;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.Logging.Abstractions;
 using Npgsql;
-using Testcontainers.PostgreSql;
 using Xunit;
 
 namespace Jellyfin.Database.Tests.PostgreSQL;
 
 /// <summary>
-/// Integration tests that verify concurrent access patterns against a real PostgreSQL 16 container.
+/// Integration tests that verify concurrent access patterns against a real PostgreSQL server.
 /// </summary>
 [Xunit.Trait("Category", "RequiresDocker")]
 public sealed class PostgreSqlConcurrencyTests : IAsyncLifetime
 {
-    private readonly PostgreSqlContainer _container;
+    private static int _databaseSequence;
+
+    private PostgreSqlTestServer? _server;
     private NpgsqlDataSource? _dataSource;
     private PostgreSqlDatabaseProvider? _provider;
 
     /// <summary>
-    /// Initializes a new instance of the <see cref="PostgreSqlConcurrencyTests"/> class.
-    /// </summary>
-    public PostgreSqlConcurrencyTests()
-    {
-        _container = new PostgreSqlBuilder("postgres:16-alpine")
-            .WithWaitStrategy(Wait.ForUnixContainer().UntilCommandIsCompleted("pg_isready"))
-            .Build();
-    }
-
-    /// <summary>
-    /// Starts the PostgreSQL container and applies migrations before any tests in the class run.
+    /// Attaches to the test server, hands this test a database of its own and applies migrations to it.
     /// </summary>
     /// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
     public async ValueTask InitializeAsync()
     {
-        await _container.StartAsync().ConfigureAwait(false);
+        _server = await PostgreSqlTestServer.StartAsync().ConfigureAwait(false);
 
-        _dataSource = new NpgsqlDataSourceBuilder(_container.GetConnectionString()).Build();
+        var databaseName = FormattableString.Invariant($"pg_concurrency_{Interlocked.Increment(ref _databaseSequence)}");
+        var connectionString = await _server.CreateDatabaseAsync(databaseName, TestContext.Current.CancellationToken).ConfigureAwait(false);
+        _dataSource = new NpgsqlDataSourceBuilder(connectionString).Build();
         _provider = new PostgreSqlDatabaseProvider(_dataSource);
 
-        // Apply migrations once for the whole test class.
         var context = CreateContext();
         await using (context.ConfigureAwait(false))
         {
-            await context.Database.MigrateAsync().ConfigureAwait(false);
+            await context.Database.MigrateAsync(TestContext.Current.CancellationToken).ConfigureAwait(false);
         }
     }
 
     /// <summary>
-    /// Stops and removes the PostgreSQL container after all tests in the class have run.
+    /// Releases the data source and the test server.
     /// </summary>
     /// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
     public async ValueTask DisposeAsync()
     {
+        // InitializeAsync can fail before the data source exists; its error must not be masked by an NRE here.
         if (_dataSource is not null)
         {
             await _dataSource.DisposeAsync().ConfigureAwait(false);
         }
 
-        await _container.DisposeAsync().ConfigureAwait(false);
+        if (_server is not null)
+        {
+            await _server.DisposeAsync().ConfigureAwait(false);
+        }
     }
 
     /// <summary>

tests/Jellyfin.Database.Tests.PostgreSQL/PostgreSqlMigrationTests.cs

@@ -1,62 +1,68 @@
+using System;
+using System.Threading;
 using System.Threading.Tasks;
-using DotNet.Testcontainers.Builders;
 using Jellyfin.Database.Implementations;
 using Jellyfin.Database.Implementations.DbConfiguration;
 using Jellyfin.Database.Implementations.Locking;
 using Jellyfin.Database.Providers.PostgreSQL;
+using Jellyfin.Server.Tests.Migrations;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.Logging.Abstractions;
 using Npgsql;
-using Testcontainers.PostgreSql;
 using Xunit;
 
 namespace Jellyfin.Database.Tests.PostgreSQL;
 
 /// <summary>
-/// Integration tests that validate PostgreSQL migrations against a real container.
+/// Integration tests that validate PostgreSQL migrations against a real server.
 /// </summary>
 [Xunit.Trait("Category", "RequiresDocker")]
 public sealed class PostgreSqlMigrationTests : IAsyncLifetime
 {
-    private readonly PostgreSqlContainer _container;
+    private static int _databaseSequence;
+
+    private PostgreSqlTestServer? _server;
+    private NpgsqlDataSource? _dataSource;
 
     /// <summary>
-    /// Initializes a new instance of the <see cref="PostgreSqlMigrationTests"/> class.
-    /// </summary>
-    public PostgreSqlMigrationTests()
-    {
-        _container = new PostgreSqlBuilder("postgres:16-alpine")
-            .WithWaitStrategy(Wait.ForUnixContainer().UntilCommandIsCompleted("pg_isready"))
-            .Build();
-    }
-
-    /// <summary>
-    /// Starts the PostgreSQL container before any tests in the class run.
+    /// Attaches to the test server and hands this test an empty database of its own.
     /// </summary>
     /// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
     public async ValueTask InitializeAsync()
     {
-        await _container.StartAsync().ConfigureAwait(false);
+        _server = await PostgreSqlTestServer.StartAsync().ConfigureAwait(false);
+
+        var databaseName = FormattableString.Invariant($"pg_migration_{Interlocked.Increment(ref _databaseSequence)}");
+        var connectionString = await _server.CreateDatabaseAsync(databaseName, TestContext.Current.CancellationToken).ConfigureAwait(false);
+        _dataSource = new NpgsqlDataSourceBuilder(connectionString).Build();
     }
 
     /// <summary>
-    /// Stops and removes the PostgreSQL container after all tests in the class have run.
+    /// Releases the data source and the test server.
     /// </summary>
     /// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
     public async ValueTask DisposeAsync()
     {
-        await _container.DisposeAsync().ConfigureAwait(false);
+        // InitializeAsync can fail before the data source exists; its error must not be masked by an NRE here.
+        if (_dataSource is not null)
+        {
+            await _dataSource.DisposeAsync().ConfigureAwait(false);
+        }
+
+        if (_server is not null)
+        {
+            await _server.DisposeAsync().ConfigureAwait(false);
+        }
     }
 
     /// <summary>
-    /// Verifies that the <c>InitialPostgreSql</c> migration applies cleanly to a fresh PostgreSQL 16 container.
+    /// Verifies that the <c>InitialPostgreSql</c> migration applies cleanly to a fresh database.
     /// </summary>
     /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
     [Fact]
     public async Task MigrateAsync_AppliesInitialMigrationCleanly()
     {
-        await using var dataSource = new NpgsqlDataSourceBuilder(_container.GetConnectionString()).Build();
-        var context = CreateContext(dataSource);
+        var context = CreateContext(_dataSource!);
         await using (context)
         {
             await context.Database.MigrateAsync(TestContext.Current.CancellationToken);
@@ -73,11 +79,7 @@ public sealed class PostgreSqlMigrationTests : IAsyncLifetime
     [Fact]
     public void CheckForUnappliedMigrations_PostgreSql()
     {
-        // Use a dummy connection string; HasPendingModelChanges() is a purely in-memory check
-        // that compares the current compiled model with the migration snapshots — no real DB needed.
-        const string dummyConnectionString = "Host=localhost;Database=jellyfin;Username=postgres;Password=postgres";
-        using var dataSource = new NpgsqlDataSourceBuilder(dummyConnectionString).Build();
-        using var context = CreateContext(dataSource);
+        using var context = CreateContext(_dataSource!);
 
         Assert.False(
             context.Database.HasPendingModelChanges(),

tests/Jellyfin.Database.Tests.PostgreSQL/PostgreSqlProviderTests.cs

@@ -2,71 +2,67 @@ using System;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
-using DotNet.Testcontainers.Builders;
 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.Tests.Migrations;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.Logging.Abstractions;
 using Npgsql;
-using Testcontainers.PostgreSql;
 using Xunit;
 
 namespace Jellyfin.Database.Tests.PostgreSQL;
 
 /// <summary>
-/// Integration tests for CRUD operations, optimisation, and purge against a real PostgreSQL 16 container.
+/// Integration tests for CRUD operations, optimisation, and purge against a real PostgreSQL server.
 /// </summary>
 [Xunit.Trait("Category", "RequiresDocker")]
 public sealed class PostgreSqlProviderTests : IAsyncLifetime
 {
-    private readonly PostgreSqlContainer _container;
+    private static int _databaseSequence;
+
+    private PostgreSqlTestServer? _server;
     private NpgsqlDataSource? _dataSource;
     private PostgreSqlDatabaseProvider? _provider;
 
     /// <summary>
-    /// Initializes a new instance of the <see cref="PostgreSqlProviderTests"/> class.
-    /// </summary>
-    public PostgreSqlProviderTests()
-    {
-        _container = new PostgreSqlBuilder("postgres:16-alpine")
-            .WithWaitStrategy(Wait.ForUnixContainer().UntilCommandIsCompleted("pg_isready"))
-            .Build();
-    }
-
-    /// <summary>
-    /// Starts the PostgreSQL container and applies migrations before any tests in the class run.
+    /// Attaches to the test server, hands this test a database of its own and applies migrations to it.
     /// </summary>
     /// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
     public async ValueTask InitializeAsync()
     {
-        await _container.StartAsync().ConfigureAwait(false);
+        _server = await PostgreSqlTestServer.StartAsync().ConfigureAwait(false);
 
-        _dataSource = new NpgsqlDataSourceBuilder(_container.GetConnectionString()).Build();
+        var databaseName = FormattableString.Invariant($"pg_provider_{Interlocked.Increment(ref _databaseSequence)}");
+        var connectionString = await _server.CreateDatabaseAsync(databaseName, TestContext.Current.CancellationToken).ConfigureAwait(false);
+        _dataSource = new NpgsqlDataSourceBuilder(connectionString).Build();
         _provider = new PostgreSqlDatabaseProvider(_dataSource);
 
-        // Apply migrations once for the whole test class.
         var context = CreateContext();
         await using (context.ConfigureAwait(false))
         {
-            await context.Database.MigrateAsync().ConfigureAwait(false);
+            await context.Database.MigrateAsync(TestContext.Current.CancellationToken).ConfigureAwait(false);
         }
     }
 
     /// <summary>
-    /// Stops and removes the PostgreSQL container after all tests in the class have run.
+    /// Releases the data source and the test server.
     /// </summary>
     /// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
     public async ValueTask DisposeAsync()
     {
+        // InitializeAsync can fail before the data source exists; its error must not be masked by an NRE here.
         if (_dataSource is not null)
         {
             await _dataSource.DisposeAsync().ConfigureAwait(false);
         }
 
-        await _container.DisposeAsync().ConfigureAwait(false);
+        if (_server is not null)
+        {
+            await _server.DisposeAsync().ConfigureAwait(false);
+        }
     }
 
     /// <summary>
@@ -155,11 +151,15 @@ public sealed class PostgreSqlProviderTests : IAsyncLifetime
         var ctx = CreateContext();
         await using (ctx)
         {
-            var userId = Guid.NewGuid();
+            // DisplayPreferences.UserId is a foreign key onto Users, which PostgreSQL enforces and SQLite does not.
+            var user = new User("prefsuser", "Jellyfin.Server.Implementations.Users.DefaultAuthenticationProvider", "Jellyfin.Server.Implementations.Users.DefaultPasswordResetProvider");
+            ctx.Users.Add(user);
+            await ctx.SaveChangesAsync(TestContext.Current.CancellationToken);
+
             var itemId = Guid.NewGuid();
 
             // Create
-            var prefs = new DisplayPreferences(userId, itemId, "TestClient");
+            var prefs = new DisplayPreferences(user.Id, itemId, "TestClient");
             ctx.DisplayPreferences.Add(prefs);
             await ctx.SaveChangesAsync(TestContext.Current.CancellationToken);
 
@@ -278,7 +278,7 @@ public sealed class PostgreSqlProviderTests : IAsyncLifetime
 
             // session_replication_role should be reset to 'origin' (default)
             var role = await ctx.Database
-                .SqlQueryRaw<string>("SELECT current_setting('session_replication_role')")
+                .SqlQueryRaw<string>("SELECT current_setting('session_replication_role') AS \"Value\"")
                 .FirstAsync(TestContext.Current.CancellationToken);
             Assert.Equal("origin", role);
         }

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

@@ -31,6 +31,8 @@
   <ItemGroup>
     <ProjectReference Include="..\..\Emby.Server.Implementations\Emby.Server.Implementations.csproj" />
     <ProjectReference Include="..\..\Jellyfin.Server.Implementations\Jellyfin.Server.Implementations.csproj" />
+    <ProjectReference Include="..\..\MediaBrowser.Providers\MediaBrowser.Providers.csproj" />
+    <ProjectReference Include="..\..\src\Jellyfin.LiveTv\Jellyfin.LiveTv.csproj" />
     <ProjectReference Include="..\Jellyfin.Server.Integration.Tests\Jellyfin.Server.Integration.Tests.csproj" />
     <ProjectReference Include="..\..\src\Jellyfin.Database\Jellyfin.Database.Implementations\Jellyfin.Database.Implementations.csproj" />
   </ItemGroup>

tests/Jellyfin.Server.Implementations.Tests/Library/LibraryManagerSortTests.cs

@@ -7,6 +7,7 @@ using Emby.Naming.Common;
 using Emby.Server.Implementations.Library;
 using Emby.Server.Implementations.Sorting;
 using Jellyfin.Data.Enums;
+using Jellyfin.Database.Implementations.Entities;
 using Jellyfin.Database.Implementations.Enums;
 using MediaBrowser.Controller.Configuration;
 using MediaBrowser.Controller.Entities;
@@ -63,13 +64,49 @@ public class LibraryManagerSortTests
         Assert.Equal(new[] { "Alpha", "Mike", "Zulu" }, sorted.Select(i => i.Name));
     }
 
+    [Fact]
+    public void Sort_ComparerThatIgnoresPrefetchedUserData_StillSortsFromLiveReads()
+    {
+        var alpha = new Audio { Name = "Alpha", SortName = "Alpha", Id = Guid.NewGuid() };
+        var zulu = new Audio { Name = "Zulu", SortName = "Zulu", Id = Guid.NewGuid() };
+        var playCounts = new Dictionary<Guid, int> { [alpha.Id] = 1, [zulu.Id] = 9 };
+
+        var userDataManager = new Mock<IUserDataManager>();
+        userDataManager
+            .Setup(u => u.GetUserData(It.IsAny<User>(), It.IsAny<BaseItem>()))
+            .Returns<User, BaseItem>((_, item) => new UserItemData { Key = item.Id.ToString("N"), PlayCount = playCounts[item.Id] });
+        userDataManager
+            .Setup(u => u.GetUserDataBatch(It.IsAny<IReadOnlyList<BaseItem>>(), It.IsAny<User>()))
+            .Returns(new Dictionary<Guid, UserItemData>());
+
+        var libraryManager = CreateLibraryManager(
+            new IBaseItemComparer[] { new PluginPlayCountComparer() },
+            userDataManager);
+
+        var sorted = libraryManager.Sort(
+            new BaseItem[] { alpha, zulu },
+            new User("sorter", "provider", "provider"),
+            new[] { (ItemSortBy.PlayCount, SortOrder.Descending) }).ToArray();
+
+        Assert.Equal(new[] { "Zulu", "Alpha" }, sorted.Select(i => i.Name));
+        userDataManager.Verify(u => u.GetUserData(It.IsAny<User>(), It.IsAny<BaseItem>()), Times.AtLeastOnce);
+    }
+
     private static Folder MakeFolder(string name, DateTime dateLastMediaAdded)
         => new() { Name = name, Id = Guid.NewGuid(), DateLastMediaAdded = dateLastMediaAdded };
 
-    private static Emby.Server.Implementations.Library.LibraryManager CreateLibraryManager(IReadOnlyCollection<IBaseItemComparer> comparers)
+    private static Emby.Server.Implementations.Library.LibraryManager CreateLibraryManager(
+        IReadOnlyCollection<IBaseItemComparer> comparers,
+        Mock<IUserDataManager>? userDataManager = null)
     {
         var fixture = new Fixture().Customize(new AutoMoqCustomization());
         fixture.Register(() => new NamingOptions());
+
+        if (userDataManager is not null)
+        {
+            fixture.Inject(userDataManager.Object);
+        }
+
         var configMock = fixture.Freeze<Mock<IServerConfigurationManager>>();
         configMock.Setup(c => c.ApplicationPaths.ProgramDataPath).Returns("/data");
         BaseItem.ConfigurationManager ??= configMock.Object;
@@ -86,4 +123,22 @@ public class LibraryManagerSortTests
                 fixture.Create<IEnumerable<ILibraryPostScanTask>>()))
             .Create();
     }
+
+    /// <summary>
+    /// A comparer of the shape a third-party plugin ships: it implements
+    /// <see cref="IUserBaseItemComparer"/> without ever mentioning PrefetchedUserData.
+    /// </summary>
+    public sealed class PluginPlayCountComparer : IUserBaseItemComparer
+    {
+        public User User { get; set; } = null!;
+
+        public IUserManager UserManager { get; set; } = null!;
+
+        public IUserDataManager UserDataManager { get; set; } = null!;
+
+        public ItemSortBy Type => ItemSortBy.PlayCount;
+
+        public int Compare(BaseItem? x, BaseItem? y)
+            => UserDataManager.GetUserData(User, x!)!.PlayCount.CompareTo(UserDataManager.GetUserData(User, y!)!.PlayCount);
+    }
 }

tests/Jellyfin.Server.Implementations.Tests/Library/UserDataManagerTests.cs

@@ -1,5 +1,4 @@
 using System;
-using System.Collections.Generic;
 using Emby.Server.Implementations.Library;
 using Jellyfin.Database.Implementations;
 using Jellyfin.Database.Implementations.Entities;
@@ -49,6 +48,12 @@ public sealed class UserDataManagerTests : IDisposable
         {
             Id = Guid.NewGuid()
         };
+
+        using (var ctx = CreateDbContext())
+        {
+            ctx.Users.Add(_user);
+            ctx.SaveChanges();
+        }
     }
 
     public void Dispose()
@@ -78,6 +83,23 @@ public sealed class UserDataManagerTests : IDisposable
         };
     }
 
+    private void Seed(AudioBook item, params UserData[] rows)
+    {
+        using var ctx = CreateDbContext();
+        ctx.BaseItems.Add(new BaseItemEntity { Id = item.Id, Type = typeof(AudioBook).FullName! });
+        ctx.UserData.AddRange(rows);
+        ctx.SaveChanges();
+    }
+
+    private User CreateOtherUser()
+    {
+        var user = new User("other", "auth-provider", "reset-provider") { Id = Guid.NewGuid() };
+        using var ctx = CreateDbContext();
+        ctx.Users.Add(user);
+        ctx.SaveChanges();
+        return user;
+    }
+
     private UserData CreateUserDataRow(AudioBook item, string key, long positionTicks)
     {
         return new UserData
@@ -98,11 +120,10 @@ public sealed class UserDataManagerTests : IDisposable
         var currentKey = item.GetUserDataKeys()[0];
 
         // the retired-key row comes first to ensure selection is by key, not row order
-        item.UserData = new List<UserData>
-        {
+        Seed(
+            item,
             CreateUserDataRow(item, "Author-Old Album-0001Old File Name", 111),
-            CreateUserDataRow(item, currentKey, 222)
-        };
+            CreateUserDataRow(item, currentKey, 222));
 
         var userData = _userDataManager.GetUserData(_user, item);
 
@@ -117,11 +138,10 @@ public sealed class UserDataManagerTests : IDisposable
         var item = CreateAudioBook();
         var idKey = item.GetUserDataKeys()[1];
 
-        item.UserData = new List<UserData>
-        {
+        Seed(
+            item,
             CreateUserDataRow(item, "Author-Old Album-0001Old File Name", 111),
-            CreateUserDataRow(item, idKey, 333)
-        };
+            CreateUserDataRow(item, idKey, 333));
 
         var userData = _userDataManager.GetUserData(_user, item);
 
@@ -135,10 +155,7 @@ public sealed class UserDataManagerTests : IDisposable
     {
         var item = CreateAudioBook();
 
-        item.UserData = new List<UserData>
-        {
-            CreateUserDataRow(item, "Author-Old Album-0001Old File Name", 111)
-        };
+        Seed(item, CreateUserDataRow(item, "Author-Old Album-0001Old File Name", 111));
 
         var userData = _userDataManager.GetUserData(_user, item);
 
@@ -150,7 +167,7 @@ public sealed class UserDataManagerTests : IDisposable
     public void GetUserData_NoRows_ReturnsDefaultWithPrimaryKey()
     {
         var item = CreateAudioBook();
-        item.UserData = new List<UserData>();
+        Seed(item);
 
         var userData = _userDataManager.GetUserData(_user, item);
 
@@ -166,13 +183,9 @@ public sealed class UserDataManagerTests : IDisposable
         var currentKey = item.GetUserDataKeys()[0];
 
         var otherUserRow = CreateUserDataRow(item, currentKey, 999);
-        otherUserRow.UserId = Guid.NewGuid();
+        otherUserRow.UserId = CreateOtherUser().Id;
 
-        item.UserData = new List<UserData>
-        {
-            otherUserRow,
-            CreateUserDataRow(item, currentKey, 222)
-        };
+        Seed(item, otherUserRow, CreateUserDataRow(item, currentKey, 222));
 
         var userData = _userDataManager.GetUserData(_user, item);
 
@@ -183,23 +196,15 @@ public sealed class UserDataManagerTests : IDisposable
     [Fact]
     public void GetUserDataBatch_DatabaseFallback_ResolvesRowsByKeyOrder()
     {
-        // no preloaded navigation data, so the batch takes the database fallback
         var fossilItem = CreateAudioBook();
         var retiredItem = CreateAudioBook();
 
-        using (var ctx = CreateDbContext())
-        {
-            ctx.Users.Add(_user);
-            ctx.BaseItems.Add(new BaseItemEntity { Id = fossilItem.Id, Type = typeof(AudioBook).FullName! });
-            ctx.BaseItems.Add(new BaseItemEntity { Id = retiredItem.Id, Type = typeof(AudioBook).FullName! });
-
-            // the stale id-key row is inserted first so selection by row order would return it
-            ctx.UserData.AddRange(
-                CreateUserDataRow(fossilItem, fossilItem.GetUserDataKeys()[1], 111),
-                CreateUserDataRow(fossilItem, fossilItem.GetUserDataKeys()[0], 222),
-                CreateUserDataRow(retiredItem, "Author-Old Album-0001Old File Name", 333));
-            ctx.SaveChanges();
-        }
+        // the stale id-key row is inserted first so selection by row order would return it
+        Seed(
+            fossilItem,
+            CreateUserDataRow(fossilItem, fossilItem.GetUserDataKeys()[1], 111),
+            CreateUserDataRow(fossilItem, fossilItem.GetUserDataKeys()[0], 222));
+        Seed(retiredItem, CreateUserDataRow(retiredItem, "Author-Old Album-0001Old File Name", 333));
 
         var result = _userDataManager.GetUserDataBatch([fossilItem, retiredItem], _user);
 

tests/Jellyfin.Server.Implementations.Tests/ScheduledTasks/ScanLeaderOptionsTests.cs

@@ -1,6 +1,8 @@
 using System;
 using System.Collections.Generic;
+using System.IO;
 using System.Linq;
+using System.Reflection;
 using System.Runtime.CompilerServices;
 using Emby.Server.Implementations.ScheduledTasks.Tasks;
 using MediaBrowser.Controller.ScheduledTasks;
@@ -11,6 +13,14 @@ namespace Jellyfin.Server.Implementations.Tests.ScheduledTasks;
 
 public class ScanLeaderOptionsTests
 {
+    private static readonly Assembly[] _taskAssemblies =
+    {
+        typeof(DeleteTranscodeFileTask).Assembly,
+        typeof(MediaBrowser.Providers.Lyric.LyricScheduledTask).Assembly,
+        typeof(Jellyfin.LiveTv.Guide.RefreshGuideScheduledTask).Assembly,
+        typeof(Jellyfin.MediaEncoding.Hls.ScheduledTasks.KeyframeExtractionScheduledTask).Assembly
+    };
+
     /// <summary>
     /// A gated key that matches no registered task silently stops gating anything, so the default
     /// set is pinned to the task keys that actually exist in the build.
@@ -18,28 +28,92 @@ public class ScanLeaderOptionsTests
     [Fact]
     public void DefaultGatedTaskKeys_Should_MatchRegisteredScheduledTasks()
     {
-        var registeredKeys = DiscoverScheduledTaskKeys();
+        var registeredKeys = DiscoverScheduledTaskKeys(_taskAssemblies);
 
         Assert.NotEmpty(registeredKeys);
-        Assert.Empty(new ScanLeaderOptions().GatedTaskKeys.Except(registeredKeys, StringComparer.Ordinal));
+
+        var unmatched = new ScanLeaderOptions().GatedTaskKeys.Except(registeredKeys, StringComparer.Ordinal).ToList();
+        Assert.True(
+            unmatched.Count == 0,
+            $"Gated keys match no scheduled task: {string.Join(", ", unmatched)}. Known keys: {string.Join(", ", registeredKeys.Order(StringComparer.Ordinal))}");
     }
 
-    private static HashSet<string> DiscoverScheduledTaskKeys()
+    /// <summary>
+    /// A key dropped from the default set silently un-gates that task on every replica, so the whole
+    /// set is pinned against a hand-maintained expectation rather than read back from the options.
+    /// </summary>
+    [Fact]
+    public void DefaultGatedTaskKeys_Should_BeTheExpectedSet()
     {
-        var keys = new HashSet<string>(StringComparer.Ordinal);
-        var assemblies = new[]
+        string[] expected =
         {
-            typeof(DeleteTranscodeFileTask).Assembly,
-            typeof(Jellyfin.MediaEncoding.Hls.ScheduledTasks.KeyframeExtractionScheduledTask).Assembly
+            "AudioNormalization",
+            "CleanupUserDataTask",
+            "DownloadLyrics",
+            "DownloadSubtitles",
+            "KeyframeExtraction",
+            "MoveTrickplayImages",
+            "OptimizeDatabaseTask",
+            "PluginUpdates",
+            "RefreshChapterImages",
+            "RefreshGuide",
+            "RefreshInternetChannels",
+            "RefreshLibrary",
+            "RefreshPeople",
+            "RefreshTrickplayImages",
+            "TaskExtractMediaSegments",
+            "TmdbRefreshUpcomingEpisodes"
         };
 
-        foreach (var type in assemblies.SelectMany(a => a.GetTypes()))
+        var actual = new ScanLeaderOptions().GatedTaskKeys;
+        var missing = expected.Except(actual, StringComparer.Ordinal).ToList();
+        var unexpected = actual.Except(expected, StringComparer.Ordinal).ToList();
+
+        Assert.True(
+            missing.Count == 0 && unexpected.Count == 0,
+            $"Default gated task keys drifted. Missing: {Describe(missing)}. Unexpected: {Describe(unexpected)}.");
+    }
+
+    /// <summary>
+    /// The key universe is only as complete as the assemblies it is read from, so a task added to an
+    /// unscanned assembly must fail here rather than narrow what the previous test can catch.
+    /// </summary>
+    [Fact]
+    public void TaskAssemblies_Should_CoverEveryAssemblyDeclaringScheduledTasks()
+    {
+        var scanned = _taskAssemblies.Select(a => a.GetName().Name).ToHashSet(StringComparer.Ordinal);
+        var missing = new List<string>();
+
+        foreach (var path in Directory.EnumerateFiles(AppContext.BaseDirectory, "*.dll"))
         {
-            if (type.IsAbstract || type.IsInterface || !typeof(IScheduledTask).IsAssignableFrom(type))
+            var name = Path.GetFileNameWithoutExtension(path);
+            if (scanned.Contains(name)
+                || name.EndsWith(".Tests", StringComparison.Ordinal)
+                || !(name.StartsWith("Jellyfin.", StringComparison.Ordinal)
+                    || name.StartsWith("Emby.", StringComparison.Ordinal)
+                    || name.StartsWith("MediaBrowser.", StringComparison.Ordinal)))
             {
                 continue;
             }
 
+            if (GetScheduledTaskTypes(Assembly.LoadFrom(path)).Any())
+            {
+                missing.Add(name);
+            }
+        }
+
+        Assert.True(missing.Count == 0, $"Assemblies declaring scheduled tasks but not scanned: {string.Join(", ", missing)}");
+    }
+
+    private static string Describe(IReadOnlyCollection<string> keys)
+        => keys.Count == 0 ? "none" : string.Join(", ", keys.Order(StringComparer.Ordinal));
+
+    private static HashSet<string> DiscoverScheduledTaskKeys(IEnumerable<Assembly> assemblies)
+    {
+        var keys = new HashSet<string>(StringComparer.Ordinal);
+
+        foreach (var type in assemblies.SelectMany(GetScheduledTaskTypes))
+        {
             // Task keys are constant expressions, so an uninitialised instance is enough to read
             // them without standing up each task's dependency graph.
             var task = (IScheduledTask)RuntimeHelpers.GetUninitializedObject(type);
@@ -48,4 +122,21 @@ public class ScanLeaderOptionsTests
 
         return keys;
     }
+
+    private static IEnumerable<Type> GetScheduledTaskTypes(Assembly assembly)
+    {
+        Type?[] types;
+        try
+        {
+            types = assembly.GetTypes();
+        }
+        catch (ReflectionTypeLoadException ex)
+        {
+            types = ex.Types;
+        }
+
+        return types
+            .Where(t => t is not null && !t.IsAbstract && !t.IsInterface && typeof(IScheduledTask).IsAssignableFrom(t))
+            .Select(t => t!);
+    }
 }

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.Implementations.Tests/TV/TVSeriesManagerNextUpTests.cs

@@ -0,0 +1,94 @@
+using System;
+using System.Collections.Generic;
+using System.Globalization;
+using System.Linq;
+using Emby.Server.Implementations.TV;
+using Jellyfin.Database.Implementations.Entities;
+using MediaBrowser.Controller.Configuration;
+using MediaBrowser.Controller.Dto;
+using MediaBrowser.Controller.Entities;
+using MediaBrowser.Controller.Entities.TV;
+using MediaBrowser.Controller.Library;
+using MediaBrowser.Controller.Persistence;
+using MediaBrowser.Model.Configuration;
+using MediaBrowser.Model.Querying;
+using Moq;
+using Xunit;
+
+namespace Jellyfin.Server.Implementations.Tests.TV;
+
+public class TVSeriesManagerNextUpTests
+{
+    [Theory]
+    [InlineData(1)]
+    [InlineData(25)]
+    [InlineData(200)]
+    public void GetNextUp_ReadsUserDataInABoundedNumberOfQueries(int seriesCount)
+    {
+        var user = new User("next-up", "provider", "provider");
+        var libraryManager = new Mock<ILibraryManager>();
+        var userDataManager = new Mock<IUserDataManager>();
+
+        var seriesKeys = Enumerable.Range(0, seriesCount)
+            .Select(i => i.ToString(CultureInfo.InvariantCulture))
+            .ToList();
+
+        var batch = seriesKeys.ToDictionary(
+            key => key,
+            key => new NextUpEpisodeBatchResult
+            {
+                NextUp = new Episode { Id = Guid.NewGuid(), Name = "Next " + key },
+                LastWatched = new Episode { Id = Guid.NewGuid(), Name = "Watched " + key }
+            });
+
+        libraryManager
+            .Setup(l => l.GetNextUpSeriesKeys(It.IsAny<InternalItemsQuery>(), It.IsAny<IReadOnlyCollection<BaseItem>>(), It.IsAny<DateTime>()))
+            .Returns(seriesKeys);
+        libraryManager
+            .Setup(l => l.GetNextUpEpisodesBatch(It.IsAny<InternalItemsQuery>(), It.IsAny<IReadOnlyList<string>>(), It.IsAny<bool>(), It.IsAny<bool>()))
+            .Returns(batch);
+        libraryManager.Setup(l => l.GetLinkedAlternateVersions(It.IsAny<Video>())).Returns([]);
+        libraryManager.Setup(l => l.GetLocalAlternateVersionIds(It.IsAny<Video>())).Returns([]);
+
+        var batchReads = 0;
+        userDataManager
+            .Setup(u => u.GetUserDataBatch(It.IsAny<IReadOnlyList<BaseItem>>(), It.IsAny<User>()))
+            .Returns<IReadOnlyList<BaseItem>, User>((items, _) =>
+            {
+                batchReads++;
+                return items.DistinctBy(i => i.Id).ToDictionary(
+                    i => i.Id,
+                    i => new UserItemData { Key = i.Id.ToString("N", CultureInfo.InvariantCulture) });
+            });
+
+        var previousLibraryManager = BaseItem.LibraryManager;
+        BaseItem.LibraryManager = libraryManager.Object;
+
+        try
+        {
+            var manager = new TVSeriesManager(userDataManager.Object, libraryManager.Object, CreateConfigurationManager());
+
+            var result = manager.GetNextUp(
+                new NextUpQuery { User = user, EnableTotalRecordCount = true },
+                [],
+                new DtoOptions(false));
+
+            Assert.Equal(seriesCount, result.TotalRecordCount);
+
+            // Selection, the resume check and the last played date: three reads whatever the library holds.
+            Assert.Equal(3, batchReads);
+            userDataManager.Verify(u => u.GetUserData(It.IsAny<User>(), It.IsAny<BaseItem>()), Times.Never);
+        }
+        finally
+        {
+            BaseItem.LibraryManager = previousLibraryManager;
+        }
+    }
+
+    private static IServerConfigurationManager CreateConfigurationManager()
+    {
+        var configurationManager = new Mock<IServerConfigurationManager>();
+        configurationManager.SetupGet(c => c.Configuration).Returns(new ServerConfiguration());
+        return configurationManager.Object;
+    }
+}

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

@@ -0,0 +1,66 @@
+using System;
+using System.Threading.Tasks;
+using StackExchange.Redis;
+using Testcontainers.Redis;
+
+namespace Jellyfin.Server.Tests.HighAvailability;
+
+/// <summary>
+/// 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
+{
+    /// <summary>
+    /// The connection string of an already running server.
+    /// </summary>
+    public const string ConnectionStringVariable = "JELLYFIN_TEST_REDIS";
+
+    private readonly RedisContainer? _container;
+
+    private RedisTestServer(RedisContainer? container, string connectionString)
+    {
+        _container = container;
+        ConnectionString = connectionString;
+    }
+
+    /// <summary>
+    /// Gets the connection string of the running server.
+    /// </summary>
+    public string ConnectionString { get; }
+
+    /// <summary>
+    /// Starts or attaches to a server and connects to it.
+    /// </summary>
+    /// <returns>The running server.</returns>
+    public static async Task<RedisTestServer> StartAsync()
+    {
+        var provided = Environment.GetEnvironmentVariable(ConnectionStringVariable);
+        if (!string.IsNullOrWhiteSpace(provided))
+        {
+            return new RedisTestServer(null, provided);
+        }
+
+        var container = new RedisBuilder("valkey/valkey:8-alpine").Build();
+        await container.StartAsync().ConfigureAwait(false);
+
+        return new RedisTestServer(container, container.GetConnectionString());
+    }
+
+    /// <summary>
+    /// Opens a connection to the server.
+    /// </summary>
+    /// <returns>The connection.</returns>
+    public async Task<IConnectionMultiplexer> ConnectAsync()
+        => await ConnectionMultiplexer.ConnectAsync(ConnectionString).ConfigureAwait(false);
+
+    /// <inheritdoc/>
+    public async ValueTask DisposeAsync()
+    {
+        if (_container is not null)
+        {
+            await _container.DisposeAsync().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/Item/PostgreSqlNextUpServiceTests.cs

@@ -0,0 +1,234 @@
+using System;
+using System.Collections.Generic;
+using System.Security.Claims;
+using System.Threading;
+using System.Threading.Tasks;
+using Emby.Server.Implementations.Data;
+using Jellyfin.Api.Constants;
+using Jellyfin.Api.Controllers;
+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.Item;
+using Jellyfin.Server.Tests.Migrations;
+using MediaBrowser.Controller.Dto;
+using MediaBrowser.Controller.Entities;
+using MediaBrowser.Controller.Library;
+using MediaBrowser.Controller.Persistence;
+using MediaBrowser.Controller.TV;
+using MediaBrowser.Model.Querying;
+using Microsoft.AspNetCore.Http;
+using Microsoft.AspNetCore.Mvc;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.Logging.Abstractions;
+using Moq;
+using Npgsql;
+using Xunit;
+using BaseItemKind = Jellyfin.Data.Enums.BaseItemKind;
+using User = Jellyfin.Database.Implementations.Entities.User;
+
+namespace Jellyfin.Server.Tests.Item;
+
+/// <summary>
+/// Drives the Next Up cutoff from the controller into a real PostgreSQL. The model binder hands a
+/// query-string date over as <see cref="DateTimeKind.Unspecified"/>, and Npgsql refuses to write anything
+/// but <see cref="DateTimeKind.Utc"/> to <c>timestamp with time zone</c>; SQLite takes every kind, so an
+/// unnormalised cutoff only ever fails here.
+/// </summary>
+[Trait("Category", "RequiresDocker")]
+public sealed class PostgreSqlNextUpServiceTests : IAsyncLifetime
+{
+    private static readonly Guid _libraryId = Guid.Parse("aaaaaaaa-0000-0000-0000-000000000001");
+    private static readonly Guid _otherLibraryId = Guid.Parse("aaaaaaaa-0000-0000-0000-000000000002");
+    private static readonly Guid _userId = Guid.Parse("bbbbbbbb-0000-0000-0000-000000000001");
+
+    private static readonly Guid _recentWatchedId = Guid.Parse("cccccccc-0000-0000-0000-000000000001");
+    private static readonly Guid _recentOlderId = Guid.Parse("cccccccc-0000-0000-0000-000000000002");
+    private static readonly Guid _staleWatchedId = Guid.Parse("cccccccc-0000-0000-0000-000000000003");
+    private static readonly Guid _unwatchedId = Guid.Parse("cccccccc-0000-0000-0000-000000000004");
+    private static readonly Guid _foreignLibraryId = Guid.Parse("cccccccc-0000-0000-0000-000000000005");
+
+    private static readonly DateTime _recentPlayedAt = new DateTime(2026, 3, 1, 12, 0, 0, DateTimeKind.Utc);
+    private static readonly DateTime _stalePlayedAt = new DateTime(2020, 1, 1, 12, 0, 0, DateTimeKind.Utc);
+
+    private readonly ItemTypeLookup _itemTypeLookup = new();
+    private readonly User _user = new User("next-up", "auth", "reset") { Id = _userId };
+
+    private PostgreSqlTestServer _server = null!;
+    private NpgsqlDataSource _dataSource = null!;
+    private NextUpService _service = null!;
+
+    public async ValueTask InitializeAsync()
+    {
+        _server = await PostgreSqlTestServer.StartAsync().ConfigureAwait(false);
+        var connectionString = await _server.CreateDatabaseAsync("next_up_service", TestContext.Current.CancellationToken).ConfigureAwait(false);
+        _dataSource = new NpgsqlDataSourceBuilder(connectionString).Build();
+
+        var context = CreateDbContext();
+        await using (context.ConfigureAwait(false))
+        {
+            await context.Database.EnsureCreatedAsync(TestContext.Current.CancellationToken).ConfigureAwait(false);
+        }
+
+        var factory = new Mock<IDbContextFactory<JellyfinDbContext>>();
+        factory.Setup(f => f.CreateDbContext()).Returns(CreateDbContext);
+        factory.Setup(f => f.CreateDbContextAsync(It.IsAny<CancellationToken>())).ReturnsAsync(CreateDbContext);
+
+        _service = new NextUpService(factory.Object, _itemTypeLookup, new Mock<IItemQueryHelpers>().Object);
+
+        await SeedAsync().ConfigureAwait(false);
+    }
+
+    public async ValueTask DisposeAsync()
+    {
+        await _dataSource.DisposeAsync().ConfigureAwait(false);
+        await _server.DisposeAsync().ConfigureAwait(false);
+    }
+
+    /// <summary>
+    /// A cutoff on the query string, which the model binder leaves unspecified.
+    /// </summary>
+    [Fact]
+    public void GetNextUpSeriesKeys_WithSuppliedCutoff_DropsSeriesPlayedBeforeIt()
+    {
+        var cutoff = RunController(new DateTime(2025, 1, 1, 0, 0, 0, DateTimeKind.Unspecified)).NextUpDateCutoff;
+
+        var keys = _service.GetNextUpSeriesKeys(CreateFilter(), cutoff);
+
+        Assert.Equal(new[] { "series-recent" }, keys);
+    }
+
+    /// <summary>
+    /// The home-screen row, where the client sends no cutoff and the query default stands in.
+    /// </summary>
+    [Fact]
+    public void GetNextUpSeriesKeys_WithoutSuppliedCutoff_ReturnsWatchedSeriesNewestFirst()
+    {
+        var cutoff = RunController(null).NextUpDateCutoff;
+
+        var keys = _service.GetNextUpSeriesKeys(CreateFilter(), cutoff);
+
+        Assert.Equal(new[] { "series-recent", "series-stale" }, keys);
+    }
+
+    /// <summary>
+    /// Calls <c>GET /Shows/NextUp</c> and hands back the query it built for the series lookup.
+    /// </summary>
+    private NextUpQuery RunController(DateTime? nextUpDateCutoff)
+    {
+        var userManager = new Mock<IUserManager>();
+        userManager.Setup(m => m.GetUserById(_userId)).Returns(_user);
+
+        var dtoService = new Mock<IDtoService>();
+        dtoService.Setup(s => s.GetBaseItemDtos(
+                It.IsAny<IReadOnlyList<BaseItem>>(),
+                It.IsAny<DtoOptions>(),
+                It.IsAny<User>(),
+                It.IsAny<BaseItem>(),
+                It.IsAny<bool>()))
+            .Returns([]);
+
+        NextUpQuery? captured = null;
+        var tvSeriesManager = new Mock<ITVSeriesManager>();
+        tvSeriesManager.Setup(m => m.GetNextUp(It.IsAny<NextUpQuery>(), It.IsAny<DtoOptions>()))
+            .Callback<NextUpQuery, DtoOptions>((query, _) => captured = query)
+            .Returns(new QueryResult<BaseItem>());
+
+        var controller = new TvShowsController(
+            userManager.Object,
+            new Mock<ILibraryManager>().Object,
+            dtoService.Object,
+            tvSeriesManager.Object)
+        {
+            ControllerContext = new ControllerContext
+            {
+                HttpContext = new DefaultHttpContext
+                {
+                    User = new ClaimsPrincipal(new ClaimsIdentity([new Claim(InternalClaimTypes.UserId, _userId.ToString("D"))], "Test"))
+                }
+            }
+        };
+
+        controller.GetNextUp(null, null, null, [], null, null, null, null, [], null, nextUpDateCutoff);
+
+        return captured!;
+    }
+
+    private InternalItemsQuery CreateFilter()
+    {
+        return new InternalItemsQuery(_user) { TopParentIds = [_libraryId] };
+    }
+
+    private JellyfinDbContext CreateDbContext()
+    {
+        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 SeedAsync()
+    {
+        var context = CreateDbContext();
+        await using (context.ConfigureAwait(false))
+        {
+            context.Users.Add(_user);
+
+            // The newest play of a series decides its place, so the older episode must not pull it down.
+            var recentWatched = AddEpisode(context, _recentWatchedId, "series-recent", _libraryId);
+            AddUserData(context, recentWatched, _recentPlayedAt);
+            var recentOlder = AddEpisode(context, _recentOlderId, "series-recent", _libraryId);
+            AddUserData(context, recentOlder, _stalePlayedAt);
+
+            var staleWatched = AddEpisode(context, _staleWatchedId, "series-stale", _libraryId);
+            AddUserData(context, staleWatched, _stalePlayedAt);
+
+            // Never played, and played but outside the requested libraries: both stay out.
+            AddEpisode(context, _unwatchedId, "series-unwatched", _libraryId);
+            var foreign = AddEpisode(context, _foreignLibraryId, "series-foreign", _otherLibraryId);
+            AddUserData(context, foreign, _recentPlayedAt);
+
+            await context.SaveChangesAsync(TestContext.Current.CancellationToken).ConfigureAwait(false);
+        }
+    }
+
+    private BaseItemEntity AddEpisode(JellyfinDbContext context, Guid id, string seriesKey, Guid topParentId)
+    {
+        var episode = new BaseItemEntity
+        {
+            Id = id,
+            Type = _itemTypeLookup.BaseItemKindNames[BaseItemKind.Episode],
+            Name = seriesKey + "-" + id.ToString("N"),
+            SeriesPresentationUniqueKey = seriesKey,
+            PresentationUniqueKey = id.ToString("N"),
+            TopParentId = topParentId,
+            MediaType = "Video",
+            IsFolder = false,
+            IsVirtualItem = false
+        };
+
+        context.BaseItems.Add(episode);
+        return episode;
+    }
+
+    private void AddUserData(JellyfinDbContext context, BaseItemEntity item, DateTime lastPlayedDate)
+    {
+        context.UserData.Add(new UserData
+        {
+            CustomDataKey = item.Id.ToString("N"),
+            ItemId = item.Id,
+            Item = item,
+            UserId = _userId,
+            User = _user,
+            LastPlayedDate = lastPlayedDate,
+            Played = true,
+            PlayCount = 1
+        });
+    }
+}

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

@@ -11,7 +11,9 @@
     <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" />
     <PackageReference Include="xunit.runner.visualstudio">
       <PrivateAssets>all</PrivateAssets>

tests/Jellyfin.Server.Tests/Library/UserDataManagerReplicaTests.cs

@@ -0,0 +1,243 @@
+using System;
+using System.Collections.Generic;
+using System.Globalization;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Emby.Server.Implementations.Library;
+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.Tests.Migrations;
+using MediaBrowser.Controller.Configuration;
+using MediaBrowser.Model.Configuration;
+using MediaBrowser.Model.Entities;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.Logging.Abstractions;
+using Moq;
+using Npgsql;
+using Xunit;
+using AudioBook = MediaBrowser.Controller.Entities.AudioBook;
+
+namespace Jellyfin.Server.Tests.Library;
+
+/// <summary>
+/// Two independently constructed <see cref="UserDataManager"/> instances over one PostgreSQL database are the
+/// in-process stand-in for two replicas sharing one database: what either of them writes, the other has to
+/// see on its very next read, and a read-modify-write on one must not roll back the other's.
+/// </summary>
+[Trait("Category", "RequiresDocker")]
+public sealed class UserDataManagerReplicaTests : IClassFixture<UserDataManagerReplicaTests.DatabaseFixture>
+{
+    private static readonly long _quarterIn = TimeSpan.FromMinutes(20).Ticks;
+
+    private readonly NpgsqlDataSource _dataSource;
+
+    public UserDataManagerReplicaTests(DatabaseFixture fixture)
+    {
+        _dataSource = fixture.DataSource;
+    }
+
+    /// <summary>
+    /// A resume position written by the replica serving the playback tick has to be the position the next
+    /// request reads, whichever replica it lands on - both through the single item read the write path uses
+    /// and through the batch read the library pages render from.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    [Fact]
+    public async Task ResumePositionWrittenOnOneReplica_IsReadOnAnother()
+    {
+        var cancellationToken = TestContext.Current.CancellationToken;
+        var itemId = Guid.NewGuid();
+        var user = await CreateUserAndItemAsync(_dataSource, itemId, cancellationToken);
+
+        var replicaA = CreateManager(_dataSource);
+        var replicaB = CreateManager(_dataSource);
+        var itemOnA = new AudioBook { Id = itemId, Name = "Replica Book" };
+        var itemOnB = new AudioBook { Id = itemId, Name = "Replica Book" };
+
+        var early = replicaA.GetUserData(user, itemOnA)!;
+        early.PlaybackPositionTicks = TimeSpan.FromMinutes(5).Ticks;
+        replicaA.SaveUserData(user, itemOnA, early, UserDataSaveReason.PlaybackProgress, cancellationToken);
+
+        // Replica B materialised the item before the later tick, so it holds the earlier row in memory.
+        itemOnB.UserData = await LoadUserDataAsync(_dataSource, itemId, cancellationToken);
+
+        var later = replicaA.GetUserData(user, itemOnA)!;
+        later.PlaybackPositionTicks = _quarterIn;
+        replicaA.SaveUserData(user, itemOnA, later, UserDataSaveReason.PlaybackProgress, cancellationToken);
+
+        Assert.Equal(_quarterIn, replicaB.GetUserData(user, itemOnB)!.PlaybackPositionTicks);
+        Assert.Equal(_quarterIn, replicaB.GetUserDataBatch([itemOnB], user)[itemId].PlaybackPositionTicks);
+    }
+
+    /// <summary>
+    /// The playback tick is a read-modify-write of the whole row, so a tick served by one replica must build
+    /// on the favourite another replica just recorded instead of writing it back out.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    [Fact]
+    public async Task PlaybackTickOnOneReplica_KeepsFavouriteSetOnAnother()
+    {
+        var cancellationToken = TestContext.Current.CancellationToken;
+        var itemId = Guid.NewGuid();
+        var user = await CreateUserAndItemAsync(_dataSource, itemId, cancellationToken);
+
+        var replicaA = CreateManager(_dataSource);
+        var replicaB = CreateManager(_dataSource);
+        var itemOnA = new AudioBook { Id = itemId, Name = "Replica Book" };
+        var itemOnB = new AudioBook { Id = itemId, Name = "Replica Book" };
+
+        var seed = replicaA.GetUserData(user, itemOnA)!;
+        seed.PlaybackPositionTicks = TimeSpan.FromMinutes(5).Ticks;
+        replicaA.SaveUserData(user, itemOnA, seed, UserDataSaveReason.PlaybackProgress, cancellationToken);
+
+        // Replica B is serving the playback session and read the item before the favourite was recorded.
+        itemOnB.UserData = await LoadUserDataAsync(_dataSource, itemId, cancellationToken);
+
+        var favourited = replicaA.GetUserData(user, itemOnA)!;
+        favourited.IsFavorite = true;
+        replicaA.SaveUserData(user, itemOnA, favourited, UserDataSaveReason.UpdateUserRating, cancellationToken);
+
+        var tick = replicaB.GetUserData(user, itemOnB)!;
+        tick.PlaybackPositionTicks = _quarterIn;
+        replicaB.SaveUserData(user, itemOnB, tick, UserDataSaveReason.PlaybackProgress, cancellationToken);
+
+        var stored = replicaA.GetUserData(user, itemOnA)!;
+        Assert.True(stored.IsFavorite);
+        Assert.Equal(_quarterIn, stored.PlaybackPositionTicks);
+    }
+
+    /// <summary>
+    /// A tick that lands on the other replica has to carry the position forward from where the session
+    /// actually is, not from the position that replica happened to have in memory.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    [Fact]
+    public async Task PlaybackTickOnOneReplica_ResumesFromThePositionAnotherWrote()
+    {
+        var cancellationToken = TestContext.Current.CancellationToken;
+        var itemId = Guid.NewGuid();
+        var user = await CreateUserAndItemAsync(_dataSource, itemId, cancellationToken);
+
+        var replicaA = CreateManager(_dataSource);
+        var replicaB = CreateManager(_dataSource);
+        var itemOnA = new AudioBook { Id = itemId, Name = "Replica Book" };
+        var itemOnB = new AudioBook { Id = itemId, Name = "Replica Book" };
+
+        var seed = replicaA.GetUserData(user, itemOnA)!;
+        seed.PlaybackPositionTicks = TimeSpan.FromMinutes(5).Ticks;
+        replicaA.SaveUserData(user, itemOnA, seed, UserDataSaveReason.PlaybackProgress, cancellationToken);
+
+        itemOnB.UserData = await LoadUserDataAsync(_dataSource, itemId, cancellationToken);
+
+        // The viewer seeks forward and the tick reporting it lands on replica A.
+        var seeked = replicaA.GetUserData(user, itemOnA)!;
+        seeked.PlaybackPositionTicks = _quarterIn;
+        replicaA.SaveUserData(user, itemOnA, seeked, UserDataSaveReason.PlaybackProgress, cancellationToken);
+
+        // The next tick lands on replica B, which adds ten seconds to whatever it reads.
+        var tick = replicaB.GetUserData(user, itemOnB)!;
+        tick.PlaybackPositionTicks += TimeSpan.FromSeconds(10).Ticks;
+        replicaB.SaveUserData(user, itemOnB, tick, UserDataSaveReason.PlaybackProgress, cancellationToken);
+
+        var stored = replicaA.GetUserData(user, itemOnA)!;
+        Assert.Equal(_quarterIn + TimeSpan.FromSeconds(10).Ticks, stored.PlaybackPositionTicks);
+    }
+
+    private static UserDataManager CreateManager(NpgsqlDataSource dataSource)
+    {
+        var config = new Mock<IServerConfigurationManager>();
+        config.SetupGet(c => c.Configuration).Returns(new ServerConfiguration());
+        return new UserDataManager(config.Object, new DataSourceContextFactory(dataSource));
+    }
+
+    private static async Task<ICollection<UserData>> LoadUserDataAsync(NpgsqlDataSource dataSource, Guid itemId, CancellationToken cancellationToken)
+    {
+        var context = CreateContext(dataSource);
+        await using (context.ConfigureAwait(false))
+        {
+            return await context.UserData
+                .AsNoTracking()
+                .Where(e => e.ItemId.Equals(itemId))
+                .ToArrayAsync(cancellationToken)
+                .ConfigureAwait(false);
+        }
+    }
+
+    private static async Task<User> CreateUserAndItemAsync(NpgsqlDataSource dataSource, Guid itemId, CancellationToken cancellationToken)
+    {
+        var context = CreateContext(dataSource);
+        await using (context.ConfigureAwait(false))
+        {
+            var user = new User("replica-user-" + itemId.ToString("N", CultureInfo.InvariantCulture), "provider", "provider");
+            context.Users.Add(user);
+            context.BaseItems.Add(new BaseItemEntity { Id = itemId, Type = typeof(AudioBook).FullName! });
+            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));
+    }
+
+    /// <summary>
+    /// Hands every <see cref="UserDataManager"/> 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>
+    /// Builds the schema once for the whole class. Every test keeps to its own user and item, so one
+    /// database serves all of them and the shared server is spared three schema builds.
+    /// </summary>
+    public sealed class DatabaseFixture : IAsyncLifetime
+    {
+        private PostgreSqlTestServer _server = null!;
+
+        public NpgsqlDataSource DataSource { get; private set; } = null!;
+
+        /// <inheritdoc/>
+        public async ValueTask InitializeAsync()
+        {
+            _server = await PostgreSqlTestServer.StartAsync().ConfigureAwait(false);
+            var connectionString = await _server.CreateDatabaseAsync("userdata_replica", CancellationToken.None).ConfigureAwait(false);
+            DataSource = new NpgsqlDataSourceBuilder(connectionString).Build();
+
+            var context = CreateContext(DataSource);
+            await using (context.ConfigureAwait(false))
+            {
+                await context.Database.EnsureCreatedAsync(CancellationToken.None).ConfigureAwait(false);
+            }
+        }
+
+        /// <inheritdoc/>
+        public async ValueTask DisposeAsync()
+        {
+            await DataSource.DisposeAsync().ConfigureAwait(false);
+            await _server.DisposeAsync().ConfigureAwait(false);
+        }
+    }
+}