diff --git a/Emby.Server.Implementations/Session/RedisPodMessageBus.cs b/Emby.Server.Implementations/Session/RedisPodMessageBus.cs index 955be0454d..3db25ab013 100644 --- a/Emby.Server.Implementations/Session/RedisPodMessageBus.cs +++ b/Emby.Server.Implementations/Session/RedisPodMessageBus.cs @@ -1,10 +1,12 @@ using System; +using System.Collections.Concurrent; using System.Text.Json; using System.Threading; using System.Threading.Tasks; using Jellyfin.Extensions.Json; using MediaBrowser.Controller.Session; using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; using StackExchange.Redis; namespace Emby.Server.Implementations.Session; @@ -12,64 +14,47 @@ namespace Emby.Server.Implementations.Session; /// /// A Redis pub/sub . Every instance subscribes to a channel named after /// itself, which keeps addressed delivery working without the instances being routable to each other. +/// A request is answered on the sender's own channel, so the sender learns what the receiver did with +/// it rather than only that something was subscribed. /// public sealed class RedisPodMessageBus : IPodMessageBus { private const string ChannelPrefix = "jellyfin:pod:"; - private static readonly TimeSpan _publishTimeout = TimeSpan.FromSeconds(5); - private static readonly JsonSerializerOptions _jsonOptions = JsonDefaults.Options; + private readonly ConcurrentDictionary> _pending = new(StringComparer.Ordinal); private readonly ISubscriber _subscriber; private readonly ILogger _logger; + private readonly TimeSpan _timeout; + + private Func>? _handler; /// /// Initializes a new instance of the class. /// /// The Redis connection multiplexer. + /// The session directory configuration options. + /// The identity of this instance. /// The logger. - public RedisPodMessageBus(IConnectionMultiplexer redis, ILogger logger) + public RedisPodMessageBus( + IConnectionMultiplexer redis, + IOptions options, + string podId, + ILogger logger) { ArgumentNullException.ThrowIfNull(redis); + ArgumentNullException.ThrowIfNull(options); + ArgumentException.ThrowIfNullOrEmpty(podId); _subscriber = redis.GetSubscriber(); _logger = logger; - PodId = PodIdentity.Current; - } - - /// - public string PodId { get; } - - /// - public async Task PublishAsync(string targetPod, PodMessage message, CancellationToken cancellationToken = default) - { - ArgumentException.ThrowIfNullOrEmpty(targetPod); - ArgumentNullException.ThrowIfNull(message); - - message.OriginPod = PodId; + _timeout = TimeSpan.FromSeconds(Math.Max(1, options.Value.OperationTimeoutSeconds)); + PodId = 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; - } - } - - /// - public void Subscribe(Func handler) - { - ArgumentNullException.ThrowIfNull(handler); - - try - { - _subscriber.Subscribe(RedisChannel.Literal(ChannelPrefix + PodId), (_, value) => Dispatch(handler, value)); + _subscriber.Subscribe(RedisChannel.Literal(ChannelPrefix + PodId), (_, value) => Dispatch(value)); } catch (Exception ex) { @@ -77,19 +62,120 @@ public sealed class RedisPodMessageBus : IPodMessageBus } } - private async void Dispatch(Func handler, RedisValue value) + /// + public string PodId { get; } + + /// + public async Task RequestAsync(string targetPod, PodMessage message, CancellationToken cancellationToken = default) { + ArgumentException.ThrowIfNullOrEmpty(targetPod); + ArgumentNullException.ThrowIfNull(message); + + message.OriginPod = PodId; + message.CorrelationId = Guid.NewGuid().ToString("N"); + + var acknowledged = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _pending[message.CorrelationId] = acknowledged; + try { - var message = JsonSerializer.Deserialize(value.ToString(), _jsonOptions); - if (message is not null) + var subscribers = await _subscriber.PublishAsync( + RedisChannel.Literal(ChannelPrefix + targetPod), + JsonSerializer.Serialize(message, _jsonOptions)).WaitAsync(_timeout, cancellationToken).ConfigureAwait(false); + + if (subscribers == 0) { - await handler(message).ConfigureAwait(false); + return false; } + + return await acknowledged.Task.WaitAsync(_timeout, cancellationToken).ConfigureAwait(false); + } + catch (TimeoutException) + { + _logger.LogWarning("Instance {TargetPod} did not acknowledge a {Kind} message within {Timeout}.", targetPod, message.Kind, _timeout); + return false; + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + _logger.LogWarning(ex, "Failed to send a {Kind} message to {TargetPod}.", message.Kind, targetPod); + return false; + } + finally + { + _pending.TryRemove(message.CorrelationId, out _); + } + } + + /// + public void Subscribe(Func> handler) + { + ArgumentNullException.ThrowIfNull(handler); + + _handler = handler; + } + + private async void Dispatch(RedisValue value) + { + PodMessage? message = null; + + try + { + message = JsonSerializer.Deserialize(value.ToString(), _jsonOptions); + if (message is null) + { + return; + } + + if (string.Equals(message.Kind, PodMessage.AckKind, StringComparison.Ordinal)) + { + if (_pending.TryRemove(message.CorrelationId, out var acknowledged)) + { + acknowledged.TrySetResult(message.Handled); + } + + return; + } + + var handler = _handler; + var handled = handler is not null && await handler(message).ConfigureAwait(false); + + await AcknowledgeAsync(message, handled).ConfigureAwait(false); } catch (Exception ex) { _logger.LogWarning(ex, "Failed to handle a message routed to this instance."); + + if (message is not null && !string.Equals(message.Kind, PodMessage.AckKind, StringComparison.Ordinal)) + { + await AcknowledgeAsync(message, false).ConfigureAwait(false); + } + } + } + + private async Task AcknowledgeAsync(PodMessage message, bool handled) + { + if (string.IsNullOrEmpty(message.CorrelationId) || string.IsNullOrEmpty(message.OriginPod)) + { + return; + } + + var ack = new PodMessage + { + Kind = PodMessage.AckKind, + OriginPod = PodId, + CorrelationId = message.CorrelationId, + Handled = handled + }; + + try + { + await _subscriber.PublishAsync( + RedisChannel.Literal(ChannelPrefix + message.OriginPod), + JsonSerializer.Serialize(ack, _jsonOptions)).WaitAsync(_timeout).ConfigureAwait(false); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Failed to acknowledge a {Kind} message to {OriginPod}.", message.Kind, message.OriginPod); } } } diff --git a/Emby.Server.Implementations/Session/RedisSessionDirectory.cs b/Emby.Server.Implementations/Session/RedisSessionDirectory.cs index e31d1ddb9a..b6953d1a49 100644 --- a/Emby.Server.Implementations/Session/RedisSessionDirectory.cs +++ b/Emby.Server.Implementations/Session/RedisSessionDirectory.cs @@ -15,28 +15,32 @@ namespace Emby.Server.Implementations.Session; /// /// A Redis-backed . A session is owned by the instance holding its -/// connection: ownership is claimed through a check-and-set, 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. +/// connection: ownership is claimed through a check-and-set against a connection epoch handed out by +/// Redis, so an instance that only served a request for the session cannot take it from the instance the +/// device is actually connected to, and no instance's clock is compared against another's. Each entry is +/// a key with an expiry, so the sessions of an instance that stops refreshing them disappear on their own. /// public sealed class RedisSessionDirectory : ISessionDirectory { private const string KeyPrefix = "jellyfin:session:"; private const string OwnerKeyPrefix = "jellyfin:sessionowner:"; + private const string EpochKeyPrefix = "jellyfin:sessionepoch:"; /// - /// Lua script for an atomic ownership claim. The owner key holds connectedTicks|pod, 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. + /// Lua script for an atomic ownership claim. The owner key holds epoch|pod, where the epoch is + /// zero for an instance that holds no connection. Another instance takes ownership only by presenting + /// a connection epoch newer than the recorded one, so neither the instances serving the session's + /// requests nor two instances without a connection can take it from the one that has it. /// private const string ClaimScript = @" +redis.call('PEXPIRE', KEYS[3], ARGV[5]) local current = redis.call('GET', KEYS[1]) if current then local separator = string.find(current, '|', 1, true) local connected = tonumber(string.sub(current, 1, separator - 1)) local owner = string.sub(current, separator + 1) - if owner ~= ARGV[1] and connected > 0 and tonumber(ARGV[2]) <= connected then + local claiming = tonumber(ARGV[2]) + if owner ~= ARGV[1] and (claiming == 0 or claiming <= connected) then return 0 end end @@ -56,6 +60,15 @@ if string.sub(current, separator + 1) ~= ARGV[1] then return 0 end redis.call('DEL', KEYS[1], KEYS[2]) return 1"; + /// + /// Lua script allocating the next connection epoch. The counter outlives the entries that reference + /// it, so it never restarts underneath a recorded epoch. + /// + private const string EpochScript = @" +local epoch = redis.call('INCR', KEYS[1]) +redis.call('PEXPIRE', KEYS[1], ARGV[1]) +return epoch"; + private static readonly JsonSerializerOptions _jsonOptions = JsonDefaults.Options; private readonly IConnectionMultiplexer _redis; @@ -85,10 +98,26 @@ return 1"; private long EntryTtlMs => Math.Max(1, _options.EntryTtlSeconds) * 1000L; + // Outlives the entries that name an epoch, so a live entry never outlives the counter it came from. + private long EpochTtlMs => EntryTtlMs * 4; + private TimeSpan OperationTimeout => TimeSpan.FromSeconds(Math.Max(1, _options.OperationTimeoutSeconds)); /// - public async Task PublishAsync(SessionDirectoryEntry entry, long connectedUtcTicks, CancellationToken cancellationToken = default) + public async Task AllocateConnectionEpochAsync(string sessionId, CancellationToken cancellationToken = default) + { + ArgumentException.ThrowIfNullOrEmpty(sessionId); + + var epoch = (long?)await _db.ScriptEvaluateAsync( + EpochScript, + keys: new RedisKey[] { EpochKeyPrefix + sessionId }, + values: new RedisValue[] { EpochTtlMs }).WaitAsync(OperationTimeout, cancellationToken).ConfigureAwait(false); + + return epoch ?? 0; + } + + /// + public async Task PublishAsync(SessionDirectoryEntry entry, long connectionEpoch, CancellationToken cancellationToken = default) { ArgumentNullException.ThrowIfNull(entry); @@ -102,13 +131,14 @@ return 1"; { var claimed = (long?)await _db.ScriptEvaluateAsync( ClaimScript, - keys: new RedisKey[] { OwnerKeyPrefix + sessionId, KeyPrefix + sessionId }, + keys: new RedisKey[] { OwnerKeyPrefix + sessionId, KeyPrefix + sessionId, EpochKeyPrefix + sessionId }, values: new RedisValue[] { entry.OwnerPod, - connectedUtcTicks.ToString(CultureInfo.InvariantCulture), + connectionEpoch.ToString(CultureInfo.InvariantCulture), JsonSerializer.Serialize(entry, _jsonOptions), - EntryTtlMs + EntryTtlMs, + EpochTtlMs }).WaitAsync(OperationTimeout, cancellationToken).ConfigureAwait(false); return claimed == 1; @@ -139,17 +169,11 @@ return 1"; /// public async Task GetAsync(string sessionId, CancellationToken cancellationToken = default) { - try - { - var raw = await _db.StringGetAsync(KeyPrefix + sessionId).WaitAsync(OperationTimeout, cancellationToken).ConfigureAwait(false); + // A store that cannot be read says nothing about where the session is, so the failure is raised + // rather than reported as "no such session", which would be acted on as a local-only session. + var raw = await _db.StringGetAsync(KeyPrefix + sessionId).WaitAsync(OperationTimeout, cancellationToken).ConfigureAwait(false); - return raw.HasValue ? Deserialize(raw) : null; - } - catch (Exception ex) - { - _logger.LogWarning(ex, "Failed to read session {SessionId} from the directory.", sessionId); - return null; - } + return raw.HasValue ? Deserialize(raw) : null; } /// @@ -157,49 +181,61 @@ return 1"; { var entries = new List(); - try + foreach (var server in _redis.GetServers()) { - foreach (var server in _redis.GetServers()) + if (!server.IsConnected) { - if (!server.IsConnected) - { - continue; - } + continue; + } - var keys = new List(); + var keys = new List(); + + try + { await foreach (var key in server.KeysAsync(database: _db.Database, pattern: KeyPrefix + "*", pageSize: 1000).WithCancellation(cancellationToken).ConfigureAwait(false)) { keys.Add(key); } + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + // Degrade to the sessions that could be read rather than failing the request outright. + _logger.LogWarning(ex, "Failed to list the session directory on {Server}; its sessions are not reported.", server.EndPoint); + continue; + } - var values = await Task.WhenAll(keys.Select(key => _db.StringGetAsync(key))) - .WaitAsync(OperationTimeout, cancellationToken).ConfigureAwait(false); - - foreach (var raw in values) + foreach (var raw in await Task.WhenAll(keys.Select(key => ReadAsync(key, cancellationToken))).ConfigureAwait(false)) + { + if (!raw.HasValue) { - if (!raw.HasValue) - { - continue; - } + continue; + } - var entry = Deserialize(raw); - if (entry?.Session is not null) - { - entries.Add(entry); - } + 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(); - } return entries; } + // One unreadable key must not discard the entries that did load. + private async Task ReadAsync(RedisKey key, CancellationToken cancellationToken) + { + try + { + return await _db.StringGetAsync(key).WaitAsync(OperationTimeout, cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + _logger.LogWarning(ex, "Failed to read a session directory entry."); + return RedisValue.Null; + } + } + private SessionDirectoryEntry? Deserialize(RedisValue raw) { try diff --git a/Emby.Server.Implementations/Session/RemoteSessionController.cs b/Emby.Server.Implementations/Session/RemoteSessionController.cs index e1c3cb005d..eb134be469 100644 --- a/Emby.Server.Implementations/Session/RemoteSessionController.cs +++ b/Emby.Server.Implementations/Session/RemoteSessionController.cs @@ -13,7 +13,7 @@ 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. +/// instance, which writes them to the connection it owns and reports back whether it did. /// public sealed class RemoteSessionController : ISessionController { @@ -56,7 +56,7 @@ public sealed class RemoteSessionController : ISessionController Data = JsonSerializer.Serialize(data, JsonDefaults.Options) }; - var delivered = await _bus.PublishAsync( + var delivered = await _bus.RequestAsync( _ownerPod, new PodMessage { @@ -65,12 +65,12 @@ public sealed class RemoteSessionController : ISessionController }, cancellationToken).ConfigureAwait(false); - if (delivered == 0) + if (!delivered) { - _logger.LogWarning("Instance {OwnerPod} holds session {SessionId} but is not listening; the {MessageType} message was not delivered.", _ownerPod, _sessionId, name); + _logger.LogWarning("Instance {OwnerPod} did not write the {MessageType} message for session {SessionId} to a connection.", _ownerPod, name, _sessionId); throw new ResourceNotFoundException( - string.Format(CultureInfo.InvariantCulture, "The instance holding session {0} is unreachable.", _sessionId)); + string.Format(CultureInfo.InvariantCulture, "No instance holds a connection to session {0}.", _sessionId)); } } } diff --git a/Emby.Server.Implementations/Session/SessionManager.cs b/Emby.Server.Implementations/Session/SessionManager.cs index 5a62ecca77..58a23c16b6 100644 --- a/Emby.Server.Implementations/Session/SessionManager.cs +++ b/Emby.Server.Implementations/Session/SessionManager.cs @@ -73,6 +73,7 @@ namespace Emby.Server.Implementations.Session private readonly ConcurrentDictionary _connectionEpochs = new(StringComparer.Ordinal); private readonly ConcurrentDictionary _lastDirectoryPublish = new(StringComparer.Ordinal); + private readonly ConcurrentDictionary _directoryOwned = new(StringComparer.Ordinal); private readonly ConcurrentDictionary> _activeLiveStreamSessions = new(StringComparer.OrdinalIgnoreCase); @@ -80,6 +81,7 @@ namespace Emby.Server.Implementations.Session private Timer _idleTimer; private Timer _inactiveTimer; private Timer _directoryTimer; + private int _refreshingDirectory; private DtoOptions _itemInfoDtoOptions; private bool _disposed; @@ -364,6 +366,20 @@ namespace Emby.Server.Implementations.Session } } + // A playback transition changes what every instance's session list shows, so it is not left to + // the throttle. + private Task PublishToDirectoryNowAsync(SessionInfo session) + { + if (!_directoryEnabled || string.IsNullOrEmpty(session.Id)) + { + return Task.CompletedTask; + } + + _lastDirectoryPublish[session.Id] = Environment.TickCount64; + + return PublishToDirectoryAsync(session); + } + private async Task PublishToDirectoryAsync(SessionInfo session) { if (!_directoryEnabled || string.IsNullOrEmpty(session.Id)) @@ -373,16 +389,16 @@ namespace Emby.Server.Implementations.Session try { - var connectedUtcTicks = GetConnectionEpoch(session); + var connectionEpoch = await GetConnectionEpochAsync(session).ConfigureAwait(false); - await _sessionDirectory.PublishAsync( + _directoryOwned[session.Id] = await _sessionDirectory.PublishAsync( new SessionDirectoryEntry { OwnerPod = _podMessageBus.PodId, - HoldsConnection = connectedUtcTicks > 0, + HoldsConnection = connectionEpoch > 0, Session = ToSessionInfoDto(session) }, - connectedUtcTicks).ConfigureAwait(false); + connectionEpoch).ConfigureAwait(false); } catch (Exception ex) { @@ -391,8 +407,10 @@ namespace Emby.Server.Implementations.Session } // 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) + // controller claims with epoch zero, which never displaces an entry another instance holds. The + // epoch is handed out by the shared store, so the epochs of two instances are ordered by one + // clock rather than by whichever machine's wall clock wrote them. + private async Task GetConnectionEpochAsync(SessionInfo session) { if (!session.SessionControllers.Any(i => i.IsSessionActive)) { @@ -400,13 +418,21 @@ namespace Emby.Server.Implementations.Session return 0; } - return _connectionEpochs.GetOrAdd(session.Id, _ => DateTime.UtcNow.Ticks); + if (_connectionEpochs.TryGetValue(session.Id, out var epoch)) + { + return epoch; + } + + var allocated = await _sessionDirectory.AllocateConnectionEpochAsync(session.Id).ConfigureAwait(false); + + return _connectionEpochs.GetOrAdd(session.Id, allocated); } private async ValueTask RemoveFromDirectoryAsync(SessionInfo session) { _connectionEpochs.TryRemove(session.Id, out _); _lastDirectoryPublish.TryRemove(session.Id, out _); + _directoryOwned.TryRemove(session.Id, out _); if (!_directoryEnabled || string.IsNullOrEmpty(session.Id)) { @@ -416,19 +442,37 @@ namespace Emby.Server.Implementations.Session await _sessionDirectory.RemoveAsync(session.Id, _podMessageBus.PodId).ConfigureAwait(false); } - private async void RefreshSessionDirectory(object state) + private void RefreshSessionDirectory(object state) + { + if (Interlocked.CompareExchange(ref _refreshingDirectory, 1, 0) == 0) + { + _ = RefreshSessionDirectoryAsync(); + } + } + + // Only the entries this instance can hold are refreshed: republishing a copy of a session another + // instance owns just has the claim refused. + private async Task RefreshSessionDirectoryAsync() { try { foreach (var session in _activeConnections.Values) { - await PublishToDirectoryAsync(session).ConfigureAwait(false); + if (session.SessionControllers.Any(i => i.IsSessionActive) + || (_directoryOwned.TryGetValue(session.Id, out var owned) && owned)) + { + await PublishToDirectoryAsync(session).ConfigureAwait(false); + } } } catch (Exception ex) { _logger.LogDebug(ex, "Error refreshing the session directory."); } + finally + { + Interlocked.Exchange(ref _refreshingDirectory, 0); + } } private async Task> GetRemoteEntriesAsync(CancellationToken cancellationToken) @@ -479,34 +523,38 @@ namespace Emby.Server.Implementations.Session Capabilities = dto.Capabilities?.ToClientCapabilities() }; - if (entry.HoldsConnection) - { - session.AddController(new RemoteSessionController(_podMessageBus, _logger, entry.OwnerPod, dto.Id, dto.SupportsMediaControl)); - } + session.AddController(new RemoteSessionController(_podMessageBus, _logger, entry.OwnerPod, dto.Id, dto.SupportsMediaControl)); return session; } - private Task OnPodMessage(PodMessage message) + // What is returned here becomes the sender's answer, so a message this instance could not carry + // out is reported as undelivered instead of being counted by the sender as having arrived. + private Task OnPodMessage(PodMessage message) { switch (message.Kind) { case RoutedSessionMessage.Kind: return OnRoutedSessionMessage(message); case RoutedAdditionalUserChange.Kind: - OnRoutedAdditionalUserChange(message); - return Task.CompletedTask; + return Task.FromResult(OnRoutedAdditionalUserChange(message)); + case RoutedNowViewingItem.Kind: + return Task.FromResult(OnRoutedNowViewingItem(message)); + case RoutedPlaybackReport.StartKind: + case RoutedPlaybackReport.ProgressKind: + case RoutedPlaybackReport.StoppedKind: + return OnRoutedPlaybackReport(message); default: - return Task.CompletedTask; + return Task.FromResult(false); } } - private async Task OnRoutedSessionMessage(PodMessage message) + private async Task OnRoutedSessionMessage(PodMessage message) { var routed = JsonSerializer.Deserialize(message.Payload, JsonDefaults.Options); if (routed is null) { - return; + return false; } var session = Sessions.FirstOrDefault(i => string.Equals(i.Id, routed.SessionId, StringComparison.Ordinal)); @@ -518,7 +566,7 @@ namespace Emby.Server.Implementations.Session "A {MessageType} message for session {Session} was routed to this instance, which no longer holds its connection.", routed.MessageType, routed.SessionId); - return; + return false; } using var data = JsonDocument.Parse(routed.Data); @@ -526,9 +574,11 @@ namespace Emby.Server.Implementations.Session { await controller.SendMessage(routed.MessageType, routed.MessageId, data.RootElement, CancellationToken.None).ConfigureAwait(false); } + + return true; } - private void OnRoutedAdditionalUserChange(PodMessage message) + private bool OnRoutedAdditionalUserChange(PodMessage message) { var routed = JsonSerializer.Deserialize(message.Payload, JsonDefaults.Options); var session = routed is null @@ -537,7 +587,7 @@ namespace Emby.Server.Implementations.Session if (session is null) { - return; + return false; } if (routed.Add) @@ -548,6 +598,95 @@ namespace Emby.Server.Implementations.Session { DetachAdditionalUser(session, routed.UserId); } + + QueueDirectoryPublish(session); + + return true; + } + + private bool OnRoutedNowViewingItem(PodMessage message) + { + var routed = JsonSerializer.Deserialize(message.Payload, JsonDefaults.Options); + var session = routed is null + ? null + : Sessions.FirstOrDefault(i => string.Equals(i.Id, routed.SessionId, StringComparison.Ordinal)); + + if (session is null) + { + return false; + } + + SetNowViewingItem(session, routed.ItemId); + + return true; + } + + private async Task OnRoutedPlaybackReport(PodMessage message) + { + try + { + switch (message.Kind) + { + case RoutedPlaybackReport.StartKind: + await OnPlaybackStartCore(Deserialize(message)).ConfigureAwait(false); + return true; + case RoutedPlaybackReport.ProgressKind: + return await OnPlaybackProgressCore(Deserialize(message), false).ConfigureAwait(false); + default: + await OnPlaybackStoppedCore(Deserialize(message)).ConfigureAwait(false); + return true; + } + } + catch (Exception ex) + { + _logger.LogWarning(ex, "A {Kind} report routed to this instance could not be applied.", message.Kind); + return false; + } + } + + private static T Deserialize(PodMessage message) + => JsonSerializer.Deserialize(message.Payload, JsonDefaults.Options); + + // A sessionId-addressed mutation belongs to the instance whose copy of the session everyone else + // is shown. An undeliverable route falls back to handling it here, which is what a deployment + // without a directory does anyway. + private async Task TryRouteToOwnerAsync(string sessionId, string kind, object payload, CancellationToken cancellationToken) + { + if (!_directoryEnabled || string.IsNullOrEmpty(sessionId)) + { + return false; + } + + try + { + var entry = await _sessionDirectory.GetAsync(sessionId, cancellationToken).ConfigureAwait(false); + + if (entry is null || string.Equals(entry.OwnerPod, _podMessageBus.PodId, StringComparison.Ordinal)) + { + return false; + } + + var routed = await _podMessageBus.RequestAsync( + entry.OwnerPod, + new PodMessage + { + Kind = kind, + Payload = JsonSerializer.Serialize(payload, JsonDefaults.Options) + }, + cancellationToken).ConfigureAwait(false); + + if (!routed) + { + _logger.LogWarning("Instance {OwnerPod} did not apply the {Kind} report for session {Session}; it is applied here instead.", entry.OwnerPod, kind, sessionId); + } + + return routed; + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + _logger.LogWarning(ex, "Could not reach the owner of session {Session}; the {Kind} report is applied here instead.", sessionId, kind); + return false; + } } /// @@ -1014,6 +1153,16 @@ namespace Emby.Server.Implementations.Session ArgumentNullException.ThrowIfNull(info); + if (await TryRouteToOwnerAsync(info.SessionId, RoutedPlaybackReport.StartKind, info, CancellationToken.None).ConfigureAwait(false)) + { + return; + } + + await OnPlaybackStartCore(info).ConfigureAwait(false); + } + + private async Task OnPlaybackStartCore(PlaybackStartInfo info) + { var session = GetSession(info.SessionId); var libraryItem = info.ItemId.IsEmpty() @@ -1080,6 +1229,8 @@ namespace Emby.Server.Implementations.Session _logger); StartCheckTimers(); + + await PublishToDirectoryNowAsync(session).ConfigureAwait(false); } /// @@ -1146,10 +1297,23 @@ namespace Emby.Server.Implementations.Session ArgumentNullException.ThrowIfNull(info); + // An automated report is generated from the copy of the session this instance already holds, + // so it is never the one that belongs somewhere else. + if (!isAutomated + && await TryRouteToOwnerAsync(info.SessionId, RoutedPlaybackReport.ProgressKind, info, CancellationToken.None).ConfigureAwait(false)) + { + return; + } + + await OnPlaybackProgressCore(info, isAutomated).ConfigureAwait(false); + } + + private async Task OnPlaybackProgressCore(PlaybackProgressInfo info, bool isAutomated) + { var session = GetSession(info.SessionId, false); if (session is null) { - return; + return false; } var libraryItem = info.ItemId.IsEmpty() @@ -1206,6 +1370,10 @@ namespace Emby.Server.Implementations.Session } StartCheckTimers(); + + QueueDirectoryPublish(session); + + return true; } private void OnPlaybackProgress(User user, BaseItem item, PlaybackProgressInfo info) @@ -1302,6 +1470,16 @@ namespace Emby.Server.Implementations.Session ArgumentNullException.ThrowIfNull(info); + if (await TryRouteToOwnerAsync(info.SessionId, RoutedPlaybackReport.StoppedKind, info, CancellationToken.None).ConfigureAwait(false)) + { + return; + } + + await OnPlaybackStoppedCore(info).ConfigureAwait(false); + } + + private async Task OnPlaybackStoppedCore(PlaybackStopInfo info) + { var session = GetSession(info.SessionId); session.StopAutomaticProgress(); @@ -1406,6 +1584,8 @@ namespace Emby.Server.Implementations.Session await _eventManager.PublishAsync(eventArgs).ConfigureAwait(false); EventHelper.QueueEventIfNotNull(PlaybackStopped, this, eventArgs, _logger); + + await PublishToDirectoryNowAsync(session).ConfigureAwait(false); } private bool OnPlaybackStopped(User user, BaseItem item, long? positionTicks, bool playbackFailed) @@ -1465,8 +1645,11 @@ namespace Emby.Server.Implementations.Session return session; } - // 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. + // Resolves the session a message has to be written to. A local SessionInfo without a live + // controller is a copy left behind by a request this instance happened to serve, not the + // connection, so it is never the answer: either the owner named by the directory can take the + // message or nothing can, and the caller is told so rather than handed a copy that silently + // swallows it. A directory that cannot be read raises rather than reading as "no such session". private async Task GetSessionToRemoteControl(string sessionId) { var local = Sessions.FirstOrDefault(i => string.Equals(i.Id, sessionId, StringComparison.Ordinal)); @@ -1476,15 +1659,21 @@ namespace Emby.Server.Implementations.Session return local; } - var session = await GetRemoteSession(sessionId).ConfigureAwait(false) ?? local; + return await GetRemoteSession(sessionId).ConfigureAwait(false) + ?? throw new ResourceNotFoundException( + string.Format(CultureInfo.InvariantCulture, "No instance holds a connection to session {0}.", sessionId)); + } - if (session is null) - { - throw new ResourceNotFoundException( + // Resolves a session to authorize against or to change server-side state on. Nothing is written to + // a connection here, so a copy without one still answers the question. + private async Task GetSessionForControl(string sessionId) + { + var local = Sessions.FirstOrDefault(i => string.Equals(i.Id, sessionId, StringComparison.Ordinal)); + + return local + ?? await GetRemoteSession(sessionId).ConfigureAwait(false) + ?? throw new ResourceNotFoundException( string.Format(CultureInfo.InvariantCulture, "Session {0} not found.", sessionId)); - } - - return session; } /// @@ -1553,7 +1742,7 @@ namespace Emby.Server.Implementations.Session if (!string.IsNullOrEmpty(controllingSessionId)) { - var controllingSession = await GetSessionToRemoteControl(controllingSessionId).ConfigureAwait(false); + var controllingSession = await GetSessionForControl(controllingSessionId).ConfigureAwait(false); AssertCanControl(session, controllingSession); } @@ -1562,7 +1751,14 @@ namespace Emby.Server.Implementations.Session private static async Task SendMessageToSession(SessionInfo session, SessionMessageType name, T data, CancellationToken cancellationToken) { - var controllers = session.SessionControllers; + var controllers = session.SessionControllers.Where(i => i.IsSessionActive).ToList(); + + if (controllers.Count == 0) + { + throw new ResourceNotFoundException( + string.Format(CultureInfo.InvariantCulture, "No instance holds a connection to session {0}.", session.Id)); + } + var messageId = Guid.NewGuid(); foreach (var controller in controllers) @@ -1664,7 +1860,7 @@ namespace Emby.Server.Implementations.Session if (!string.IsNullOrEmpty(controllingSessionId)) { - var controllingSession = await GetSessionToRemoteControl(controllingSessionId).ConfigureAwait(false); + var controllingSession = await GetSessionForControl(controllingSessionId).ConfigureAwait(false); AssertCanControl(session, controllingSession); if (!controllingSession.UserId.IsEmpty()) { @@ -1682,10 +1878,10 @@ namespace Emby.Server.Implementations.Session // 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); + var session = GetConnectedSession(sessionId); if (session is null) { - _logger.LogDebug("SyncPlay command for session {Session} dropped; it is not held by this instance.", sessionId); + _logger.LogDebug("SyncPlay command for session {Session} dropped; this instance does not hold its connection.", sessionId); return; } @@ -1697,16 +1893,25 @@ namespace Emby.Server.Implementations.Session { CheckDisposed(); - var session = GetSession(sessionId, false); + var session = GetConnectedSession(sessionId); if (session is null) { - _logger.LogDebug("SyncPlay group update for session {Session} dropped; it is not held by this instance.", sessionId); + _logger.LogDebug("SyncPlay group update for session {Session} dropped; this instance does not hold its connection.", sessionId); return; } await SendMessageToSession(session, SessionMessageType.SyncPlayGroupUpdate, command, cancellationToken).ConfigureAwait(false); } + // Both instances keep a copy of a session whose requests they have served, so holding a copy is + // not holding the connection. + private SessionInfo GetConnectedSession(string sessionId) + { + var session = Sessions.FirstOrDefault(i => string.Equals(i.Id, sessionId, StringComparison.Ordinal)); + + return session?.SessionControllers.Any(i => i.IsSessionActive) == true ? session : null; + } + private IEnumerable TranslateItemForPlayback(Guid id, User user) { var item = _libraryManager.GetItemById(id); @@ -1799,7 +2004,7 @@ namespace Emby.Server.Implementations.Session if (!string.IsNullOrEmpty(controllingSessionId)) { - var controllingSession = await GetSessionToRemoteControl(controllingSessionId).ConfigureAwait(false); + var controllingSession = await GetSessionForControl(controllingSessionId).ConfigureAwait(false); AssertCanControl(session, controllingSession); if (!controllingSession.UserId.IsEmpty()) { @@ -1883,11 +2088,11 @@ namespace Emby.Server.Implementations.Session { CheckDisposed(); - var session = await GetSessionToRemoteControl(sessionId).ConfigureAwait(false); + var session = await GetSessionForControl(sessionId).ConfigureAwait(false); if (!string.IsNullOrEmpty(controllingSessionId)) { - var controllingSession = await GetSessionToRemoteControl(controllingSessionId).ConfigureAwait(false); + var controllingSession = await GetSessionForControl(controllingSessionId).ConfigureAwait(false); AssertCanControl(session, controllingSession); AssertCanAttachUser(controllingSession, userId); } @@ -1900,13 +2105,14 @@ namespace Emby.Server.Implementations.Session var user = _userManager.GetUserById(userId) ?? throw new ArgumentException("The requested user does not exist."); + await RouteAdditionalUserChange(sessionId, userId, true).ConfigureAwait(false); + var local = GetSession(sessionId, false); if (local is not null) { AttachAdditionalUser(local, userId, user.Username); + QueueDirectoryPublish(local); } - - await RouteAdditionalUserChange(sessionId, userId, true).ConfigureAwait(false); } /// @@ -1922,11 +2128,11 @@ namespace Emby.Server.Implementations.Session { CheckDisposed(); - var session = await GetSessionToRemoteControl(sessionId).ConfigureAwait(false); + var session = await GetSessionForControl(sessionId).ConfigureAwait(false); if (!string.IsNullOrEmpty(controllingSessionId)) { - AssertCanControl(session, await GetSessionToRemoteControl(controllingSessionId).ConfigureAwait(false)); + AssertCanControl(session, await GetSessionForControl(controllingSessionId).ConfigureAwait(false)); } if (session.UserId.Equals(userId)) @@ -1934,13 +2140,14 @@ namespace Emby.Server.Implementations.Session throw new ArgumentException("The requested user is already the primary user of the session."); } + await RouteAdditionalUserChange(sessionId, userId, false).ConfigureAwait(false); + var local = GetSession(sessionId, false); if (local is not null) { DetachAdditionalUser(local, userId); + QueueDirectoryPublish(local); } - - await RouteAdditionalUserChange(sessionId, userId, false).ConfigureAwait(false); } private static void AttachAdditionalUser(SessionInfo session, Guid userId, string userName) @@ -1987,7 +2194,7 @@ namespace Emby.Server.Implementations.Session Add = add }; - var delivered = await _podMessageBus.PublishAsync( + var delivered = await _podMessageBus.RequestAsync( entry.OwnerPod, new PodMessage { @@ -1995,9 +2202,10 @@ namespace Emby.Server.Implementations.Session Payload = JsonSerializer.Serialize(payload, JsonDefaults.Options) }).ConfigureAwait(false); - if (delivered == 0) + if (!delivered) { - _logger.LogWarning("Instance {OwnerPod} holds session {Session} but is not listening; the additional user change was not applied there.", entry.OwnerPod, sessionId); + throw new ResourceNotFoundException( + string.Format(CultureInfo.InvariantCulture, "The instance holding session {0} did not apply the change.", sessionId)); } } @@ -2297,19 +2505,36 @@ namespace Emby.Server.Implementations.Session } /// - public void ReportNowViewingItem(string controllingSessionId, string sessionId, string itemId) + public async Task ReportNowViewingItem(string controllingSessionId, string sessionId, string itemId) { ArgumentException.ThrowIfNullOrEmpty(itemId); - var item = _libraryManager.GetItemById(new Guid(itemId)); - var session = GetSession(sessionId); + var session = await GetSessionForControl(sessionId).ConfigureAwait(false); if (!string.IsNullOrEmpty(controllingSessionId)) { - AssertCanControl(session, GetSession(controllingSessionId)); + AssertCanControl(session, await GetSessionForControl(controllingSessionId).ConfigureAwait(false)); } - session.NowViewingItem = GetItemInfo(item, null); + var payload = new RoutedNowViewingItem { SessionId = sessionId, ItemId = itemId }; + + if (await TryRouteToOwnerAsync(sessionId, RoutedNowViewingItem.Kind, payload, CancellationToken.None).ConfigureAwait(false)) + { + return; + } + + var local = GetSession(sessionId, false); + if (local is not null) + { + SetNowViewingItem(local, itemId); + } + } + + private void SetNowViewingItem(SessionInfo session, string itemId) + { + session.NowViewingItem = GetItemInfo(_libraryManager.GetItemById(new Guid(itemId)), null); + + QueueDirectoryPublish(session); } /// diff --git a/Jellyfin.Api/Controllers/SessionController.cs b/Jellyfin.Api/Controllers/SessionController.cs index 05506d7b0d..3e4fe3367e 100644 --- a/Jellyfin.Api/Controllers/SessionController.cs +++ b/Jellyfin.Api/Controllers/SessionController.cs @@ -416,7 +416,7 @@ public class SessionController : BaseJellyfinApiController { var currentSessionId = await RequestHelpers.GetSessionId(_sessionManager, _userManager, HttpContext).ConfigureAwait(false); - _sessionManager.ReportNowViewingItem(currentSessionId, sessionId ?? currentSessionId, itemId); + await _sessionManager.ReportNowViewingItem(currentSessionId, sessionId ?? currentSessionId, itemId).ConfigureAwait(false); return NoContent(); } diff --git a/Jellyfin.Server/Extensions/SessionDirectoryServiceCollectionExtensions.cs b/Jellyfin.Server/Extensions/SessionDirectoryServiceCollectionExtensions.cs index 33977ef9a4..fc00725a83 100644 --- a/Jellyfin.Server/Extensions/SessionDirectoryServiceCollectionExtensions.cs +++ b/Jellyfin.Server/Extensions/SessionDirectoryServiceCollectionExtensions.cs @@ -5,6 +5,7 @@ using MediaBrowser.Controller.Session; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; using StackExchange.Redis; namespace Jellyfin.Server.Extensions; @@ -38,46 +39,56 @@ public static class SessionDirectoryServiceCollectionExtensions "Session directory: {Directory}. The session list and remote control only reach the sessions this instance holds; set {Key} to share them.", nameof(NullSessionDirectory), TranscodeStoreOptions.RedisConnectionStringKey); + } + else + { + logger.LogInformation( + "Session directory: {Directory}. Sessions are visible to, and controllable from, every instance.", + nameof(RedisSessionDirectory)); - serviceCollection.AddSingleton(NullSessionDirectory.Instance); - return serviceCollection.AddSingleton(NullPodMessageBus.Instance); + serviceCollection.AddSingleton(SharedSessionServices.Create); } - logger.LogInformation( - "Session directory: {Directory}. Sessions are visible to, and controllable from, every instance.", - nameof(RedisSessionDirectory)); + serviceCollection.AddSingleton(sp => sp.GetService()?.Bus ?? NullPodMessageBus.Instance); - 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)); + return serviceCollection.AddSingleton(sp => sp.GetService()?.Directory ?? 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) + // A Redis directory paired with a no-op bus would leave every routed command permanently + // undeliverable, so the pair is built together and falls back together. + private sealed class SharedSessionServices { - try + private SharedSessionServices(ISessionDirectory directory, IPodMessageBus bus) { - return factory(); + Directory = directory; + Bus = bus; } - 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; + public ISessionDirectory Directory { get; } + + public IPodMessageBus Bus { get; } + + // Fail open: an unreachable Redis degrades to the single-instance behaviour rather than aborting startup. + public static SharedSessionServices Create(IServiceProvider serviceProvider) + { + try + { + var redis = serviceProvider.GetRequiredService(); + var options = serviceProvider.GetRequiredService>(); + + return new SharedSessionServices( + new RedisSessionDirectory(redis, options, serviceProvider.GetRequiredService>()), + new RedisPodMessageBus(redis, options, PodIdentity.Current, serviceProvider.GetRequiredService>())); + } + catch (Exception ex) + { + serviceProvider.GetRequiredService>().LogError( + ex, + "Redis is configured but unavailable, so sessions will not be shared between instances. Check {Key}.", + TranscodeStoreOptions.RedisConnectionStringKey); + + return new SharedSessionServices(NullSessionDirectory.Instance, NullPodMessageBus.Instance); + } } } } diff --git a/MediaBrowser.Controller/Session/IPodMessageBus.cs b/MediaBrowser.Controller/Session/IPodMessageBus.cs index 4caa6c90c7..3aca9e7181 100644 --- a/MediaBrowser.Controller/Session/IPodMessageBus.cs +++ b/MediaBrowser.Controller/Session/IPodMessageBus.cs @@ -16,18 +16,20 @@ public interface IPodMessageBus string PodId { get; } /// - /// 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. + /// Sends a message to one instance and waits for that instance to report what it did with it. The + /// number of subscribers only proves the target's connection to the broker is up, so delivery is + /// taken from the acknowledgement of the instance that has to act on the message. /// /// The instance to deliver to. /// The message. /// The cancellation token. - /// The number of instances the message reached. - Task PublishAsync(string targetPod, PodMessage message, CancellationToken cancellationToken = default); + /// true if the target acknowledged having carried the message out. + Task RequestAsync(string targetPod, PodMessage message, CancellationToken cancellationToken = default); /// - /// Registers a handler for the messages addressed to this instance. + /// Registers a handler for the messages addressed to this instance. Whatever the handler returns is + /// sent back to the origin as the acknowledgement. /// /// The handler. - void Subscribe(Func handler); + void Subscribe(Func> handler); } diff --git a/MediaBrowser.Controller/Session/ISessionDirectory.cs b/MediaBrowser.Controller/Session/ISessionDirectory.cs index 5a35782571..fe5562566a 100644 --- a/MediaBrowser.Controller/Session/ISessionDirectory.cs +++ b/MediaBrowser.Controller/Session/ISessionDirectory.cs @@ -10,16 +10,25 @@ namespace MediaBrowser.Controller.Session; /// public interface ISessionDirectory { + /// + /// Allocates the next connection epoch for a session. The counter lives in the shared store, so the + /// epochs of every instance are ordered by one clock instead of being compared across machines. + /// + /// The session identifier. + /// The cancellation token. + /// The allocated epoch, which is greater than every epoch allocated for the session before it. + Task AllocateConnectionEpochAsync(string sessionId, CancellationToken cancellationToken = default); + /// /// Claims a session for the publishing instance and restarts its expiry. The claim is refused when /// another instance holds the connection, so an instance that merely served a request for the session /// cannot take ownership of it. /// /// The entry. - /// When the publishing instance's connection to the session was established, or zero when it holds none. + /// The epoch of the publishing instance's connection to the session, or zero when it holds none. /// The cancellation token. /// true if the entry was written. - Task PublishAsync(SessionDirectoryEntry entry, long connectedUtcTicks, CancellationToken cancellationToken = default); + Task PublishAsync(SessionDirectoryEntry entry, long connectionEpoch, CancellationToken cancellationToken = default); /// /// Removes an entry, but only while the calling instance still owns it. @@ -36,6 +45,7 @@ public interface ISessionDirectory /// The session identifier. /// The cancellation token. /// The entry, or null when the session is in no instance's directory. + /// The store could not be read. An unreadable store is not an absent session. Task GetAsync(string sessionId, CancellationToken cancellationToken = default); /// diff --git a/MediaBrowser.Controller/Session/ISessionManager.cs b/MediaBrowser.Controller/Session/ISessionManager.cs index 3dbd716bfd..70609e95d1 100644 --- a/MediaBrowser.Controller/Session/ISessionManager.cs +++ b/MediaBrowser.Controller/Session/ISessionManager.cs @@ -260,7 +260,8 @@ namespace MediaBrowser.Controller.Session /// The controlling session identifier. /// The session identifier. /// The item identifier. - void ReportNowViewingItem(string controllingSessionId, string sessionId, string itemId); + /// A task representing the operation. + Task ReportNowViewingItem(string controllingSessionId, string sessionId, string itemId); /// /// Authenticates the new session. diff --git a/MediaBrowser.Controller/Session/NullPodMessageBus.cs b/MediaBrowser.Controller/Session/NullPodMessageBus.cs index a0b175a53e..1e95d6d907 100644 --- a/MediaBrowser.Controller/Session/NullPodMessageBus.cs +++ b/MediaBrowser.Controller/Session/NullPodMessageBus.cs @@ -18,11 +18,11 @@ public sealed class NullPodMessageBus : IPodMessageBus public string PodId => PodIdentity.Current; /// - public Task PublishAsync(string targetPod, PodMessage message, CancellationToken cancellationToken = default) - => Task.FromResult(0L); + public Task RequestAsync(string targetPod, PodMessage message, CancellationToken cancellationToken = default) + => Task.FromResult(false); /// - public void Subscribe(Func handler) + public void Subscribe(Func> handler) { } } diff --git a/MediaBrowser.Controller/Session/NullSessionDirectory.cs b/MediaBrowser.Controller/Session/NullSessionDirectory.cs index 63ef843f9f..fa4ca717c6 100644 --- a/MediaBrowser.Controller/Session/NullSessionDirectory.cs +++ b/MediaBrowser.Controller/Session/NullSessionDirectory.cs @@ -17,7 +17,11 @@ public sealed class NullSessionDirectory : ISessionDirectory public static NullSessionDirectory Instance { get; } = new NullSessionDirectory(); /// - public Task PublishAsync(SessionDirectoryEntry entry, long connectedUtcTicks, CancellationToken cancellationToken = default) + public Task AllocateConnectionEpochAsync(string sessionId, CancellationToken cancellationToken = default) + => Task.FromResult(0L); + + /// + public Task PublishAsync(SessionDirectoryEntry entry, long connectionEpoch, CancellationToken cancellationToken = default) => Task.FromResult(false); /// diff --git a/MediaBrowser.Controller/Session/PodMessage.cs b/MediaBrowser.Controller/Session/PodMessage.cs index 1912b40300..3a0801521f 100644 --- a/MediaBrowser.Controller/Session/PodMessage.cs +++ b/MediaBrowser.Controller/Session/PodMessage.cs @@ -6,6 +6,11 @@ namespace MediaBrowser.Controller.Session; /// public sealed class PodMessage { + /// + /// The of the acknowledgement the receiving instance sends back. + /// + public const string AckKind = "Ack"; + /// /// Gets or sets the payload discriminator. /// @@ -16,6 +21,17 @@ public sealed class PodMessage /// public string OriginPod { get; set; } = string.Empty; + /// + /// Gets or sets the identifier tying an acknowledgement to the message it answers. + /// + public string CorrelationId { get; set; } = string.Empty; + + /// + /// Gets or sets a value indicating whether the receiving instance carried the message out. Only + /// meaningful on an message. + /// + public bool Handled { get; set; } + /// /// Gets or sets the serialized payload. /// diff --git a/MediaBrowser.Controller/Session/RoutedNowViewingItem.cs b/MediaBrowser.Controller/Session/RoutedNowViewingItem.cs new file mode 100644 index 0000000000..71eb0ce55f --- /dev/null +++ b/MediaBrowser.Controller/Session/RoutedNowViewingItem.cs @@ -0,0 +1,23 @@ +namespace MediaBrowser.Controller.Session; + +/// +/// A now-viewing report for a session held by another instance, carried as a . +/// The calling instance has already authorized it. +/// +public sealed class RoutedNowViewingItem +{ + /// + /// The this payload travels under. + /// + public const string Kind = "NowViewingItem"; + + /// + /// Gets or sets the session the report applies to. + /// + public string SessionId { get; set; } = string.Empty; + + /// + /// Gets or sets the item being viewed. + /// + public string ItemId { get; set; } = string.Empty; +} diff --git a/MediaBrowser.Controller/Session/RoutedPlaybackReport.cs b/MediaBrowser.Controller/Session/RoutedPlaybackReport.cs new file mode 100644 index 0000000000..70397216f7 --- /dev/null +++ b/MediaBrowser.Controller/Session/RoutedPlaybackReport.cs @@ -0,0 +1,23 @@ +namespace MediaBrowser.Controller.Session; + +/// +/// The values a playback report travels under when the session it is +/// addressed to is held by another instance. The payload is the report itself. +/// +public static class RoutedPlaybackReport +{ + /// + /// A PlaybackStartInfo. + /// + public const string StartKind = "PlaybackStart"; + + /// + /// A PlaybackProgressInfo. + /// + public const string ProgressKind = "PlaybackProgress"; + + /// + /// A PlaybackStopInfo. + /// + public const string StoppedKind = "PlaybackStopped"; +} diff --git a/tests/Jellyfin.Server.Implementations.Tests/SessionManager/SessionManagerTests.cs b/tests/Jellyfin.Server.Implementations.Tests/SessionManager/SessionManagerTests.cs index 5edbfde690..5228c94154 100644 --- a/tests/Jellyfin.Server.Implementations.Tests/SessionManager/SessionManagerTests.cs +++ b/tests/Jellyfin.Server.Implementations.Tests/SessionManager/SessionManagerTests.cs @@ -129,6 +129,7 @@ public class SessionManagerTests await using var sessionManager = CreateSessionManager(victim, attacker); var victimSession = await LogSessionActivity(sessionManager, victim); + victimSession.AddController(new StubSessionController()); var attackerSession = await LogSessionActivity(sessionManager, attacker); await Assert.ThrowsAsync(() => sessionManager.SendMessageCommand( @@ -147,6 +148,7 @@ public class SessionManagerTests await using var sessionManager = CreateSessionManager(victim, attacker); var victimSession = await LogSessionActivity(sessionManager, victim); + victimSession.AddController(new StubSessionController()); var controllingSession = await LogSessionActivity(sessionManager, attacker); await sessionManager.SendMessageCommand( @@ -255,4 +257,15 @@ public class SessionManagerTests // from the request headers and are not bound to the access token of the calling user. private static Task LogSessionActivity(ISessionManager sessionManager, User user) => sessionManager.LogSessionActivity("Jellyfin Web", "1.0.0", "victim-tv-01", "device_name", "127.0.0.1", user); + + // A session a command can be delivered to is one with a live connection. + private sealed class StubSessionController : ISessionController + { + public bool IsSessionActive => true; + + public bool SupportsMediaControl => true; + + public Task SendMessage(SessionMessageType name, Guid messageId, T data, CancellationToken cancellationToken) + => Task.CompletedTask; + } } diff --git a/tests/Jellyfin.Server.Tests/HighAvailability/SessionDirectoryRegistrationTests.cs b/tests/Jellyfin.Server.Tests/HighAvailability/SessionDirectoryRegistrationTests.cs new file mode 100644 index 0000000000..60f7c70d0d --- /dev/null +++ b/tests/Jellyfin.Server.Tests/HighAvailability/SessionDirectoryRegistrationTests.cs @@ -0,0 +1,52 @@ +using System.Collections.Generic; +using Jellyfin.Server.Extensions; +using MediaBrowser.Controller.MediaEncoding; +using MediaBrowser.Controller.Session; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Xunit; + +namespace Jellyfin.Server.Tests.HighAvailability; + +/// +/// The directory and the bus are two halves of one mechanism: a shared directory paired with a bus that +/// reaches nobody advertises sessions from every instance and then fails every command sent to one. +/// +public static class SessionDirectoryRegistrationTests +{ + [Fact] + public static void AddSessionDirectory_WithoutARedisConnection_RegistersNeitherHalf() + { + var services = Build(configured: false); + + Assert.IsType(services.GetRequiredService()); + Assert.IsType(services.GetRequiredService()); + } + + [Fact] + public static void AddSessionDirectory_WithAnUnusableRedisConnection_FallsBackOnBothHalves() + { + // The multiplexer is deliberately absent, which is what an unreachable Redis amounts to here. + var services = Build(configured: true); + + Assert.IsType(services.GetRequiredService()); + Assert.IsType(services.GetRequiredService()); + } + + private static ServiceProvider Build(bool configured) + { + var settings = new Dictionary(); + if (configured) + { + settings[TranscodeStoreOptions.RedisConnectionStringKey] = "127.0.0.1:6379"; + } + + var configuration = new ConfigurationBuilder().AddInMemoryCollection(settings).Build(); + + return new ServiceCollection() + .AddLogging() + .AddSessionDirectory(configuration, NullLogger.Instance) + .BuildServiceProvider(); + } +} diff --git a/tests/Jellyfin.Server.Tests/HighAvailability/SessionDirectoryReplicaTests.cs b/tests/Jellyfin.Server.Tests/HighAvailability/SessionDirectoryReplicaTests.cs index ef0fe965d7..9a6aadcc8b 100644 --- a/tests/Jellyfin.Server.Tests/HighAvailability/SessionDirectoryReplicaTests.cs +++ b/tests/Jellyfin.Server.Tests/HighAvailability/SessionDirectoryReplicaTests.cs @@ -1,4 +1,7 @@ using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Globalization; using System.Linq; using System.Text.Json; using System.Threading; @@ -18,9 +21,13 @@ using MediaBrowser.Controller.Dto; using MediaBrowser.Controller.Events; using MediaBrowser.Controller.Library; using MediaBrowser.Controller.Session; +using MediaBrowser.Model.Configuration; +using MediaBrowser.Model.Dto; using MediaBrowser.Model.Session; +using MediaBrowser.Model.SyncPlay; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; using Moq; @@ -229,6 +236,287 @@ public sealed class SessionDirectoryReplicaTests : IAsyncLifetime cancellationToken)); } + /// + /// The owner's entry is only as fresh as its last refresh, so a websocket that closes in between + /// leaves an entry claiming a connection that is gone. The command has to be reported undelivered, + /// which only the replica that would have written it to the socket can say. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task MessageRoutedToAnOwnerWhoseSocketDied_IsReportedAsUndelivered() + { + var cancellationToken = TestContext.Current.CancellationToken; + await using var replicaA = CreateReplica("pod-a"); + await using var replicaB = CreateReplica("pod-b"); + + var session = await Request(replicaA, "device-dead-socket"); + var controller = new RecordingSessionController(); + session.AddController(controller); + await replicaA.OnSessionControllerConnected(session); + + // The socket closes. Nothing rewrites the entry: it still names pod-a and still says the + // connection is held, exactly as it does for the rest of the refresh interval. + controller.IsSessionActive = false; + + var entry = await _directory.GetAsync(session.Id, cancellationToken); + Assert.NotNull(entry); + Assert.Equal("pod-a", entry.OwnerPod); + Assert.True(entry.HoldsConnection); + + await Assert.ThrowsAsync( + () => replicaB.SendMessageCommand( + string.Empty, + session.Id, + new MessageCommand { Header = "Header", Text = "Dinner is ready" }, + cancellationToken)); + } + + /// + /// A device reconnecting lands on either replica, so both can hold a live connection for the same + /// deterministic session id at once. Exactly one of them owns the entry, and it stays that one. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task BothReplicasHoldingAConnection_AgreeOnOneOwner() + { + var cancellationToken = TestContext.Current.CancellationToken; + var options = new SessionDirectoryOptions { EntryTtlSeconds = 60, RefreshIntervalSeconds = 1 }; + await using var replicaA = CreateReplica("pod-a", options); + await using var replicaB = CreateReplica("pod-b", options); + + var sessionA = await Request(replicaA, "device-two-sockets"); + sessionA.AddController(new RecordingSessionController()); + await replicaA.OnSessionControllerConnected(sessionA); + + var sessionB = await Request(replicaB, "device-two-sockets"); + sessionB.AddController(new RecordingSessionController()); + await replicaB.OnSessionControllerConnected(sessionB); + + Assert.Equal(sessionA.Id, sessionB.Id); + + // The later connection owns the session; both replicas keep republishing theirs. + for (var i = 0; i < 8; i++) + { + await Task.Delay(500, cancellationToken); + + var entry = await _directory.GetAsync(sessionA.Id, cancellationToken); + Assert.NotNull(entry); + Assert.Equal("pod-b", entry.OwnerPod); + Assert.True(entry.HoldsConnection); + } + + var listedByA = await replicaA.GetSessions(_user.Id, null, null, null, false, cancellationToken); + Assert.Single(listedByA, i => string.Equals(i.Id, sessionA.Id, StringComparison.Ordinal)); + } + + /// + /// A session with no websocket is claimed with a zero epoch by every replica that serves a request + /// for it. The first claim has to stand, or the listed session flips between two partial copies. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task TwoReplicasWithoutAConnection_DoNotTakeTheSessionFromEachOther() + { + var cancellationToken = TestContext.Current.CancellationToken; + var options = new SessionDirectoryOptions { EntryTtlSeconds = 60, RefreshIntervalSeconds = 1 }; + await using var replicaA = CreateReplica("pod-a", options); + await using var replicaB = CreateReplica("pod-b", options); + + var session = await Request(replicaA, "device-no-socket"); + await Request(replicaB, "device-no-socket"); + + var claimed = await _directory.GetAsync(session.Id, cancellationToken); + Assert.NotNull(claimed); + Assert.False(claimed.HoldsConnection); + + var owner = claimed.OwnerPod; + + Assert.False(await _directory.PublishAsync( + new SessionDirectoryEntry + { + OwnerPod = owner == "pod-a" ? "pod-b" : "pod-a", + HoldsConnection = false, + Session = claimed.Session + }, + 0, + cancellationToken)); + + for (var i = 0; i < 6; i++) + { + await Request(replicaB, "device-no-socket"); + await Task.Delay(400, cancellationToken); + await Request(replicaA, "device-no-socket"); + + var entry = await _directory.GetAsync(session.Id, cancellationToken); + Assert.NotNull(entry); + Assert.Equal(owner, entry.OwnerPod); + } + } + + /// + /// Without sticky sessions a playback report lands on either replica while the websocket stays on + /// one. The report belongs to the replica everyone else is shown, so it is applied there and the + /// session reads as playing from every replica rather than idle on all of them. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task PlaybackReportedToTheNonOwner_IsVisibleFromBothReplicas() + { + var cancellationToken = TestContext.Current.CancellationToken; + await using var replicaA = CreateReplica("pod-a"); + await using var replicaB = CreateReplica("pod-b"); + + var session = await Request(replicaA, "device-playing"); + session.AddController(new RecordingSessionController()); + await replicaA.OnSessionControllerConnected(session); + + // The load balancer hands the playback report to the replica without the websocket. + await Request(replicaB, "device-playing"); + await replicaB.OnPlaybackStart(new PlaybackStartInfo + { + SessionId = session.Id, + Item = new BaseItemDto { Id = Guid.NewGuid(), Name = "Routed Movie" }, + PositionTicks = 0 + }); + + Assert.Equal("Routed Movie", session.NowPlayingItem?.Name); + + var listedByA = await replicaA.GetSessions(_user.Id, null, null, null, false, cancellationToken); + var listedByB = await replicaB.GetSessions(_user.Id, null, null, null, false, cancellationToken); + + Assert.Equal("Routed Movie", Single(listedByA, session.Id).NowPlayingItem?.Name); + Assert.Equal("Routed Movie", Single(listedByB, session.Id).NowPlayingItem?.Name); + + await replicaB.OnPlaybackStopped(new PlaybackStopInfo { SessionId = session.Id, PositionTicks = 1 }); + + Assert.Null(session.NowPlayingItem); + Assert.Null(Single(await replicaB.GetSessions(_user.Id, null, null, null, false, cancellationToken), session.Id).NowPlayingItem); + } + + /// + /// A report that cannot be handed to the owner is still applied here, so an unreachable owner is + /// never worse than the single-instance behaviour of keeping the state on the replica that served + /// the request. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task PlaybackReportedWithTheOwnerUnreachable_IsAppliedLocally() + { + var cancellationToken = TestContext.Current.CancellationToken; + await using var replicaA = CreateReplica("pod-a"); + await using var replicaB = CreateReplica("pod-b"); + + var session = await Request(replicaA, "device-orphaned"); + session.AddController(new RecordingSessionController()); + await replicaA.OnSessionControllerConnected(session); + + var local = await Request(replicaB, "device-orphaned"); + + var entry = await _directory.GetAsync(session.Id, cancellationToken); + Assert.NotNull(entry); + entry.OwnerPod = "pod-gone"; + Assert.True(await _directory.PublishAsync(entry, long.MaxValue, cancellationToken)); + + await replicaB.OnPlaybackStart(new PlaybackStartInfo + { + SessionId = session.Id, + Item = new BaseItemDto { Id = Guid.NewGuid(), Name = "Orphaned Movie" }, + PositionTicks = 0 + }); + + Assert.Equal("Orphaned Movie", local.NowPlayingItem?.Name); + } + + /// + /// A directory that cannot be read says nothing about where a session is. Treating the failure as + /// "no such entry" hands the command to a local copy with no connection, which reports success and + /// delivers nothing. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task DirectoryReadFailingDuringRemoteControl_DoesNotSilentlyDoNothing() + { + var cancellationToken = TestContext.Current.CancellationToken; + var failing = new FailableSessionDirectory(new RedisSessionDirectory( + _connection, + Options.Create(new SessionDirectoryOptions()), + NullLogger.Instance)); + + await using var replicaA = CreateReplica("pod-a"); + await using var replicaB = CreateReplica("pod-b", failing); + + var session = await Request(replicaA, "device-unreadable"); + session.AddController(new RecordingSessionController()); + await replicaA.OnSessionControllerConnected(session); + + // The replica serving the request holds a copy of the session, and only a copy. + await Request(replicaB, "device-unreadable"); + + failing.FailReads = true; + + await Assert.ThrowsAsync( + () => replicaB.SendMessageCommand( + string.Empty, + session.Id, + new MessageCommand { Header = "Header", Text = "Dinner is ready" }, + cancellationToken)); + } + + /// + /// Ownership is decided by comparing the two replicas' connection epochs, so the epochs cannot come + /// from the replicas' own clocks: a lagging clock would keep a genuinely newer connection from ever + /// taking the session. They are handed out per session by the shared store instead. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task ConnectionEpochs_AreHandedOutByTheStore() + { + var cancellationToken = TestContext.Current.CancellationToken; + await using var replicaA = CreateReplica("pod-a"); + + var session = await Request(replicaA, "device-epoch"); + session.AddController(new RecordingSessionController()); + await replicaA.OnSessionControllerConnected(session); + + var recorded = await ReadOwnerEpoch(session.Id, cancellationToken); + var allocated = await _directory.AllocateConnectionEpochAsync(session.Id, cancellationToken); + + // A counter the store owns, not a reading of any replica's clock. + Assert.Equal(1, recorded); + Assert.Equal(2, allocated); + Assert.Equal(1, await _directory.AllocateConnectionEpochAsync(session.Id + "-other", cancellationToken)); + } + + /// + /// SyncPlay groups are still instance-local, so the replica serving the request has to notice that + /// its copy of the session has no connection. Holding a copy is not holding the connection, and a + /// command handed to a copy would be dropped without a word. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task SyncPlayCommandOnTheReplicaWithoutTheConnection_IsSkippedAndLogged() + { + var cancellationToken = TestContext.Current.CancellationToken; + var logger = new CapturingLogger(); + await using var replicaA = CreateReplica("pod-a"); + await using var replicaB = CreateReplica("pod-b", logger: logger); + + var session = await Request(replicaA, "device-syncplay"); + var controller = new RecordingSessionController(); + session.AddController(controller); + await replicaA.OnSessionControllerConnected(session); + + await Request(replicaB, "device-syncplay"); + + await replicaB.SendSyncPlayCommand( + session.Id, + new SendCommand(Guid.NewGuid(), Guid.NewGuid(), DateTime.UtcNow, SendCommandType.Pause, 0, DateTime.UtcNow), + cancellationToken); + + Assert.Contains(logger.Messages, i => i.Contains("SyncPlay command", StringComparison.Ordinal) && i.Contains(session.Id, StringComparison.Ordinal)); + Assert.False(controller.HasMessage); + } + /// /// 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. @@ -325,6 +613,16 @@ public sealed class SessionDirectoryReplicaTests : IAsyncLifetime Assert.DoesNotContain(listedByB, i => string.Equals(i.Id, session.Id, StringComparison.Ordinal)); } + private async Task ReadOwnerEpoch(string sessionId, CancellationToken cancellationToken) + { + var raw = await _connection.GetDatabase().StringGetAsync("jellyfin:sessionowner:" + sessionId).WaitAsync(cancellationToken); + + return long.Parse(raw.ToString().Split('|')[0], CultureInfo.InvariantCulture); + } + + private static SessionInfoDto Single(IReadOnlyList sessions, string sessionId) + => Assert.Single(sessions, i => string.Equals(i.Id, sessionId, StringComparison.Ordinal)); + private static async Task WaitUntil(Func condition, CancellationToken cancellationToken) { var deadline = DateTime.UtcNow.AddSeconds(10); @@ -350,7 +648,7 @@ public sealed class SessionDirectoryReplicaTests : IAsyncLifetime private Task Request(SessionManager replica, string deviceId) => replica.LogSessionActivity(AppName, AppVersion, deviceId, DeviceName, RemoteEndPoint, _user); - private SessionManager CreateReplica(string podId, SessionDirectoryOptions? options = null) + private SessionManager CreateReplica(string podId, SessionDirectoryOptions? options = null, ILogger? logger = null) { options ??= new SessionDirectoryOptions { EntryTtlSeconds = 60, RefreshIntervalSeconds = 3600 }; @@ -359,13 +657,16 @@ public sealed class SessionDirectoryReplicaTests : IAsyncLifetime Options.Create(options), NullLogger.Instance); - return CreateReplica(podId, options, directory, CreateBus(podId)); + return CreateReplica(podId, options, directory, CreateBus(podId, options), logger); } - private SessionManager CreateReplica(string podId, ISessionDirectory directory, IPodMessageBus bus) - => CreateReplica(podId, new SessionDirectoryOptions(), directory, bus); + private SessionManager CreateReplica(string podId, ISessionDirectory directory) + => CreateReplica(podId, new SessionDirectoryOptions(), directory, CreateBus(podId, new SessionDirectoryOptions()), null); - private SessionManager CreateReplica(string podId, SessionDirectoryOptions options, ISessionDirectory directory, IPodMessageBus bus) + private SessionManager CreateReplica(string podId, ISessionDirectory directory, IPodMessageBus bus) + => CreateReplica(podId, new SessionDirectoryOptions(), directory, bus, null); + + private SessionManager CreateReplica(string podId, SessionDirectoryOptions options, ISessionDirectory directory, IPodMessageBus bus, ILogger? logger) { var userManager = new Mock(); userManager.Setup(i => i.GetUserById(_user.Id)).Returns(_user); @@ -374,11 +675,14 @@ public sealed class SessionDirectoryReplicaTests : IAsyncLifetime var appHost = new Mock(); appHost.SetupGet(i => i.SystemId).Returns("server-" + podId); + var configurationManager = new Mock(); + configurationManager.SetupGet(i => i.Configuration).Returns(new ServerConfiguration()); + return new SessionManager( - NullLogger.Instance, + logger ?? NullLogger.Instance, Mock.Of(), Mock.Of(), - Mock.Of(), + configurationManager.Object, Mock.Of(), userManager.Object, Mock.Of(), @@ -393,22 +697,12 @@ public sealed class SessionDirectoryReplicaTests : IAsyncLifetime 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 IPodMessageBus CreateBus(string podId, SessionDirectoryOptions options) + => new RedisPodMessageBus( + _connection, + Options.Create(options), + podId, + NullLogger.Instance); /// /// Hands every replica its own context over the one shared database, the way the pooled factory does @@ -426,6 +720,60 @@ public sealed class SessionDirectoryReplicaTests : IAsyncLifetime public JellyfinDbContext CreateDbContext() => CreateContext(_dataSource); } + /// + /// Keeps what a replica logged so a skipped route can be told apart from a silent drop. + /// + /// The category the logger belongs to. + private sealed class CapturingLogger : ILogger + { + private readonly ConcurrentQueue _messages = new(); + + public IEnumerable Messages => _messages; + + public IDisposable BeginScope(TState state) + where TState : notnull + => NullLogger.Instance.BeginScope(state); + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func formatter) + => _messages.Enqueue(formatter(state, exception)); + } + + /// + /// A directory whose reads can be made to fail the way an unreachable valkey does. + /// + private sealed class FailableSessionDirectory : ISessionDirectory + { + private readonly ISessionDirectory _inner; + + public FailableSessionDirectory(ISessionDirectory inner) + { + _inner = inner; + } + + public bool FailReads { get; set; } + + public Task AllocateConnectionEpochAsync(string sessionId, CancellationToken cancellationToken = default) + => _inner.AllocateConnectionEpochAsync(sessionId, cancellationToken); + + public Task PublishAsync(SessionDirectoryEntry entry, long connectionEpoch, CancellationToken cancellationToken = default) + => _inner.PublishAsync(entry, connectionEpoch, cancellationToken); + + public Task RemoveAsync(string sessionId, string ownerPod, CancellationToken cancellationToken = default) + => _inner.RemoveAsync(sessionId, ownerPod, cancellationToken); + + public Task GetAsync(string sessionId, CancellationToken cancellationToken = default) + => FailReads + ? throw new RedisTimeoutException("The session directory is unreachable.", CommandStatus.Unknown) + : _inner.GetAsync(sessionId, cancellationToken); + + public Task> GetAllAsync(CancellationToken cancellationToken = default) + => FailReads + ? throw new RedisTimeoutException("The session directory is unreachable.", CommandStatus.Unknown) + : _inner.GetAllAsync(cancellationToken); + } + /// /// Stands in for the websocket the owning replica holds. /// @@ -433,10 +781,12 @@ public sealed class SessionDirectoryReplicaTests : IAsyncLifetime { private readonly TaskCompletionSource<(SessionMessageType MessageType, string Data)> _received = new(); - public bool IsSessionActive => true; + public bool IsSessionActive { get; set; } = true; public bool SupportsMediaControl => true; + public bool HasMessage => _received.Task.IsCompleted; + public Task SendMessage(SessionMessageType name, Guid messageId, T data, CancellationToken cancellationToken) { _received.TrySetResult((name, JsonSerializer.Serialize(data)));