diff --git a/.woodpecker/ci.yaml b/.woodpecker/ci.yaml index c1c5e741f0..28857ceb17 100644 --- a/.woodpecker/ci.yaml +++ b/.woodpecker/ci.yaml @@ -47,6 +47,7 @@ steps: # Its data directory lives on the step's ephemeral storage, not on the workspace volume. # Both projects attach to it through JELLYFIN_TEST_POSTGRES and give every test a database of # its own, so nothing here depends on a docker daemon. + # Valkey runs in the step for the same reason, reached through JELLYFIN_TEST_REDIS. - name: postgres-migration-chain image: mcr.microsoft.com/dotnet/sdk:10.0 depends_on: @@ -55,13 +56,16 @@ 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" diff --git a/Emby.Server.Implementations/Session/RedisPodMessageBus.cs b/Emby.Server.Implementations/Session/RedisPodMessageBus.cs new file mode 100644 index 0000000000..bad3d80ecd --- /dev/null +++ b/Emby.Server.Implementations/Session/RedisPodMessageBus.cs @@ -0,0 +1,92 @@ +using System; +using System.Text.Json; +using System.Threading.Tasks; +using Jellyfin.Extensions.Json; +using MediaBrowser.Controller.Session; +using Microsoft.Extensions.Logging; +using StackExchange.Redis; + +namespace Emby.Server.Implementations.Session; + +/// +/// A Redis pub/sub . Every instance subscribes to a channel named after +/// itself, which keeps addressed delivery working without the instances being routable to each other. +/// +public sealed class RedisPodMessageBus : IPodMessageBus +{ + private const string ChannelPrefix = "jellyfin:pod:"; + + private static readonly JsonSerializerOptions _jsonOptions = JsonDefaults.Options; + + private readonly ISubscriber _subscriber; + private readonly ILogger _logger; + + /// + /// Initializes a new instance of the class. + /// + /// The Redis connection multiplexer. + /// The logger. + public RedisPodMessageBus(IConnectionMultiplexer redis, ILogger logger) + { + ArgumentNullException.ThrowIfNull(redis); + + _subscriber = redis.GetSubscriber(); + _logger = logger; + PodId = PodIdentity.Current; + } + + /// + public string PodId { get; } + + /// + public void Publish(string targetPod, PodMessage message) + { + ArgumentException.ThrowIfNullOrEmpty(targetPod); + ArgumentNullException.ThrowIfNull(message); + + message.OriginPod = PodId; + + try + { + _subscriber.Publish( + RedisChannel.Literal(ChannelPrefix + targetPod), + JsonSerializer.Serialize(message, _jsonOptions), + CommandFlags.FireAndForget); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Failed to send a {Kind} message to {TargetPod}.", message.Kind, targetPod); + } + } + + /// + public void Subscribe(Func 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 handler, RedisValue value) + { + try + { + var message = JsonSerializer.Deserialize(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."); + } + } +} diff --git a/Emby.Server.Implementations/Session/RedisSessionDirectory.cs b/Emby.Server.Implementations/Session/RedisSessionDirectory.cs new file mode 100644 index 0000000000..20980983a6 --- /dev/null +++ b/Emby.Server.Implementations/Session/RedisSessionDirectory.cs @@ -0,0 +1,139 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using Jellyfin.Extensions.Json; +using MediaBrowser.Controller.Session; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using StackExchange.Redis; + +namespace Emby.Server.Implementations.Session; + +/// +/// A Redis-backed . Each entry is a key with an expiry, so the sessions of +/// an instance that stops refreshing them disappear on their own. +/// +public sealed class RedisSessionDirectory : ISessionDirectory +{ + private const string KeyPrefix = "jellyfin:session:"; + + private static readonly JsonSerializerOptions _jsonOptions = JsonDefaults.Options; + + private readonly IConnectionMultiplexer _redis; + private readonly IDatabase _db; + private readonly SessionDirectoryOptions _options; + private readonly ILogger _logger; + + /// + /// Initializes a new instance of the class. + /// + /// The Redis connection multiplexer. + /// The session directory configuration options. + /// The logger. + public RedisSessionDirectory( + IConnectionMultiplexer redis, + IOptions options, + ILogger logger) + { + ArgumentNullException.ThrowIfNull(redis); + ArgumentNullException.ThrowIfNull(options); + + _redis = redis; + _db = redis.GetDatabase(); + _options = options.Value; + _logger = logger; + } + + private TimeSpan EntryTtl => TimeSpan.FromSeconds(Math.Max(1, _options.EntryTtlSeconds)); + + /// + public async Task PublishAsync(SessionDirectoryEntry entry, CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(entry); + + var sessionId = entry.Session?.Id; + if (string.IsNullOrEmpty(sessionId)) + { + return; + } + + try + { + await _db.StringSetAsync(KeyPrefix + sessionId, JsonSerializer.Serialize(entry, _jsonOptions), EntryTtl).ConfigureAwait(false); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Failed to publish session {SessionId}; it stays invisible to the other instances.", sessionId); + } + } + + /// + public async Task RemoveAsync(string sessionId, CancellationToken cancellationToken = default) + { + try + { + await _db.KeyDeleteAsync(KeyPrefix + sessionId).ConfigureAwait(false); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Failed to remove session {SessionId}; it expires on its own.", sessionId); + } + } + + /// + public async Task> GetAllAsync(CancellationToken cancellationToken = default) + { + var entries = new List(); + + try + { + foreach (var server in _redis.GetServers()) + { + if (!server.IsConnected) + { + continue; + } + + var keys = new List(); + 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))).ConfigureAwait(false); + + foreach (var raw in values) + { + if (!raw.HasValue) + { + continue; + } + + try + { + var entry = JsonSerializer.Deserialize(raw.ToString(), _jsonOptions); + if (entry?.Session is not null) + { + entries.Add(entry); + } + } + catch (JsonException ex) + { + _logger.LogWarning(ex, "Failed to deserialize a session directory entry."); + } + } + } + } + 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(); + } + + return entries; + } +} diff --git a/Emby.Server.Implementations/Session/RemoteSessionController.cs b/Emby.Server.Implementations/Session/RemoteSessionController.cs new file mode 100644 index 0000000000..51cd17a91a --- /dev/null +++ b/Emby.Server.Implementations/Session/RemoteSessionController.cs @@ -0,0 +1,65 @@ +using System; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using Jellyfin.Extensions.Json; +using MediaBrowser.Controller.Session; +using MediaBrowser.Model.Session; + +namespace Emby.Server.Implementations.Session; + +/// +/// Stands in for the websocket of a session another instance holds: messages are forwarded to that +/// instance, which writes them to the connection it owns. +/// +public sealed class RemoteSessionController : ISessionController +{ + private readonly IPodMessageBus _bus; + private readonly string _ownerPod; + private readonly string _sessionId; + + /// + /// Initializes a new instance of the class. + /// + /// The cross-instance bus. + /// The instance holding the connection. + /// The session identifier. + /// Whether the owner reported the session as active. + /// Whether the owner reported the session as controllable. + public RemoteSessionController(IPodMessageBus bus, string ownerPod, string sessionId, bool isSessionActive, bool supportsMediaControl) + { + _bus = bus; + _ownerPod = ownerPod; + _sessionId = sessionId; + IsSessionActive = isSessionActive; + SupportsMediaControl = supportsMediaControl; + } + + /// + public bool IsSessionActive { get; } + + /// + public bool SupportsMediaControl { get; } + + /// + public Task SendMessage(SessionMessageType name, Guid messageId, T data, CancellationToken cancellationToken) + { + var routed = new RoutedSessionMessage + { + SessionId = _sessionId, + MessageType = name, + MessageId = messageId, + Data = JsonSerializer.Serialize(data, JsonDefaults.Options) + }; + + _bus.Publish( + _ownerPod, + new PodMessage + { + Kind = RoutedSessionMessage.Kind, + Payload = JsonSerializer.Serialize(routed, JsonDefaults.Options) + }); + + return Task.CompletedTask; + } +} diff --git a/Emby.Server.Implementations/Session/SessionManager.cs b/Emby.Server.Implementations/Session/SessionManager.cs index 13bf42f437..ae3f3c1fb9 100644 --- a/Emby.Server.Implementations/Session/SessionManager.cs +++ b/Emby.Server.Implementations/Session/SessionManager.cs @@ -5,6 +5,7 @@ using System.Collections.Concurrent; using System.Collections.Generic; using System.Globalization; using System.Linq; +using System.Text.Json; using System.Threading; using System.Threading.Tasks; using Jellyfin.Data; @@ -15,6 +16,7 @@ using Jellyfin.Database.Implementations.Entities; using Jellyfin.Database.Implementations.Entities.Security; using Jellyfin.Database.Implementations.Enums; using Jellyfin.Extensions; +using Jellyfin.Extensions.Json; using MediaBrowser.Common.Events; using MediaBrowser.Common.Extensions; using MediaBrowser.Controller; @@ -39,6 +41,7 @@ using MediaBrowser.Model.SyncPlay; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; using Episode = MediaBrowser.Controller.Entities.TV.Episode; namespace Emby.Server.Implementations.Session @@ -60,6 +63,9 @@ 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 CancellationTokenRegistration _shutdownCallback; private readonly ConcurrentDictionary _activeConnections = new(StringComparer.OrdinalIgnoreCase); @@ -69,6 +75,7 @@ namespace Emby.Server.Implementations.Session private Timer _idleTimer; private Timer _inactiveTimer; + private Timer _directoryTimer; private DtoOptions _itemInfoDtoOptions; private bool _disposed; @@ -89,6 +96,9 @@ namespace Emby.Server.Implementations.Session /// Instance of interface. /// Instance of interface. /// Instance of interface. + /// Instance of interface. + /// Instance of interface. + /// The session directory options. public SessionManager( ILogger logger, IEventManager eventManager, @@ -102,7 +112,10 @@ namespace Emby.Server.Implementations.Session IServerApplicationHost appHost, IDeviceManager deviceManager, IMediaSourceManager mediaSourceManager, - IHostApplicationLifetime hostApplicationLifetime) + IHostApplicationLifetime hostApplicationLifetime, + ISessionDirectory sessionDirectory, + IPodMessageBus podMessageBus, + IOptions sessionDirectoryOptions) { _logger = logger; _eventManager = eventManager; @@ -116,9 +129,19 @@ 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; + + if (_sessionDirectory is not NullSessionDirectory) + { + _podMessageBus.Subscribe(OnPodMessage); + var interval = TimeSpan.FromSeconds(Math.Max(1, _sessionDirectoryOptions.RefreshIntervalSeconds)); + _directoryTimer = new Timer(RefreshSessionDirectory, null, interval, interval); + } } /// @@ -218,6 +241,8 @@ namespace Emby.Server.Implementations.Session _eventManager.Publish(new SessionEndedEventArgs(info)); + await RemoveFromDirectoryAsync(info).ConfigureAwait(false); + await info.DisposeAsync().ConfigureAwait(false); } @@ -288,6 +313,8 @@ namespace Emby.Server.Implementations.Session }); } + await PublishToDirectoryAsync(session).ConfigureAwait(false); + return session; } @@ -302,6 +329,134 @@ namespace Emby.Server.Implementations.Session SessionInfo = session }, _logger); + + // The session only becomes remote-controllable once it has a connection, so the other + // instances have to be told again now that it does. + _ = PublishToDirectoryAsync(session); + } + + private bool DirectoryEnabled => _sessionDirectory is not NullSessionDirectory; + + private async Task PublishToDirectoryAsync(SessionInfo session) + { + if (!DirectoryEnabled || string.IsNullOrEmpty(session.Id)) + { + return; + } + + try + { + await _sessionDirectory.PublishAsync( + new SessionDirectoryEntry + { + OwnerPod = _podMessageBus.PodId, + Session = ToSessionInfoDto(session) + }).ConfigureAwait(false); + } + catch (Exception ex) + { + _logger.LogDebug(ex, "Error publishing session {Session} to the directory.", session.Id); + } + } + + private async ValueTask RemoveFromDirectoryAsync(SessionInfo session) + { + if (!DirectoryEnabled || string.IsNullOrEmpty(session.Id)) + { + return; + } + + await _sessionDirectory.RemoveAsync(session.Id).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> GetRemoteEntriesAsync(CancellationToken cancellationToken) + { + if (!DirectoryEnabled) + { + return Array.Empty(); + } + + var entries = await _sessionDirectory.GetAllAsync(cancellationToken).ConfigureAwait(false); + + return entries + .Where(entry => entry.Session is not null + && !string.Equals(entry.OwnerPod, _podMessageBus.PodId, StringComparison.Ordinal)) + .ToList(); + } + + private async Task GetRemoteSession(string sessionId) + { + var entries = await GetRemoteEntriesAsync(CancellationToken.None).ConfigureAwait(false); + var entry = entries.FirstOrDefault(i => string.Equals(i.Session.Id, sessionId, StringComparison.Ordinal)); + + if (entry is null) + { + return null; + } + + var dto = entry.Session; + var session = new SessionInfo(this, _logger) + { + Id = dto.Id, + UserId = dto.UserId, + UserName = dto.UserName, + Client = dto.Client, + DeviceId = dto.DeviceId, + DeviceName = dto.DeviceName, + DeviceType = dto.DeviceType, + ApplicationVersion = dto.ApplicationVersion, + RemoteEndPoint = dto.RemoteEndPoint, + LastActivityDate = dto.LastActivityDate, + ServerId = dto.ServerId, + AdditionalUsers = dto.AdditionalUsers ?? [], + Capabilities = dto.Capabilities?.ToClientCapabilities() + }; + + session.AddController(new RemoteSessionController(_podMessageBus, entry.OwnerPod, dto.Id, dto.IsActive, dto.SupportsMediaControl)); + + return session; + } + + private async Task OnPodMessage(PodMessage message) + { + if (!string.Equals(message.Kind, RoutedSessionMessage.Kind, StringComparison.Ordinal)) + { + return; + } + + var routed = JsonSerializer.Deserialize(message.Payload, JsonDefaults.Options); + if (routed is null) + { + return; + } + + var session = Sessions.FirstOrDefault(i => string.Equals(i.Id, routed.SessionId, StringComparison.Ordinal)); + if (session is null) + { + _logger.LogDebug("Session {Session} was routed here but is no longer held by this instance.", routed.SessionId); + return; + } + + using var data = JsonDocument.Parse(routed.Data); + foreach (var controller in session.SessionControllers) + { + await controller.SendMessage(routed.MessageType, routed.MessageId, data.RootElement, CancellationToken.None).ConfigureAwait(false); + } } /// @@ -1219,10 +1374,12 @@ namespace Emby.Server.Implementations.Session return session; } - private SessionInfo GetSessionToRemoteControl(string sessionId) + // A session held by another instance is reachable too: the returned SessionInfo carries a + // controller that forwards to its owner instead of writing to a local connection. + private async Task GetSessionToRemoteControl(string sessionId) { - // Accept either device id or session id - var session = Sessions.FirstOrDefault(i => string.Equals(i.Id, sessionId, StringComparison.Ordinal)); + var session = Sessions.FirstOrDefault(i => string.Equals(i.Id, sessionId, StringComparison.Ordinal)) + ?? await GetRemoteSession(sessionId).ConfigureAwait(false); if (session is null) { @@ -1291,19 +1448,19 @@ namespace Emby.Server.Implementations.Session } /// - public Task SendGeneralCommand(string controllingSessionId, string sessionId, GeneralCommand command, CancellationToken cancellationToken) + public async Task SendGeneralCommand(string controllingSessionId, string sessionId, GeneralCommand command, CancellationToken cancellationToken) { CheckDisposed(); - var session = GetSessionToRemoteControl(sessionId); + var session = await GetSessionToRemoteControl(sessionId).ConfigureAwait(false); if (!string.IsNullOrEmpty(controllingSessionId)) { - var controllingSession = GetSession(controllingSessionId); + var controllingSession = await 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(SessionInfo session, SessionMessageType name, T data, CancellationToken cancellationToken) @@ -1340,7 +1497,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 +1567,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()) { @@ -1521,15 +1678,15 @@ namespace Emby.Server.Implementations.Session } /// - public Task SendPlaystateCommand(string controllingSessionId, string sessionId, PlaystateRequest command, CancellationToken cancellationToken) + public async Task SendPlaystateCommand(string controllingSessionId, string sessionId, PlaystateRequest command, CancellationToken cancellationToken) { CheckDisposed(); - var session = GetSessionToRemoteControl(sessionId); + var session = await GetSessionToRemoteControl(sessionId).ConfigureAwait(false); if (!string.IsNullOrEmpty(controllingSessionId)) { - var controllingSession = GetSession(controllingSessionId); + var controllingSession = await GetSessionToRemoteControl(controllingSessionId).ConfigureAwait(false); AssertCanControl(session, controllingSession); if (!controllingSession.UserId.IsEmpty()) { @@ -1537,7 +1694,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) @@ -2060,14 +2217,21 @@ namespace Emby.Server.Implementations.Session } /// - public IReadOnlyList GetSessions( + public async Task> GetSessions( Guid userId, string deviceId, int? activeWithinSeconds, Guid? controllableUserToCheck, - bool isApiKey) + bool isApiKey, + CancellationToken cancellationToken) { - var result = Sessions; + var remote = await GetRemoteEntriesAsync(cancellationToken).ConfigureAwait(false); + + IEnumerable result = Sessions + .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 +2279,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 +2300,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 +2323,18 @@ namespace Emby.Server.Implementations.Session result = result.Where(i => i.LastActivityDate >= minActiveDate); } - return result.Select(ToSessionInfoDto).ToList(); + return result.ToList(); + } + + private static bool ContainsUser(SessionInfoDto session, Guid userId) + { + if (session.UserId.Equals(userId)) + { + return true; + } + + return session.AdditionalUsers is not null + && session.AdditionalUsers.Any(i => i.UserId.Equals(userId)); } /// @@ -2234,6 +2409,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 +2438,7 @@ namespace Emby.Server.Implementations.Session // Close open websockets to allow Kestrel to shut down cleanly foreach (var session in _activeConnections.Values) { + await RemoveFromDirectoryAsync(session).ConfigureAwait(false); await session.DisposeAsync().ConfigureAwait(false); } diff --git a/Jellyfin.Api/Controllers/SessionController.cs b/Jellyfin.Api/Controllers/SessionController.cs index 84c2d90fb1..7d6a162d8f 100644 --- a/Jellyfin.Api/Controllers/SessionController.cs +++ b/Jellyfin.Api/Controllers/SessionController.cs @@ -52,18 +52,19 @@ public class SessionController : BaseJellyfinApiController [HttpGet("Sessions")] [Authorize] [ProducesResponseType(StatusCodes.Status200OK)] - public ActionResult> GetSessions( + public async Task>> GetSessions( [FromQuery] Guid? controllableByUserId, [FromQuery] string? deviceId, [FromQuery] int? activeWithinSeconds) { Guid? controllableUserToCheck = controllableByUserId is null ? null : RequestHelpers.GetUserId(User, controllableByUserId); - var result = _sessionManager.GetSessions( + var result = await _sessionManager.GetSessions( User.GetUserId(), deviceId, activeWithinSeconds, controllableUserToCheck, - User.GetIsApiKey()); + User.GetIsApiKey(), + HttpContext.RequestAborted).ConfigureAwait(false); return Ok(result); } diff --git a/Jellyfin.Server/CoreAppHost.cs b/Jellyfin.Server/CoreAppHost.cs index 9ac86bb8ed..ece3cd971d 100644 --- a/Jellyfin.Server/CoreAppHost.cs +++ b/Jellyfin.Server/CoreAppHost.cs @@ -116,6 +116,10 @@ namespace Jellyfin.Server // to the other instances. Redis-backed when configured, no-op otherwise. serviceCollection.AddConfigurationInvalidationBus(_startupConfig, Logger); + // Session directory: publishes which instance holds which session and routes remote-control + // messages to it. Redis-backed when configured, no-op otherwise. + serviceCollection.AddSessionDirectory(_startupConfig, Logger); + foreach (var type in GetExportTypes()) { serviceCollection.AddSingleton(typeof(ILyricProvider), type); diff --git a/Jellyfin.Server/Extensions/SessionDirectoryServiceCollectionExtensions.cs b/Jellyfin.Server/Extensions/SessionDirectoryServiceCollectionExtensions.cs new file mode 100644 index 0000000000..33977ef9a4 --- /dev/null +++ b/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; + +/// +/// Extensions for registering the session directory and the instance-addressed message bus. +/// +public static class SessionDirectoryServiceCollectionExtensions +{ + /// + /// Registers the session directory and message bus, Redis-backed when a connection string is + /// configured and no-op otherwise, and reports the selection at . + /// + /// The service collection. + /// The configuration to read Jellyfin:SessionDirectory from. + /// The logger to report the selection on. + /// The updated service collection. + public static IServiceCollection AddSessionDirectory( + this IServiceCollection serviceCollection, + IConfiguration configuration, + ILogger logger) + { + ArgumentNullException.ThrowIfNull(configuration); + ArgumentNullException.ThrowIfNull(logger); + + serviceCollection.Configure(configuration.GetSection(SessionDirectoryOptions.ConfigurationSection)); + + if (string.IsNullOrEmpty(configuration[TranscodeStoreOptions.RedisConnectionStringKey])) + { + logger.LogInformation( + "Session directory: {Directory}. The session list and remote control only reach the sessions this instance holds; set {Key} to share them.", + nameof(NullSessionDirectory), + TranscodeStoreOptions.RedisConnectionStringKey); + + serviceCollection.AddSingleton(NullSessionDirectory.Instance); + return serviceCollection.AddSingleton(NullPodMessageBus.Instance); + } + + logger.LogInformation( + "Session directory: {Directory}. Sessions are visible to, and controllable from, every instance.", + nameof(RedisSessionDirectory)); + + serviceCollection.AddSingleton(sp => Create( + sp, + () => new RedisPodMessageBus( + sp.GetRequiredService(), + sp.GetRequiredService>()), + NullPodMessageBus.Instance)); + + return serviceCollection.AddSingleton(sp => Create( + sp, + () => new RedisSessionDirectory( + sp.GetRequiredService(), + sp.GetRequiredService>(), + sp.GetRequiredService>()), + NullSessionDirectory.Instance)); + } + + // Fail open: an unreachable Redis degrades to the single-instance behaviour rather than aborting startup. + private static T Create(IServiceProvider serviceProvider, Func factory, T fallback) + { + try + { + return factory(); + } + catch (Exception ex) + { + serviceProvider.GetRequiredService>().LogError( + ex, + "Redis is configured but unavailable, so sessions will not be shared between instances. Check {Key}.", + TranscodeStoreOptions.RedisConnectionStringKey); + + return fallback; + } + } +} diff --git a/MediaBrowser.Controller/Session/IPodMessageBus.cs b/MediaBrowser.Controller/Session/IPodMessageBus.cs new file mode 100644 index 0000000000..053299e6c2 --- /dev/null +++ b/MediaBrowser.Controller/Session/IPodMessageBus.cs @@ -0,0 +1,29 @@ +using System; +using System.Threading.Tasks; + +namespace MediaBrowser.Controller.Session; + +/// +/// Point-to-point delivery between instances: every instance listens on a channel of its own, so a +/// message can be addressed to the one instance holding a given connection. +/// +public interface IPodMessageBus +{ + /// + /// Gets the identity of this instance. + /// + string PodId { get; } + + /// + /// Sends a message to one instance. Delivery is best effort and never throws. + /// + /// The instance to deliver to. + /// The message. + void Publish(string targetPod, PodMessage message); + + /// + /// Registers a handler for the messages addressed to this instance. + /// + /// The handler. + void Subscribe(Func handler); +} diff --git a/MediaBrowser.Controller/Session/ISessionDirectory.cs b/MediaBrowser.Controller/Session/ISessionDirectory.cs new file mode 100644 index 0000000000..f1752f9f63 --- /dev/null +++ b/MediaBrowser.Controller/Session/ISessionDirectory.cs @@ -0,0 +1,35 @@ +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; + +namespace MediaBrowser.Controller.Session; + +/// +/// The shared record of which instance holds which session. Entries expire, so an instance that stops +/// refreshing them drops out of every other instance's view instead of lingering. +/// +public interface ISessionDirectory +{ + /// + /// Publishes an entry and restarts its expiry. + /// + /// The entry. + /// The cancellation token. + /// A task representing the operation. + Task PublishAsync(SessionDirectoryEntry entry, CancellationToken cancellationToken = default); + + /// + /// Removes an entry. + /// + /// The session identifier. + /// The cancellation token. + /// A task representing the operation. + Task RemoveAsync(string sessionId, CancellationToken cancellationToken = default); + + /// + /// Gets every entry that has not expired. + /// + /// The cancellation token. + /// The entries. + Task> GetAllAsync(CancellationToken cancellationToken = default); +} diff --git a/MediaBrowser.Controller/Session/ISessionManager.cs b/MediaBrowser.Controller/Session/ISessionManager.cs index 9acff745b9..58bbd72876 100644 --- a/MediaBrowser.Controller/Session/ISessionManager.cs +++ b/MediaBrowser.Controller/Session/ISessionManager.cs @@ -306,8 +306,9 @@ namespace MediaBrowser.Controller.Session /// Active within session limit. /// Filter for sessions remote controllable for this user. /// Is the request authenticated with API key. - /// IReadOnlyList{SessionInfoDto}. - IReadOnlyList GetSessions(Guid userId, string deviceId, int? activeWithinSeconds, Guid? controllableUserToCheck, bool isApiKey); + /// The cancellation token. + /// IReadOnlyList{SessionInfoDto}, including the sessions held by the other instances. + Task> GetSessions(Guid userId, string deviceId, int? activeWithinSeconds, Guid? controllableUserToCheck, bool isApiKey, CancellationToken cancellationToken); /// /// Gets the session by authentication token. diff --git a/MediaBrowser.Controller/Session/NullPodMessageBus.cs b/MediaBrowser.Controller/Session/NullPodMessageBus.cs new file mode 100644 index 0000000000..e34d6d9233 --- /dev/null +++ b/MediaBrowser.Controller/Session/NullPodMessageBus.cs @@ -0,0 +1,28 @@ +using System; +using System.Threading.Tasks; + +namespace MediaBrowser.Controller.Session; + +/// +/// The single-instance : there is no other instance to reach. +/// +public sealed class NullPodMessageBus : IPodMessageBus +{ + /// + /// Gets the shared instance. + /// + public static NullPodMessageBus Instance { get; } = new NullPodMessageBus(); + + /// + public string PodId => PodIdentity.Current; + + /// + public void Publish(string targetPod, PodMessage message) + { + } + + /// + public void Subscribe(Func handler) + { + } +} diff --git a/MediaBrowser.Controller/Session/NullSessionDirectory.cs b/MediaBrowser.Controller/Session/NullSessionDirectory.cs new file mode 100644 index 0000000000..3fcd3bed36 --- /dev/null +++ b/MediaBrowser.Controller/Session/NullSessionDirectory.cs @@ -0,0 +1,30 @@ +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; + +namespace MediaBrowser.Controller.Session; + +/// +/// The single-instance : nothing is published and no session is held +/// anywhere but here, which is exactly the behaviour of a deployment without a shared store. +/// +public sealed class NullSessionDirectory : ISessionDirectory +{ + /// + /// Gets the shared instance. + /// + public static NullSessionDirectory Instance { get; } = new NullSessionDirectory(); + + /// + public Task PublishAsync(SessionDirectoryEntry entry, CancellationToken cancellationToken = default) + => Task.CompletedTask; + + /// + public Task RemoveAsync(string sessionId, CancellationToken cancellationToken = default) + => Task.CompletedTask; + + /// + public Task> GetAllAsync(CancellationToken cancellationToken = default) + => Task.FromResult>(Array.Empty()); +} diff --git a/MediaBrowser.Controller/Session/PodIdentity.cs b/MediaBrowser.Controller/Session/PodIdentity.cs new file mode 100644 index 0000000000..b042897709 --- /dev/null +++ b/MediaBrowser.Controller/Session/PodIdentity.cs @@ -0,0 +1,14 @@ +using System; + +namespace MediaBrowser.Controller.Session; + +/// +/// The identity of this instance among the replicas sharing a deployment. +/// +public static class PodIdentity +{ + /// + /// Gets the identity of this instance. + /// + public static string Current => Environment.GetEnvironmentVariable("JELLYFIN_INSTANCE_ID") ?? Environment.MachineName; +} diff --git a/MediaBrowser.Controller/Session/PodMessage.cs b/MediaBrowser.Controller/Session/PodMessage.cs new file mode 100644 index 0000000000..1912b40300 --- /dev/null +++ b/MediaBrowser.Controller/Session/PodMessage.cs @@ -0,0 +1,23 @@ +namespace MediaBrowser.Controller.Session; + +/// +/// An envelope addressed to one instance. names the payload so that features other +/// than session routing can share the same channel. +/// +public sealed class PodMessage +{ + /// + /// Gets or sets the payload discriminator. + /// + public string Kind { get; set; } = string.Empty; + + /// + /// Gets or sets the identity of the sending instance. + /// + public string OriginPod { get; set; } = string.Empty; + + /// + /// Gets or sets the serialized payload. + /// + public string Payload { get; set; } = string.Empty; +} diff --git a/MediaBrowser.Controller/Session/RoutedSessionMessage.cs b/MediaBrowser.Controller/Session/RoutedSessionMessage.cs new file mode 100644 index 0000000000..a8ae8bb620 --- /dev/null +++ b/MediaBrowser.Controller/Session/RoutedSessionMessage.cs @@ -0,0 +1,35 @@ +using System; +using MediaBrowser.Model.Session; + +namespace MediaBrowser.Controller.Session; + +/// +/// A websocket message for a session held by another instance, carried as a . +/// +public sealed class RoutedSessionMessage +{ + /// + /// The this payload travels under. + /// + public const string Kind = "SessionMessage"; + + /// + /// Gets or sets the session the message is addressed to. + /// + public string SessionId { get; set; } = string.Empty; + + /// + /// Gets or sets the message type. + /// + public SessionMessageType MessageType { get; set; } + + /// + /// Gets or sets the message identifier. + /// + public Guid MessageId { get; set; } + + /// + /// Gets or sets the message data, serialized as JSON. + /// + public string Data { get; set; } = "null"; +} diff --git a/MediaBrowser.Controller/Session/SessionDirectoryEntry.cs b/MediaBrowser.Controller/Session/SessionDirectoryEntry.cs new file mode 100644 index 0000000000..3f8aaa12bb --- /dev/null +++ b/MediaBrowser.Controller/Session/SessionDirectoryEntry.cs @@ -0,0 +1,19 @@ +using MediaBrowser.Model.Dto; + +namespace MediaBrowser.Controller.Session; + +/// +/// A session held by one instance, as the other instances see it. +/// +public sealed class SessionDirectoryEntry +{ + /// + /// Gets or sets the identity of the instance holding the connection. + /// + public string OwnerPod { get; set; } = string.Empty; + + /// + /// Gets or sets the session as its owner last rendered it. + /// + public SessionInfoDto? Session { get; set; } +} diff --git a/MediaBrowser.Controller/Session/SessionDirectoryOptions.cs b/MediaBrowser.Controller/Session/SessionDirectoryOptions.cs new file mode 100644 index 0000000000..17ff304c2a --- /dev/null +++ b/MediaBrowser.Controller/Session/SessionDirectoryOptions.cs @@ -0,0 +1,23 @@ +namespace MediaBrowser.Controller.Session; + +/// +/// Configuration options for the session directory and the cross-instance bus that goes with it. +/// +public sealed class SessionDirectoryOptions +{ + /// + /// The configuration section these options bind from. + /// + public const string ConfigurationSection = "Jellyfin:SessionDirectory"; + + /// + /// Gets or sets how long in seconds a published entry survives without being refreshed. An instance + /// that dies stops refreshing, so its sessions leave the directory after this long. + /// + public int EntryTtlSeconds { get; set; } = 60; + + /// + /// Gets or sets how often in seconds an instance republishes the sessions it holds. + /// + public int RefreshIntervalSeconds { get; set; } = 20; +} diff --git a/README.md b/README.md index 7ac2845e7d..daffb95682 100644 --- a/README.md +++ b/README.md @@ -116,6 +116,8 @@ 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. | ### Redis connection string examples diff --git a/tests/Jellyfin.Server.Implementations.Tests/SessionManager/IdlePlaybackTests.cs b/tests/Jellyfin.Server.Implementations.Tests/SessionManager/IdlePlaybackTests.cs index 7722707cbe..b6655ec995 100644 --- a/tests/Jellyfin.Server.Implementations.Tests/SessionManager/IdlePlaybackTests.cs +++ b/tests/Jellyfin.Server.Implementations.Tests/SessionManager/IdlePlaybackTests.cs @@ -13,6 +13,7 @@ using MediaBrowser.Model.Dto; using MediaBrowser.Model.Session; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; using Moq; using Xunit; @@ -44,7 +45,10 @@ public class IdlePlaybackTests Mock.Of(), Mock.Of(), Mock.Of(), - Mock.Of()); + Mock.Of(), + NullSessionDirectory.Instance, + NullPodMessageBus.Instance, + Options.Create(new SessionDirectoryOptions())); var session = await sessionManager.LogSessionActivity( "Test Client", "1.0.0", diff --git a/tests/Jellyfin.Server.Implementations.Tests/SessionManager/SessionManagerTests.cs b/tests/Jellyfin.Server.Implementations.Tests/SessionManager/SessionManagerTests.cs index f803c69af2..d032539390 100644 --- a/tests/Jellyfin.Server.Implementations.Tests/SessionManager/SessionManagerTests.cs +++ b/tests/Jellyfin.Server.Implementations.Tests/SessionManager/SessionManagerTests.cs @@ -16,6 +16,7 @@ using MediaBrowser.Controller.Session; using MediaBrowser.Model.Session; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; using Moq; using Xunit; @@ -41,7 +42,10 @@ public class SessionManagerTests Mock.Of(), Mock.Of(), Mock.Of(), - Mock.Of()); + Mock.Of(), + NullSessionDirectory.Instance, + NullPodMessageBus.Instance, + Options.Create(new SessionDirectoryOptions())); await Assert.ThrowsAsync(exceptionType, () => sessionManager.GetAuthorizationToken( new User("test", "default", "default"), @@ -68,7 +72,10 @@ public class SessionManagerTests Mock.Of(), Mock.Of(), Mock.Of(), - Mock.Of()); + Mock.Of(), + NullSessionDirectory.Instance, + NullPodMessageBus.Instance, + Options.Create(new SessionDirectoryOptions())); await Assert.ThrowsAsync(exceptionType, () => sessionManager.AuthenticateNewSessionInternal(authenticationRequest, false)); } @@ -238,7 +245,10 @@ public class SessionManagerTests Mock.Of(), Mock.Of(), Mock.Of(), - Mock.Of()); + Mock.Of(), + NullSessionDirectory.Instance, + NullPodMessageBus.Instance, + Options.Create(new SessionDirectoryOptions())); } // All sessions are logged with the same client and device id on purpose, those values are taken diff --git a/tests/Jellyfin.Server.Tests/HighAvailability/RedisTestServer.cs b/tests/Jellyfin.Server.Tests/HighAvailability/RedisTestServer.cs new file mode 100644 index 0000000000..b839065b40 --- /dev/null +++ b/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; + +/// +/// Hands out a valkey/Redis server for the tests that need one. A server named by +/// JELLYFIN_TEST_REDIS is used as is, so CI can run one in the step instead of a docker daemon +/// of its own; without it a container is started through testcontainers. +/// +public sealed class RedisTestServer : IAsyncDisposable +{ + /// + /// The connection string of an already running server. + /// + public const string ConnectionStringVariable = "JELLYFIN_TEST_REDIS"; + + private readonly RedisContainer? _container; + + private RedisTestServer(RedisContainer? container, string connectionString) + { + _container = container; + ConnectionString = connectionString; + } + + /// + /// Gets the connection string of the running server. + /// + public string ConnectionString { get; } + + /// + /// Starts or attaches to a server and connects to it. + /// + /// The running server. + public static async Task 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()); + } + + /// + /// Opens a connection to the server. + /// + /// The connection. + public async Task ConnectAsync() + => await ConnectionMultiplexer.ConnectAsync(ConnectionString).ConfigureAwait(false); + + /// + public async ValueTask DisposeAsync() + { + if (_container is not null) + { + await _container.DisposeAsync().ConfigureAwait(false); + } + } +} diff --git a/tests/Jellyfin.Server.Tests/HighAvailability/SessionDirectoryReplicaTests.cs b/tests/Jellyfin.Server.Tests/HighAvailability/SessionDirectoryReplicaTests.cs new file mode 100644 index 0000000000..4d5f3f8464 --- /dev/null +++ b/tests/Jellyfin.Server.Tests/HighAvailability/SessionDirectoryReplicaTests.cs @@ -0,0 +1,282 @@ +using System; +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.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; + +/// +/// Two independently constructed 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. +/// +[Trait("Category", "RequiresDocker")] +public sealed class SessionDirectoryReplicaTests : IAsyncLifetime +{ + private PostgreSqlTestServer _postgres = null!; + private RedisTestServer _redis = null!; + private NpgsqlDataSource _dataSource = null!; + private IConnectionMultiplexer _connection = null!; + private User _user = null!; + + /// + public async ValueTask InitializeAsync() + { + _postgres = await PostgreSqlTestServer.StartAsync(); + _redis = await RedisTestServer.StartAsync(); + _connection = await _redis.ConnectAsync(); + + 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"); + context.Users.Add(_user); + await context.SaveChangesAsync(CancellationToken.None); + } + } + + /// + public async ValueTask DisposeAsync() + { + await _connection.DisposeAsync(); + await _dataSource.DisposeAsync(); + await _redis.DisposeAsync(); + await _postgres.DisposeAsync(); + } + + /// + /// 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. + /// + /// A representing the asynchronous operation. + [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 replicaA.LogSessionActivity("Jellyfin Web", "1.0.0", "device-listed", "Living Room TV", "127.0.0.1", _user); + + 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)); + } + + /// + /// Remote control and "send message to session" currently succeed and do nothing when the device is + /// connected to another replica; the message has to reach the connection wherever it is held. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task MessageSentOnOneReplica_ReachesTheConnectionHeldByAnother() + { + var cancellationToken = TestContext.Current.CancellationToken; + await using var replicaA = CreateReplica("pod-a"); + await using var replicaB = CreateReplica("pod-b"); + + var controller = new RecordingSessionController(); + var session = await replicaA.LogSessionActivity("Jellyfin Web", "1.0.0", "device-controlled", "Living Room TV", "127.0.0.1", _user); + session.AddController(controller); + + // Republish now that the session has a connection, the way the websocket handshake does. + await replicaA.LogSessionActivity("Jellyfin Web", "1.0.0", "device-controlled", "Living Room TV", "127.0.0.1", _user); + + 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); + } + + /// + /// 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. + /// + /// A representing the asynchronous operation. + [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 replicaA.LogSessionActivity("Jellyfin Web", "1.0.0", "device-expiring", "Living Room TV", "127.0.0.1", _user); + + 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)); + } + + /// + /// A deployment without a shared store keeps the single-instance behaviour: nothing is published and + /// the other instance sees nothing. + /// + /// A representing the asynchronous operation. + [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 replicaA.LogSessionActivity("Jellyfin Web", "1.0.0", "device-local", "Living Room TV", "127.0.0.1", _user); + + 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 SessionManager CreateReplica(string podId, SessionDirectoryOptions? options = null) + { + options ??= new SessionDirectoryOptions { EntryTtlSeconds = 60, RefreshIntervalSeconds = 3600 }; + + var directory = new RedisSessionDirectory( + _connection, + Options.Create(options), + NullLogger.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(); + userManager.Setup(i => i.GetUserById(_user.Id)).Returns(_user); + + var appHost = new Mock(); + appHost.SetupGet(i => i.SystemId).Returns("server-" + podId); + + return new SessionManager( + NullLogger.Instance, + Mock.Of(), + Mock.Of(), + Mock.Of(), + Mock.Of(), + userManager.Object, + Mock.Of(), + Mock.Of(), + Mock.Of(), + appHost.Object, + new DeviceManager(new DataSourceContextFactory(_dataSource), userManager.Object), + Mock.Of(), + Mock.Of(), + 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.Instance); + } + finally + { + Environment.SetEnvironmentVariable("JELLYFIN_INSTANCE_ID", previous); + } + } + + private static JellyfinDbContext CreateContext(NpgsqlDataSource dataSource) + { + var optionsBuilder = new DbContextOptionsBuilder(); + var provider = new PostgreSqlDatabaseProvider(dataSource); + provider.Initialise(optionsBuilder, new DatabaseConfigurationOptions { DatabaseType = "PostgreSQL" }); + return new JellyfinDbContext( + optionsBuilder.Options, + NullLogger.Instance, + provider, + new NoLockBehavior(NullLogger.Instance)); + } + + /// + /// Hands every replica its own context over the one shared database, the way the pooled factory does + /// in the server. + /// + private sealed class DataSourceContextFactory : IDbContextFactory + { + private readonly NpgsqlDataSource _dataSource; + + public DataSourceContextFactory(NpgsqlDataSource dataSource) + { + _dataSource = dataSource; + } + + public JellyfinDbContext CreateDbContext() => CreateContext(_dataSource); + } + + /// + /// Stands in for the websocket the owning replica holds. + /// + 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(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); + } +} diff --git a/tests/Jellyfin.Server.Tests/Jellyfin.Server.Tests.csproj b/tests/Jellyfin.Server.Tests/Jellyfin.Server.Tests.csproj index a96840baa1..a2adf471c5 100644 --- a/tests/Jellyfin.Server.Tests/Jellyfin.Server.Tests.csproj +++ b/tests/Jellyfin.Server.Tests/Jellyfin.Server.Tests.csproj @@ -11,7 +11,9 @@ + + all